Contract-Checked Event Buses: Making Pub/Sub Safe Between Micro-Features

At small scale, an internal event bus feels magical. One team adds "user.session.started," another listens and fires analytics, another triggers notifications. No coordination meeting, no new REST endpoints, just vibes.

Fast-forward a year: you've got dozens of topics, half a dozen "user.*" events, multiple schemas for "pet.updated," a consumer that silently drops messages because a field changed from int to str, and no one's sure which features will break if you touch that one producer.

This post is about turning that chaos into a contract-first event bus. We'll walk through how to model events as typed, versioned schemas using Pydantic or dataclasses, share explicit contracts between producers and consumers, use golden tests in CI to catch breaking changes before they ship, and make event evolution boring and predictable instead of spooky. By the end, you should have a mental blueprint for making your pub/sub layer evolve safely, even as micro-features multiply.

The quiet failure modes of pub/sub between micro-features

Most teams don't decide to build a fragile event bus, it emerges slowly from "just one more topic" decisions. Shape drift happens when a producer adds a field, changes a type, or stops sending something "optional," a consumer written six months ago assumes the old shape and quietly fails or miscomputes. Ghost consumers appear when nobody remembers that an old job still listens to a given topic, you deprecate it and only notice when a downstream ML pipeline stops updating. Implicit contracts mean the real definition of an event lives across three places, the producer code, a wiki page from last year, and a consumer-specific mapping function. Production-only surprises happen when locally your tests still pass, in staging nobody emits the weird edge-case event, and in production a slightly malformed payload from one path crashes a consumer you forgot to harden.

All of this is especially painful in micro-feature architectures, a single Python backend hosting many small, event-driven modules (alerts, AI insights, device analytics, billing nudges) communicating via an internal bus (Redis, SNS/SQS, Kafka, or even in-process queues). The root problem isn't the transport, it's that events are treated as strings and JSON blobs, not as contracts.

What we mean by a contract-checked event bus

A contract-checked event bus doesn't require exotic infrastructure. It's a discipline and a set of tools layered on top of whatever you already use, SQS, Kafka, Redis, SNS, NATS, or an internal in-memory bus.

Each event has a type and version, you don't emit "user.updated," you emit something like UserProfileUpdated.v1 with a known schema. Event schemas are code, not tribal knowledge, defined as Pydantic models or dataclasses in a central package imported by both producers and consumers. Golden tests validate the contract, for each event type you maintain sample payloads representing valid shapes and key edge cases, producers must still emit them and consumers must still accept them. CI enforces contracts, not just style, if you modify a schema in a breaking way the golden tests fail and your PR doesn't merge. Observability tells you who's using what, emissions and consumptions get logged with type, version, and feature tag, so you can map dependencies and deprecate safely.

This approach keeps your pub/sub from becoming a side-channel for surprises. It turns events into first-class APIs, with the same level of rigor you'd apply to REST or gRPC.

Modeling events as typed, versioned schemas

Assume you have a simple event bus wrapper around SQS, Kafka, or Redis. Instead of sending raw JSON dicts, define event classes:

from enum import Enum
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Literal
 
 
class EventType(str, Enum):
    PET_WEIGHT_UPDATED_V1 = "pet.weight.updated.v1"
 
 
class PetWeightUpdatedV1(BaseModel):
    event_type: Literal[EventType.PET_WEIGHT_UPDATED_V1] = EventType.PET_WEIGHT_UPDATED_V1
    pet_id: str = Field(..., description="Internal pet identifier")
    device_id: str = Field(..., description="Source device or bowl")
    weight_grams: int = Field(..., ge=0)
    measured_at: datetime
    source: str = Field(..., description="sensor|app|backfill")

The event_type is part of the payload, avoiding mixed-topic confusion and making debugging easier in logs and traces. Fields are typed and documented, Pydantic gives validation for free, your producer can't emit malformed events without failing fast. And the version is baked into the type name, so it's obvious which schema you're using.

Your event bus wrapper can then enforce these contracts:

import json
from typing import Type, TypeVar
 
T = TypeVar("T", bound=BaseModel)
 
def publish(event: BaseModel, topic: str) -> None:
    payload = event.model_dump(mode="json")
    # send to Kafka/SQS/Redis, etc.
    broker_client.publish(topic, json.dumps(payload).encode("utf-8"))
 
 
def parse_event(payload: bytes, model: Type[T]) -> T:
    data = json.loads(payload.decode("utf-8"))
    return model.model_validate(data)

Now both producers and consumers live in the same type universe, instead of each doing ad-hoc JSON parsing.

Evolving events without breaking everyone

Once events are typed, the next challenge is evolution. Requirements change, schemas must too. The trick is standardizing how you change them. Stay on the same version, say v1, when you add a nullable or optional field with a reasonable default, or widen a type in a backwards-compatible way, like an enum adding new values or a string length increasing:

class PetWeightUpdatedV1(BaseModel):
    # ... existing fields ...
    location_hint: str | None = Field(
        default=None,
        description="Optional text like 'kitchen bowl' or 'travel bowl'"
    )

Golden tests ensure existing goldens still validate, and new goldens can include the optional field.

Create a new version when you change field meaning (like weight_grams starting to include the bowl weight), change types in incompatible ways (int to str, nested structure rewritten), or drop fields consumers may still depend on:

class PetWeightUpdatedV2(BaseModel):
    event_type: Literal[EventType.PET_WEIGHT_UPDATED_V2] = EventType.PET_WEIGHT_UPDATED_V2
    pet_id: str
    device_id: str
    net_weight_grams: int
    tare_weight_grams: int
    measured_at: datetime
    source: str

Start emitting both v1 and v2 for a while, dual-write, update consumers to support v2, and deprecate v1 once usage is low enough, observability helps here.

Deprecation shouldn't be a Slack message, treat it as a mini API lifecycle: mark the version as deprecated in code and docs, add a planned removal date and owner, alert on remaining v1 traffic after a time window, and remove producer support only when no consumers rely on it. If this feels like a lot for "internal events," that's the point, they're just as critical as external APIs when micro-features depend on them.

Golden tests: turning contracts into CI enforcers

Typed schemas are great, but they don't stop someone from harmlessly rewriting an event in a way that passes type checks but breaks semantics. For each event type you maintain a set of canonical example payloads representing typical cases, edge cases (missing optional fields, enum edge values), and old variants you still support. A test ensures producers can still generate payloads matching the contracts, and consumers can still parse and handle them without raising or misbehaving.

An example layout:

events/
  contracts/
    pet_weight_updated_v1.py
    pet_weight_updated_v2.py
  goldens/
    pet_weight_updated_v1/
      typical.json
      missing_location_hint.json
    pet_weight_updated_v2/
      dual_bowl_setup.json
tests/
  test_golden_events.py

And then a test like:

import json
import pytest
from events.contracts import PetWeightUpdatedV1, PetWeightUpdatedV2
from pathlib import Path
 
GOLDENS_DIR = Path(__file__).parent.parent / "events" / "goldens"
 
@pytest.mark.parametrize("event_cls, golden_dir", [
    (PetWeightUpdatedV1, "pet_weight_updated_v1"),
    (PetWeightUpdatedV2, "pet_weight_updated_v2"),
])
def test_goldens_still_parse(event_cls, golden_dir):
    for golden_path in (GOLDENS_DIR / golden_dir).glob("*.json"):
        raw = json.loads(golden_path.read_text())
        event = event_cls.model_validate(raw)
        # optionally: assert derived invariants
        assert event.pet_id
        assert event.measured_at

Any incompatible change to the schema or consumer logic causes this test to fail before you merge. You can go further and exercise actual consumer handlers, validating not just shape but behavior against known, curated examples.

Observability: who emits what, who consumes it, and when it breaks

Contracts and tests protect you at deploy time. Observability protects you at runtime. For a contract-checked event bus, log and trace event type and version, the producing feature, the consuming feature, the outcome (processed, retried, discarded, dead-lettered), and the latency from publish to consume.

Even with a simple transport, a thin shared wrapper around publish/consume lets you emit structured logs:

def publish(event: BaseModel, topic: str, feature: str) -> None:
    payload = event.model_dump(mode="json")
    logger.info(
        "event_published",
        extra={
            "event_type": payload["event_type"],
            "version": payload["event_type"].rsplit(".v", 1)[-1],
            "topic": topic,
            "feature": feature,
        },
    )
    broker_client.publish(topic, json.dumps(payload).encode("utf-8"))

With this data, you can build a dependency map, "which features consume v1 versus v2," alert when a deprecated version is still active after a target date, and investigate incidents quickly, "which consumer is dropping events on this topic." You don't need a huge observability stack to start, structured logs filterable by event_type and consumer already take you far.

At Hoomanely, our mission is keeping pets healthier for longer by connecting smart devices, real-world behavior, and AI insights into one coherent experience for pet parents. That means a lot of micro-features talking to each other, a device session from a smart tracker or bowl emits weight samples, motion summaries, or thermal flags, AI modules produce nutrition nudges or anomaly alerts, analytics features aggregate and score behavior over days and weeks, and notification and coaching modules decide when to nudge the pet parent.

Instead of wiring these with tightly coupled APIs, we lean heavily on internal events, but that only works if those events are safe to evolve. Device and inference services publish contract-checked events like PetSessionSummary.v1 or PetWeightTrend.v2. Analytics, insights, and notification features consume them via a shared event contracts package. Golden tests in CI ensure that when we adjust a schema, like adding a "possible dehydration" flag, we don't accidentally break an older consumer. Observability ties everything together so we can see, for a given event type, which internal features rely on it. The result is we can keep adding new wellness features and AI insights without turning the system into a fragile web of implicit dependencies, crucial when shipping firmware, apps, and AI together.

Rolling this out without stopping the world

If you already have a messy event bus, you don't need a big-bang rewrite. Pick one event family to harden, something central but manageable. Define canonical schemas reflecting the current real-world shapes you see in the wild, not an idealized version. Backfill goldens from production samples, capturing sanitized actual payloads. Add a thin wrapper around publish/consume that validates event models, attaches type/version, and logs structured metadata. Migrate producers gradually, one at a time, keeping consumers tolerant where necessary. Introduce versioning rules documenting your evolution policy. Turn CI up slowly, starting with non-blocking warnings before making tests blocking. And socialize the model, treating the event contracts package as a first-class API surface with reviews and owners.

After a few cycles, engineers stop thinking "just emit JSON" and start thinking "which event contract and version should I use." That mindset shift is the real win.

Key takeaways

Pub/sub is an API, not a side-channel, if micro-features depend on events, treat those events with the same rigor as any public API. Typed, versioned schemas are the foundation, Pydantic or dataclasses give you validation, documentation, and a place to reason about changes. Golden tests turn contracts into enforcement, sample payloads plus CI ensure producers and consumers stay compatible as code evolves. Observability gives you a map, not a guess, with structured logs and simple maps you can see who emits and consumes what, and deprecate safely. And rollout can be incremental, start with a single event family, wrap publishes and consumes, and expand from there. For teams growing a constellation of micro-features, whether in pet-tech or any other domain, contract-checked event buses are one of the cleanest ways to keep flexibility and safety as you scale.