eventcommon 0.1.1

Shared event envelope, HMAC signing, and SNS/SQS outbox library for platform services.

eventcommon

Shared event-driven messaging library for platform services. Consumed as a private Python package — not deployed on its own.

Features: typed event envelopes (CloudEvents-aligned) · async SNS publisher · async SQS consumer · transactional outbox (Postgres-backed via pgcommon) · HMAC signing/verification · Prometheus metrics · OpenTelemetry tracing · RLS tenant propagation.

See ARCHITECTURE.md for design decisions and layer diagrams.

Cross-language contract: eventcommon's Envelope wire format and HMAC canonicalisation must stay byte-for-byte compatible with the Go implementation in platform-events — the two publish to and consume from the same SNS topics/SQS queues in a polyglot deployment. Any change to envelope.py or hmac.py must be cross-checked against the Go side; the interop CI job exists specifically to catch a mismatch before it reaches production. See CI.

Design principles

• Events are facts, not commands. An event records something that already happened (user.created, invoice.settled) — it is an immutable statement of past truth, not an instruction to perform an action. Producers do not direct consumers; consumers decide independently how to react. A consequence: consumers must be resilient to receiving the same fact more than once (Envelope.id idempotency) and must handle the absence of a reaction gracefully (a consumer can be added or removed without any producer change).
• Fail fast on misconfiguration — SNSPublisher raises on an empty or invalid topic_arn; SQSConsumer raises on an empty queue_url. Invalid configuration is rejected at construction time, not at first use.
• Make safe usage the default — the SQS consumer automatically propagates tenant_id into pgcommon's GUC context for RLS, extends visibility timeouts on slow handlers, and drains in-flight messages on stop(). You cannot forget these by accident when the defaults are wired correctly.
• Keep business logic free from messaging plumbing — handlers receive a typed Envelope; retry backoff, visibility extension, metric recording, and OTel span creation are invisible to the caller.
• Centralise cross-cutting concerns — HMAC signing, tenant propagation, and observability hooks live here, not scattered across every service. A single version bump propagates to all consumers.

---

Contents

• TL;DR
• System invariants
• Publishing rules
• Anti-patterns
• When NOT to use this
• Installing
• Quick start
• Event envelope
• SNS Publisher
• SQS Consumer
  • Implementing idempotency
• Transactional outbox
• HMAC helpers
  • When to use HMAC
• Observability — Prometheus and OTel
  • Logging correlation
• Backpressure considerations
• Recommended production defaults
• Configuration reference
• Error reference
• Testing in consuming services
• Testing
• CI
• Versioning and releases
• Service adoption checklist
• See also

---

TL;DR

⚠️ Exactly-once delivery is not possible anywhere in this system. SNS and SQS provide at-least-once delivery — a message may be delivered more than once under normal operating conditions (visibility timeout expiry, consumer crash mid-handler, SQS internal redelivery). FIFO topics reduce the duplicate window to 5 minutes within a MessageGroupId but do not eliminate duplicates end-to-end. Consumer idempotency is not optional — it is required for correctness regardless of whether FIFO is used. See Implementing idempotency for the concrete pattern.

• The system provides at-least-once delivery — consumers must implement idempotency using Envelope.id. See Implementing idempotency for the recommended processed_events Postgres pattern.
• Use eventcommon.envelope.new_envelope(...) to create typed events — never construct Envelope directly.
• Always pass tenant_id=rc.tenant_id and trace_id=rc.trace_id when publishing from an HTTP handler — these fields drive RLS enforcement and trace continuity.
• Use await eventcommon.outbox.enqueue(conn, env) inside a pgcommon.run_in_tx block to guarantee reliable at-least-once delivery without dual-write risk. See Publishing rules for the full decision table and crash-window explanation.
• Never call publisher.publish for domain events tied to a DB write — event loss on process crash is silent and unrecoverable. The outbox is the only safe path.
• Call eventcommon.metrics.init(service_name, build_version) once at startup to enable Prometheus metrics.
• OTel tracing is enabled by the consuming service (e.g. fastapicommon.tracing.init_from_env()), not by this library.

---

System invariants

These hold everywhere in the system, always. If your design requires violating any of them, the design needs to change — not the invariant.

Invariant	What it means for you
Delivery is at-least-once	Every consumer handler may be called more than once for the same Envelope.id. Idempotency is not a nice-to-have.
Ordering is best-effort unless FIFO per group	Standard SNS/SQS make no ordering promise. FIFO guarantees order only within a single MessageGroupId. Cross-group and cross-service ordering is never guaranteed.
Idempotency is required for all consumers	A handler that is not idempotent is not correct. There is no configuration, queue type, or delivery mode that removes this requirement.
Events are immutable once published	An event_type string and its payload contract are frozen on first production publish. Breaking changes require a new versioned type (.v2).
The producer has no knowledge of consumers	A service must never check "who is listening" before publishing. Consumers come and go; the event type remains.
Tenant context is always explicit	Every event scoped to a tenant carries tenant_id. Using SYSTEM_TENANT_ID for a tenant-scoped event is a correctness bug, not a shortcut.
Delivery timing is unbounded	Consumers must not rely on when an event arrives — only on what it says. A handler that breaks when the event arrives "late" is not correct.
Delivery latency is not SLA-backed	Event delivery is best-effort within the outbox poll interval plus SNS/SQS propagation time. There is no guaranteed maximum latency.
Idempotency must cover all externally observable side effects	A DB write, an external HTTP call, a notification, and a cache invalidation are all externally observable. Idempotency that protects the DB write but not the Stripe charge or the email is incomplete. See Side-effect classification.

---

Publishing rules

Violating the outbox rule introduces silent data-loss bugs. A domain event is permanently lost if the process crashes between the DB commit and the SNS call. There is no retry, no error, no log — the event is simply gone.

Decision table

Scenario	Correct method	Delivery guarantee
Domain event tied to a DB write — user created, invoice settled, order placed, state machine transition	outbox.enqueue inside pgcommon.run_in_tx	At-least-once ✅
Background job publishing across tenants — scheduled reconciliation, batch processing	outbox.enqueue with tenant_id=SYSTEM_TENANT_ID inside pgcommon.run_in_tx	At-least-once ✅
External notification not tied to a DB write — webhook ping after a read-only operation	publisher.publish	At-most-once ⚠️
Fire-and-forget telemetry or audit where loss is explicitly acceptable	publisher.publish	At-most-once ⚠️
Service with no PostgreSQL database (document in your service ADR)	publisher.publish	At-most-once ⚠️

The rule in one sentence: if a consumer's state would be wrong or incomplete if it never received this event, use the outbox.

Why the crash window matters

Without the outbox, there is an unrecoverable gap between the DB commit and the SNS publish:

WITHOUT OUTBOX:
  1. BEGIN
  2. INSERT INTO users ...     ← domain write
  3. COMMIT                    ← process crashes here (OOM, deploy, network drop)
  4. await sns.publish(...)    ← never executes — event lost permanently

With the outbox, the event is durable the moment the transaction commits:

WITH OUTBOX:
  1. BEGIN
  2. INSERT INTO users ...          ← domain write
  3. INSERT INTO outbox_events ...  ← event write in same transaction
  4. COMMIT                         ← both rows durable; crash here is safe
  5. [outbox runner, async]
       await publisher.publish(...) ← retried up to max_attempts on failure
       published_at = NOW()         ← idempotent completion marker

If the runner crashes between steps 4 and 5, the row is still in outbox_events with published_at IS NULL. The next runner instance (or the restarted process) picks it up on the next poll cycle.

Detecting misuse in code review

Flag any call to publisher.publish or publisher.publish_batch that appears inside a handler that also writes to the database. The correct pattern is:

# ✅ Correct — both writes in one transaction
async with pgcommon.run_in_tx(pool) as conn:
    await repo.save_user(conn, user)
    await outbox.enqueue(conn, env)

# ❌ Wrong — SNS call outside the transaction; event lost on crash
async with pgcommon.run_in_tx(pool) as conn:
    await repo.save_user(conn, user)
await publisher.publish(env)  # ← data inconsistency risk

---

Anti-patterns

Common mistakes that cause silent failures, data loss, or broken tenant isolation.

Anti-pattern	Why it's wrong	What to do instead
await publisher.publish(env) directly inside an HTTP handler for a transactional event	If the process dies after the DB commit but before publish returns, the event is silently dropped — no retry, no recovery	Use outbox.enqueue inside pgcommon.run_in_tx alongside the domain write
Ignoring Envelope.id in the consumer handler	The outbox runner delivers at-least-once; without an idempotency check, a redelivered event causes duplicate side effects	INSERT INTO processed_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING inside the same transaction as the side-effect write
Branching on env.data without checking env.type first	A handler subscribed to multiple event types will silently misparse the wrong payload	Always check env.type before parsing env.data; route to typed handlers by event type
FIFO topic with a non-stable MessageGroupId (e.g. random UUID per message)	Defeats ordering — every message lands in its own group; SNS treats them as independent and delivers concurrently	Use a deterministic, stable group ID: tenant_id, user_id, or aggregate_id — a value that must be ordered relative to itself
tenant_id=SYSTEM_TENANT_ID in an HTTP handler context	Bypasses per-tenant RLS on the consumer side; the handler's DB queries run without a tenant GUC, returning wrong row sets silently	Pass tenant_id=rc.tenant_id from the request context; only use SYSTEM_TENANT_ID for genuine cross-tenant background jobs
Parsing event payloads with a schema that rejects unknown fields (extra="forbid" in Pydantic, Model.Config.strict)	Turns every non-breaking producer schema addition into a consumer runtime error	Use permissive parsing (extra="ignore") on event payloads
Calling outbox.enqueue outside a transaction (bare connection, no active transaction)	The event is not durably linked to the domain write; partial failures can produce an event with no corresponding domain record	Always call outbox.enqueue inside pgcommon.run_in_tx; never call it on a connection with no open transaction

---

When NOT to use this

This library targets long-running platform services that publish or consume domain events. It may be more than you need if:

• One-off scripts or CLIs — a plain boto3/aioboto3 SNS call is simpler without the pool setup overhead.
• Services that only consume a single event type — if filtering, retries, and concurrency control are handled entirely by the SQS queue configuration, a thin wrapper is sufficient.
• Non-AWS message brokers — the library is SNS/SQS-specific. EventBridge, Kafka, and other backends are not supported without a new adapter.
• Services that do not use PostgreSQL — pgcommon is a hard dependency of this library regardless (the SQS consumer injects tenant GUC context on every message, not just the outbox — see ARCHITECTURE.md's Layer model). If your service has no database, you can still install and use the SNS publisher directly and accept at-most-once delivery; you simply never call outbox.enqueue/outbox.Runner.

---

Installing

eventcommon publishes to public PyPI (the GitHub repository itself stays private — see VERSIONING.md § Publishing to PyPI for why that split is deliberate):

uv add eventcommon

pip install eventcommon==0.1.0

Pin an exact version in production — see Versioning and releases for the pin-style guidance.

---

Quick start

import os

from eventcommon import metrics
from eventcommon.config import load_outbox, load_sns, load_sqs, log_warnings, runner_config_from_env
from eventcommon.consumer import SQSConsumer
from eventcommon.outbox import Runner, apply_schema
from eventcommon.publisher import SNSPublisher
from pgcommon import create_pool
from pgcommon.migrate import Runner as MigrateRunner


async def main() -> None:
    # 1. Register Prometheus metrics once.
    metrics.init(os.environ["APP_NAME"], os.environ["BUILD_VERSION"])

    # 2. Open the connection pool for the outbox runner.
    pool = await create_pool(dsn=os.environ["DATABASE_URL"])

    # 3. Apply the outbox schema migration.
    migrate_runner = MigrateRunner(dsn=os.environ["DATABASE_URL"])
    await apply_schema(migrate_runner)

    # 4. Construct the SNS publisher.
    sns_env = load_sns()
    publisher = SNSPublisher(sns_env.to_config(logger=logger))

    # 5. Start the outbox runner (delivers events asynchronously).
    outbox_env = load_outbox()
    log_warnings(outbox_env.warnings, logger=logger)
    runner = Runner(runner_config_from_env(outbox_env, pool, publisher, logger=logger))
    runner_task = asyncio.create_task(runner.start())

    # 6. Construct and start the SQS consumer.
    sqs_env = load_sqs()
    log_warnings(sqs_env.warnings, logger=logger)

    async def handle_event(env): ...  # handle event

    consumer = SQSConsumer(
        sqs_env.to_config(logger=logger), handle_event, **sqs_env.consumer_options()
    )
    consumer_task = asyncio.create_task(consumer.start())

    try:
        await asyncio.gather(runner_task, consumer_task)
    finally:
        await runner.stop()
        await consumer.stop()
        await pool.close()

---

Event envelope

Envelope[T] is the canonical wire format for all inter-service events — identical on the wire to the Go implementation's Envelope[T]. Terminology used consistently throughout this document:

Term	Meaning	Example
event	The logical domain occurrence and the Envelope that carries it	"publish an event", "the event's tenant_id"
message	The SNS/SQS infrastructure unit — what the broker delivers, retries, and deletes	"SQS message", "delete_message", "MessageGroupId"
outbox record	A row in outbox_events or outbox_dead_letters — the durable Postgres representation of an event pending publish	"claim outbox records", "failed outbox record"

{
  "id":             "01926e4f-...",
  "type":           "iam.user.created",
  "source":         "platform-iam",
  "specversion":    "1",
  "tenant_id":      "acme",
  "trace_id":       "4bf92f3577b34da6a3ce929d0e0e4736",
  "correlation_id": "...",
  "subject":        "users/01926e4f-...",
  "actor":          "[e-mail]",
  "ip_address":     "203.0.113.42",
  "user_agent":     "Mozilla/5.0 (compatible; XPert/1.0)",
  "dataschema":     "550e8400-e29b-41d4-a716-446655440000",
  "time":           "2026-05-27T12:00:00Z",
  "data":           { ... }
}

Creating envelopes

from dataclasses import dataclass

from eventcommon.envelope import new_envelope


@dataclass
class UserCreatedPayload:
    user_id: str
    email: str


# From an HTTP handler — always carry trace and tenant from the request context.
rc = request_context(request)  # raises/returns None for unauthenticated requests
env = new_envelope(
    "iam.user.created",
    "platform-iam",
    UserCreatedPayload(user_id=user.id, email=user.email),
    tenant_id=rc.tenant_id,
    trace_id=rc.trace_id,  # REQUIRED for cross-service trace continuity — see note below
    schema_version="1",
)

⚠️ Always pass trace_id=rc.trace_id from the request context. If trace_id is omitted:

• The OTel sqs.receive span on the consumer side has no parent — it starts a new disconnected trace, severing the link back to the originating HTTP request
• A production incident that begins as an HTTP 500 in service A and manifests as a data anomaly in service B becomes impossible to correlate in Tempo or Grafana without manual log searching
• The trace_id field in handler log lines will be empty, breaking the log-to-trace pivot that structured logging provides

For background jobs where there is no inbound HTTP trace, generate a new trace ID at job startup and pass it consistently across all events emitted in that job run — this at minimum groups all events from the same run together.

System-level events (background jobs, scheduled tasks, cross-tenant operations) that are not scoped to any single tenant should use SYSTEM_TENANT_ID instead:

from eventcommon.envelope import SYSTEM_TENANT_ID, new_envelope

# Background job — not associated with any specific tenant.
env = new_envelope(
    "billing.invoices.generated",
    "billing-worker",
    payload,
    tenant_id=SYSTEM_TENANT_ID,
)

⚠️ Use SYSTEM_TENANT_ID only when the event genuinely has no tenant scope — scheduled batch jobs, cross-tenant reconciliation, or platform-level operational events. Do not use it to avoid looking up a tenant ID, to simplify a code path, or because the tenant is "unknown at call time" (that last case means the upstream context is missing and should be fixed, not papered over). Misuse bypasses the tenant isolation that pgcommon's RLS policies enforce on the consumer side. A consumer handler that receives a system tenant ID will execute DB queries without a tenant GUC set, which causes RLS policies to either reject the query or return the wrong row set. The failure is silent at the event level and only surfaces as incorrect data in the consuming service.

source field contract: identifies the producing service. Rules:

Rule	Detail
Stable and globally unique	Use the canonical service name: platform-iam, billing-service, inventory-worker. It must be unique across all services in the organisation
Immutable across deployments	Do not derive it from hostname, pod name, or any runtime variable — it must be the same value in every environment
Environment-free	Never append -dev, -staging, -prod, or a region suffix. The environment is implicit in which AWS account/topic the event lands on
Set as a constant	Define it once as a module-level constant in the service (SERVICE_NAME = "platform-iam") and pass it to new_envelope and metrics.init from that single source of truth

event_type convention: <domain>.<entity>.<past-tense-verb>[.v<N>] — e.g. iam.user.created, billing.invoice.settled. The .v<N> segment is only added for breaking payload changes (v1 is implicit). See EVENT_SCHEMA_GOVERNANCE.md for the full naming rules, the event type registry, and payload evolution tiers.

Event types are immutable once published. Additive fields (optional, defaulted) are allowed in the same type. Removing, renaming, changing the type, or changing the meaning of an existing field requires a new event type (e.g. iam.user.created.v2).

Consumers must ignore unknown payload fields — but unlike Go, this is not automatic in Python. Always construct typed payloads via parse_envelope(data, payload_type=YourDataclass), never YourDataclass(**raw_dict) directly — the former filters unrecognised keys before construction, the latter raises TypeError the moment a producer adds a harmless new field. See EVENT_SCHEMA_GOVERNANCE.md § The Python unknown-field trap.

schema_version records the payload contract version in the envelope (wire field specversion, CloudEvents-aligned). Set "1" at inception. Increment only when both conditions are true: (1) a new optional field was added and (2) at least one consumer needs to branch logic based on whether that field is present. Consumers that receive an unrecognised specversion should log a warning and drop the message rather than silently misparsing it. If specversion is absent on the wire, treat it as "1" for backward compatibility.

Serialisation

# Marshal to JSON wire format.
data = env.to_json()

# Unmarshal and validate required fields (raises EnvelopeIDRequiredError etc. on a missing field).
env = parse_envelope(data, payload_type=UserCreatedPayload)

Payload typing

Envelope[T] is generic, but two concrete forms appear throughout the codebase with different purposes:

Form	When to use
Envelope[dict[str, Any]] (raw JSON)	Publishing (Publisher.publish accepts this form); handler dispatch; routing layers that inspect headers but not the payload; audit loggers; DLQ handlers that forward without parsing
Envelope[YourDataclass]	parse_envelope(data, payload_type=YourDataclass) when you own the event type and want the payload pre-parsed; mock injection in unit tests

The transport boundary always uses raw JSON. The Publisher and Handler types are both fixed to an untyped payload — this is intentional. It keeps the routing layer (SQS consumer, outbox runner) free of business-type imports.

Typed payload access happens inside the handler after the transport layer has delivered the envelope:

# ✅ Correct pattern — transport receives the raw envelope; handler parses to a typed dataclass
async def handle_user_created(env: Envelope[dict[str, Any]]) -> None:
    payload = UserCreatedPayload(**env.data)
    # payload is now typed; env.id / env.tenant_id / env.trace_id are always available directly
    await repo.create_user(payload)

Routing handlers that need to inspect the event type and forward without parsing should stay untyped throughout — no parse cost, no coupling to payload dataclasses:

# Fan-out router — dispatches by event type without ever touching the payload
async def route(env: Envelope[dict[str, Any]]) -> None:
    if env.type == "iam.user.created":
        await iam_consumer.publish(env)  # forward as-is
    elif env.type == "billing.invoice.settled":
        await billing_consumer.publish(env)
    # unknown type: discard

Envelope ID

Envelope.id is a UUID v7 (time-ordered, collation-friendly in Postgres B-tree indexes). Use it as an idempotency key on the consumer side. eventcommon._internal generates UUIDv7 itself for portability down to Python 3.12 (the stdlib uuid module only gained native uuid7() support in 3.14).

Envelope compatibility guarantees

The id, type, source, and time fields are stable — always present, never removed or renamed, format frozen within v0.x/v1.x. The remaining fields (tenant_id, trace_id, correlation_id, specversion, subject, actor, dataschema) are contextual — present when set, never removed. The library may add new optional fields in MINOR releases; existing consumers are unaffected. See ARCHITECTURE.md § Envelope compatibility guarantees for the full per-field stability class table.

---

SNS Publisher

from eventcommon.publisher import SNSConfig, SNSPublisher

publisher = SNSPublisher(
    SNSConfig(
        topic_arn=os.environ["SNS_TOPIC_ARN"],  # required — raises if empty or invalid ARN format
        region=os.environ["AWS_REGION"],
        endpoint_url=os.environ.get(
            "AWS_ENDPOINT_URL"
        ),  # set to http://localhost:4566 for LocalStack
        logger=logger,
    )
)

Publishing a single event

env = new_envelope(
    "iam.user.created",
    "platform-iam",
    payload,
    tenant_id=rc.tenant_id,
    trace_id=rc.trace_id,
)
await publisher.publish(env)

Batch publishing

# publish_batch splits automatically at the SNS hard limit of 10.
await publisher.publish_batch(envelopes)
# Partial failures raise a BatchError listing per-message errors.

⚠️ publish_batch is not atomic and must never be used for critical domain events. SNS batch publish is a best-effort call: some messages in the batch may succeed while others fail within the same call. A raised error does not mean the entire batch was rejected. For any event tied to a state change or a database write, use outbox.enqueue — partial batch failure combined with a process crash leaves no recovery path.

Callers must inspect the error to distinguish partial from total failure:

from eventcommon.errors import BatchError

try:
    await publisher.publish_batch(envelopes)
except BatchError as exc:
    # Partial failure — some messages published, some did not.
    # exc.failures is list[BatchFailure(index: int, error: Exception)].
    # Successfully published messages must NOT be retried.
    failed = [envelopes[f.index] for f in exc.failures]
    for f in exc.failures:
        logger.warning(
            "batch publish failure",
            index=f.index,
            event_id=envelopes[f.index].id,
            error=str(f.error),
        )
    # Retry failed messages individually or hand off to the outbox for durable retry.
    await publisher.publish_batch(failed)
except Exception as exc:
    # Non-BatchError: total failure (network error, auth failure, etc.) — safe to retry the full batch.
    raise RuntimeError(f"publish_batch: {exc}") from exc

Key rules:

Rule	Reason
Never retry the full batch on a BatchError	Messages that succeeded would be published a second time, creating duplicates
Always extract failed indices from exc.failures	The batch index is the only link between a failure and the original envelope
Prefer the outbox for durable retry	If the caller cannot afford to lose messages on process crash, use outbox.enqueue instead of publish_batch
publish_batch is appropriate for idempotent or at-most-once scenarios	Background notifications, cache invalidation signals, or any event where a missed or duplicate delivery is acceptable

FIFO topics

publisher = SNSPublisher(
    SNSConfig(
        topic_arn="arn:aws:sns:us-east-1:123456789012:my-topic.fifo",
        region="us-east-1",
        logger=logger,
    ),
    message_group_id=lambda env: env.tenant_id,  # group per tenant for ordered delivery
)

When topic_arn ends in .fifo, message_group_id is required; MessageDeduplicationId defaults to Envelope.id (requires content-based deduplication disabled at the topic level).

⚠️ FIFO does not mean exactly-once delivery end-to-end. This is the most common FIFO misconception:

Layer	What FIFO guarantees	What it does NOT guarantee
SNS	Deduplicates identical MessageDeduplicationId values within a 5-minute window	Deduplication outside that window; delivery to SQS is still at-least-once
SQS	Ordered delivery within a MessageGroupId; no duplicate delivery within a single consumer session	Protection against redelivery after a visibility timeout expires or a consumer crashes mid-handler
End-to-end	Ordered, deduplicated fan-out from SNS to SQS	That the consumer handler runs exactly once — it will not if the handler crashes after processing but before delete_message

FIFO gives you ordering. Idempotency still gives you safety. Use message_group_id to enforce processing order within a group (e.g. per-tenant, per-aggregate). Use Envelope.id + INSERT ... ON CONFLICT DO NOTHING to make the handler safe to run twice.

Message attributes

EventType, TenantID, Source, EventID, and Subject (when non-empty) are set as SNS message attributes. This enables SQS subscription filter policies that scope queues to specific event types, tenants, or resource subjects without deserialising the message body. Actor is an audit-trail field and is not forwarded as an SNS attribute.

---

SQS Consumer

from eventcommon.consumer import SQSConfig, SQSConsumer


async def handle_event(env: Envelope[dict[str, Any]]) -> None:
    # eventcommon has already propagated tenant_id into the pgcommon GUC context —
    # pool.with_conn()/run_in_tx() automatically enforce RLS for this tenant.
    await handle(env)


consumer = SQSConsumer(
    SQSConfig(
        queue_url=os.environ["SQS_QUEUE_URL"],  # required
        region=os.environ["AWS_REGION"],
        endpoint_url=os.environ.get("AWS_ENDPOINT_URL"),
        max_messages=10,  # 1–10; defaults to 10
        wait_seconds=20,  # long-poll; defaults to 20
        logger=logger,
    ),
    handle_event,
    concurrency=5,
    visibility_timeout=30,  # seconds
)

task = asyncio.create_task(consumer.start())  # runs until stop() is called
...
await consumer.stop()  # graceful drain — waits up to 30s for in-flight handlers

Handler contract

Handler outcome	Consumer behaviour
Returns normally	SQS message deleted from the queue
Raises an exception	SQS message left visible; retried after the visibility timeout
Parse failure (json.loads on the message body)	Message deleted immediately; counted as events_consumed_total{status=malformed}

visibility_timeout limit: SQS enforces a hard maximum of 12 hours. SQSConsumer.__init__ raises if visibility_timeout > 43200 seconds.

Task contract: each handler invocation runs in its own asyncio.Task, bounded by an asyncio.Semaphore(concurrency). On stop(), the consumer stops receiving new messages and awaits in-flight handler tasks for up to drain_timeout — handlers are not cancelled during this window, so a handler that never returns will hold up shutdown. Set your own per-handler timeout (asyncio.wait_for) rather than relying on the consumer to enforce one.

Error classification

The handler's outcome — return normally vs. raise — is the only signal the consumer uses to decide retry-or-delete. Getting this backwards determines whether a broken message sits in SQS forever or gets silently discarded.

Error class	Examples	Handler action	Outcome
Transient	DB connection timeout, downstream HTTP 503, network blip, lock contention	raise	Message left visible; SQS retries after visibility timeout; moves to DLQ after MaxReceiveCount
Permanent	Payload fails business validation, unknown event_type your handler cannot process, schema version too new to parse	Return normally + log at WARNING/ERROR	Message deleted immediately; no retry; no DLQ pressure
Programming error	AttributeError, KeyError, TypeError from a bug	raise (let it propagate)	Message retried; surfaced in events_consumed_total{status=error}; investigate immediately
Unknown / unexpected	Error from a dependency you haven't classified	raise	Retry by default; safe to escalate to DLQ if unresolved

⚠️ Misclassifying permanent errors as transient is the most common handler mistake. If a malformed payload causes the handler to raise, SQS retries it MaxReceiveCount times, then pushes it to the DLQ — where it will sit indefinitely. Return normally (and log) for any message that cannot succeed on retry regardless of how many times it is delivered.

Classification pattern:

async def handle_user_created(env: Envelope[dict[str, Any]]) -> None:
    try:
        payload = UserCreatedPayload(**env.data)
    except (TypeError, KeyError) as exc:
        # Permanent: malformed payload will never parse correctly on retry
        logger.error(
            "failed to parse UserCreatedPayload — discarding", event_id=env.id, error=str(exc)
        )
        return

    if not payload.user_id:
        # Permanent: business validation failure; retrying won't fix a missing user_id
        logger.warning("UserCreatedPayload missing user_id — discarding", event_id=env.id)
        return

    try:
        await repo.create_user(payload)
    except DatabaseError as exc:
        # Transient: DB errors may resolve; let SQS retry
        raise RuntimeError(f"create_user: {exc}") from exc

Poison messages

A poison message is structurally valid JSON — it parses without error — but cannot be processed successfully regardless of how many times it is retried. It is distinct from a transient error (which may succeed on retry) and from a malformed message (which fails JSON parsing).

Common causes:

Cause	Example
Invalid business state	user_id references a user that was deleted before the event arrived
Illegal state transition	An order.shipped event arrives for an order already in cancelled state
Missing precondition	A payment.settled event arrives but no corresponding payment.created exists
Schema version too new	specversion: "5" but this consumer only understands up to "3"

Handling pattern:

async def handle_order_shipped(env: Envelope[dict[str, Any]]) -> None:
    try:
        payload = OrderShippedPayload(**env.data)
    except (TypeError, KeyError):
        return  # malformed — discard (see Error classification)

    order = await repo.get_order(payload.order_id)
    if order is None:
        # Poison: the order doesn't exist and never will on retry.
        logger.error(
            "poison message: order not found",
            event_id=env.id,
            event_type=env.type,
            order_id=payload.order_id,
            trace_id=env.trace_id,
            tenant_id=env.tenant_id,
        )
        return  # drop — do NOT raise

    if order.status == "cancelled":
        # Poison: invalid state transition; retrying will never change the order status.
        logger.warning(
            "poison message: illegal state transition — order already cancelled",
            event_id=env.id,
            order_id=payload.order_id,
        )
        # Optional: forward to an audit or failure topic for later investigation.
        await audit_publisher.publish(
            new_envelope(
                "platform.event.poison",
                "order-consumer",
                env.data,
                tenant_id=env.tenant_id,
                correlation_id=env.id,
            )
        )
        return  # drop

    await repo.mark_shipped(order.id)

Decision rules:

1. Return normally, don't raise — raising retries the message; a semantically invalid message will never become valid on retry.
2. Log at ERROR or WARNING with the full correlation set (event_id, event_type, trace_id, tenant_id) — poison messages are silent data anomalies; the log is the only record that something was discarded.
3. Optionally forward to an audit or failure topic if the business impact is high enough to warrant investigation or replay tooling.
4. Do not forward sensitive payload data (PII, credentials) to an audit topic without confirming the destination has appropriate access controls and retention policies.

Implementing idempotency

SQS delivers messages at least once. A handler may be called more than once for the same Envelope.id due to visibility timeout expiry, a network error between handler completion and delete_message, or a consumer restart mid-batch. Use Envelope.id as the idempotency key.

Pattern 1 — Postgres unique constraint (recommended)

CREATE TABLE IF NOT EXISTS processed_events (
    event_id     TEXT        PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

async def handle_user_created(env: Envelope[dict[str, Any]]) -> None:
    async with pgcommon.run_in_tx(pool) as conn:
        row = await conn.fetchrow(
            "INSERT INTO processed_events (event_id) VALUES ($1) "
            "ON CONFLICT (event_id) DO NOTHING RETURNING event_id",
            env.id,
        )
        if row is None:
            return  # already processed by a previous delivery — safe no-op

        payload = UserCreatedPayload(**env.data)
        await repo.create_user(conn, payload)

Why inside the same transaction? If the side-effect write succeeds but the transaction rolls back before processed_events commits, the next delivery re-inserts the row and re-runs the side effect — correct behaviour. If processed_events commits but the side-effect write fails, the transaction rolls back atomically — both are absent, and the next delivery retries both.

Pattern 2 — Upsert-based idempotency

Use ON CONFLICT DO UPDATE with a sentinel column when the side effect is a row update rather than an insert — see ARCHITECTURE.md for the full SQL pattern.

Pattern 3 — Redis SETNX (for non-Postgres handlers)

async def handle_welcome_email(env: Envelope[dict[str, Any]]) -> None:
    key = f"processed:{env.id}"
    was_set = await redis.set(
        key, "1", ex=5 * 24 * 3600, nx=True
    )  # 5-day TTL matches SQS max retention
    if not was_set:
        return  # already processed
    await email_svc.send_welcome(env.data)

Caveat: Redis SETNX is not durable across Redis restarts without AOF persistence. For financial or critical events, Pattern 1 is safer.

Common mistakes

Mistake	Consequence	Fix
Check processed_events in a separate query before the transaction	TOCTOU race — two concurrent deliveries both pass the check and both execute	Always check inside the same transaction as the side-effect write
No idempotency at all	Duplicate charges, emails, or rows on any redelivery	Use Pattern 1 or 2
Idempotency outside the transaction	DB write succeeds, then processed_events insert fails → unguarded on next delivery	Idempotency claim and side-effect write must commit atomically
Never pruning processed_events	Table grows unboundedly	Nightly job: DELETE ... WHERE processed_at < NOW() - INTERVAL '5 days'

Side-effect classification

Side-effect class	Examples	Duplicate risk	Required guard
DB write	INSERT, UPDATE, state machine transition, balance change	High	Wrap in pgcommon.run_in_tx with processed_events ON CONFLICT DO NOTHING
External HTTP call (mutating)	Charge a payment, call a third-party API, trigger a webhook	High	Pass Envelope.id as the idempotency key in the downstream request
Email / SMS / push notification	Welcome email, OTP, shipping confirmation	Medium	Gate with processed_events before sending
Cache write / invalidation	Redis SET, CDN purge, in-memory state update	Low	No guard needed; if cache drives a downstream decision, guard the underlying DB write instead
Read-only / observability	Incrementing a counter, logging, emitting a metric	None	No guard needed

---

Transactional outbox

The outbox pattern eliminates dual-write risk: the event is written inside the business transaction alongside the domain mutation. If the transaction rolls back, the event is never published. The runner delivers asynchronously with at-least-once guarantee. Outbox does not guarantee global ordering — use FIFO topics with a stable message_group_id when ordering is required.

Wiring

from eventcommon.outbox import Runner, RunnerConfig, apply_schema
from pgcommon.migrate import Runner as MigrateRunner

# 1. Apply outbox schema migration (once at startup).
migrate_runner = MigrateRunner(dsn=os.environ["DATABASE_URL"])
await apply_schema(migrate_runner)

# 2. Construct the outbox runner.
runner = Runner(
    RunnerConfig(
        pool=pool,  # pgcommon.Pool — required
        publisher=publisher,  # eventcommon Publisher — required
        logger=logger,
        poll_interval=5.0,  # seconds
        batch_size=50,
        max_attempts=5,
        # Production-hardening options (all have safe defaults):
        publish_concurrency=1,  # default — SNS PublishBatch (10/API call); raise only when SNS latency-bound
        publish_timeout=10.0,  # per-record publish timeout
        drain_timeout=30.0,  # stop() waits up to this for the in-flight batch
        startup_jitter=0.2,  # random delay before first poll; desyncs replicas on rolling restart
        # To load from env: runner_config_from_env(load_outbox(), pool, publisher, logger=logger)
    )
)
runner_task = asyncio.create_task(runner.start())  # runs until stop() is called
...
if (
    err := await runner.stop()
):  # returns an error message if the in-flight batch did not drain in time
    logger.warning("outbox runner drain timeout — records retry after lease expiry", error=err)

# Optional: gate the Kubernetes readiness probe until the first poll succeeds.
await asyncio.wait_for(runner.ready.wait(), timeout=30)

Config defaults: poll_interval 5s · batch_size 50 · max_attempts 5 · publish_concurrency 1 · publish_timeout 10s · drain_timeout 30s · claim_lease_duration 10m · startup_jitter 0.

Enqueueing inside a transaction

async with pgcommon.run_in_tx(pool) as conn:
    await repo.save_user(conn, user)
    env = new_envelope(
        "iam.user.created",
        "platform-iam",
        {"user_id": user.id},
        tenant_id=rc.tenant_id,
        trace_id=rc.trace_id,
    )
    await outbox.enqueue(conn, env)

enqueue validates the envelope (non-empty id/type/source; non-zero time) and rejects payloads whose serialised size exceeds 240 KB — staying under the SNS 256 KB hard limit so an outbox record that could never publish is never persisted.

Payload size guidelines

Tier	Limit	Meaning
Hard limit	256 KB	SNS Publish API rejects messages above this — non-negotiable
Enforced limit	240 KB	outbox.enqueue rejects at this threshold
Recommended	≤ 64 KB	Comfortable budget for typical domain events

When the payload is inherently large (binary content, rendered templates, bulk export rows), store the data in object storage and put only a reference in the envelope — the consumer fetches the object after receiving the event.

Dead letters

"Dead letters" means two different things — do not confuse them. Outbox dead letters are publish failures stored in Postgres. SQS DLQ entries are consumer processing failures stored in a separate SQS queue. See ARCHITECTURE.md § Failure lifecycle for the complete side-by-side timeline and operational runbook.

Failed outbox records that exhaust max_attempts move to outbox_dead_letters — a queryable Postgres table.

# Step 1 — Inspect before acting.
records = await runner.list_dead_letters(DLQFilter(tenant_id="acme"), 50)

# Step 2a — Replay after fixing the root cause.
n = await runner.reprocess_dead_letters(100)  # unfiltered: replay all
n = await runner.reprocess_dead_letters_with(
    DLQFilter(event_type="billing.invoice.settled", tenant_id="acme"), 100
)  # filtered

# Step 2b — Discard poison pills that can never succeed.
n = await runner.discard_dead_letters(DLQFilter(event_type="legacy.sync.requested"), 1000)

DLQFilter fields are all optional (None = match all): event_type: str | None, tenant_id: str | None, failed_before: datetime | None.

Retryable failures (SNS throttling: ThrottlingException, ServiceUnavailable, InternalFailure, RequestTimeout) do not count toward max_attempts — the outbox uses threshold = max_attempts + 1 for these errors. A period of SNS unavailability will not dead-letter records that are otherwise healthy.

Pruning published records

# Delete published records older than 7 days, up to 1000 per call.
n = await runner.prune_published(older_than=timedelta(days=7), limit=1000)

older_than must be long enough that all consumers have processed the event before the row is deleted. 7 days covers most SLA windows.

Replay guarantees

Replay carries no stronger delivery guarantees than the original delivery path. Ordering is not preserved (replayed events re-enter the back of the poll queue), the same Envelope.id is presented again (idempotency guards must be in place), and replay runs concurrently with live traffic. A handler that is not idempotent is not replay-safe — and therefore not production-ready.

---

HMAC helpers

When to use HMAC

HMAC-SHA256 provides message authenticity and integrity — proof that the payload was produced by a party that holds the shared key and has not been tampered with in transit. It does not provide confidentiality or replay protection.

Scenario	HMAC required	Why
Receiving webhooks from external systems (Stripe, GitHub, etc.)	Yes	The channel is the public internet; the SNS/SQS IAM boundary does not apply
Cross-service call over an internal HTTP endpoint (not SNS/SQS)	Yes	Service-to-service HTTP has no built-in message-level auth
Zero-trust internal network where service identity is not enforced at the infra layer	Yes	HMAC adds app-layer authentication even when mTLS or IRSA is absent
Events flowing exclusively over SNS → SQS within the same AWS account and IAM boundary	No	SNS delivery is authenticated by IAM policies
Events signed by the outbox runner and delivered to an SQS queue you own	No	The publisher operates under your service's IAM role

from eventcommon import hmac as event_hmac

key = os.environ["WEBHOOK_SECRET"].encode()  # must be ≥ 32 bytes

# Sign
sig = event_hmac.sign(key, payload)

# Verify — constant-time; returns False on mismatch, never raises for a bad signature
if not event_hmac.verify(key, payload, sig):
    raise ValueError("invalid signature")

# Sign/verify a full envelope (serialises to canonical JSON first)
sig = event_hmac.sign_envelope(key, env)
ok = event_hmac.verify_envelope(key, env, sig)

verify uses hmac.compare_digest (constant-time) — never replace with string ==. Key length: sign raises KeyTooShortError for keys < 32 bytes.

Rules: always use verify_envelope (not verify) when checking a full envelope. Rotate keys by accepting both the current and previous key for a short window. Keys must be ≥ 32 bytes and stored in a secrets manager — never in source-controlled environment variables.

---

Observability — Prometheus and OTel

Both are optional. The publisher, consumer, and outbox runner work without calling metrics.init or having an OTel provider registered.

Prometheus metrics

from eventcommon import metrics

metrics.init(os.environ["APP_NAME"], os.environ["BUILD_VERSION"])  # idempotent — first caller wins

# For isolated test registries:
metrics.init_with_registry("test-svc", "v0.0.0", CollectorRegistry())

Registered metrics (identical names to the Go implementation, so Python and Go services feeding the same Grafana dashboards use one metric namespace):

Metric	Type	Labels	Description
events_published_total	Counter	service, topic, event_type, status	SNS publish attempts
events_publish_duration_seconds	Histogram	service, topic, event_type	SNS publish latency
events_consumed_total	Counter	service, queue, event_type, status	SQS messages processed (status = success/error/malformed)
events_consume_duration_seconds	Histogram	service, queue, event_type	Handler execution latency
outbox_pending_total	Gauge	service	Unpublished records in outbox_events
outbox_leased_total	Gauge	service	Records currently claimed (leased) by a runner
outbox_published_total	Counter	service, event_type, status	Records published by the runner
outbox_attempts_total	Counter	service, event_type	Total publish attempts by the runner
outbox_dead_letters_total	Counter	service, event_type	Records moved to outbox_dead_letters — alert on rate() > 0
outbox_dead_letters_reprocessed_total	Counter	service	Dead-letter records re-queued
outbox_dead_letters_discarded_total	Counter	service	Dead-letter records permanently deleted
sqs_receive_errors_total	Counter	service, queue	receive_message errors — alert on rate() > 0
sqs_delete_errors_total	Counter	service, queue	delete_message errors — non-zero rate causes duplicate delivery
outbox_poll_errors_total	Counter	service	Outbox poll cycle errors — triggers exponential backoff
outbox_mark_published_errors_total	Counter	service	mark_published failures after a successful SNS delivery
events_oversized_event_type_label_total	Counter	service	event_type values > 128 bytes, replaced with "__oversized__"

OpenTelemetry

OTel is always initialised by the consuming service (e.g. fastapicommon.tracing.init_from_env()). eventcommon calls trace.get_tracer("eventcommon") and produces no-op spans if no provider is registered.

SNS publish span: sns.publish with messaging.system=aws_sns, messaging.destination, messaging.message_id.

SQS receive span: sqs.receive with messaging.system=aws_sqs, messaging.operation=process, linked to the publisher's trace via Envelope.trace_id.

Logging correlation

Always include these four fields on every log line inside a handler or publisher: event_id (env.id), event_type (env.type), trace_id (trace_id_from_context()), tenant_id (env.tenant_id).

async def handle_user_created(env: Envelope[dict[str, Any]]) -> None:
    log = logger.bind(
        event_id=env.id,
        event_type=env.type,
        trace_id=trace_id_from_context(),
        tenant_id=env.tenant_id,
    )
    log.info("handling event")

    payload = UserCreatedPayload(**env.data)
    await repo.create_user(payload)

    log.info("event processed")

What NOT to log

Do not log	Reason
env.data (raw or parsed)	May contain PII, payment data, or credentials — log event_id instead and look up the payload if needed
Full error chains that include DSNs or URLs	DATABASE_URL and SNS_TOPIC_ARN may appear in wrapped errors — truncate or sanitise before logging
env.trace_id as an alert label	Trace IDs are high-cardinality — use them in log lines but not as alert labels

---

Backpressure considerations

Under sustained load, the SQS receive buffer and the outbox outbox_events table both grow if consumers or the outbox runner can't keep up.

Consumer-side (SQS)

Tuning order: 1) raise concurrency — costs only memory and an asyncio task; 2) scale pods horizontally — SQS distributes messages across them naturally; 3) increase SQS_MAX_MESSAGES (up to 10) only if handler latency is I/O-bound; 4) increase SQS_VISIBILITY_TIMEOUT if handlers legitimately take longer than the current timeout.

Outbox-side (Postgres → SNS)

Tuning order: 1) decrease OUTBOX_POLL_INTERVAL; 2) increase OUTBOX_BATCH_SIZE (keep below 100 to avoid long-held Postgres locks); 3) scale runner pods horizontally (SKIP LOCKED handles distribution with no coordination); 4) check SNS publish errors — a rising outbox_attempts_total/outbox_published_total gap usually indicates throttling, not a runner configuration problem.

Do not increase batch size before scaling pods on either side. Horizontal scaling is almost always preferable to a larger batch/receive size.

---

Recommended production defaults

# SQS Consumer
SQS_MAX_MESSAGES=10          # Always set to max (SQS hard limit)
SQS_WAIT_SECONDS=20          # Long-poll at max
SQS_VISIBILITY_TIMEOUT=60    # seconds — comfortably above your p99 handler latency; 30s default is tight
SQS_CONCURRENCY=5            # Start here; scale up to ~10 before considering horizontal pod scaling

# Outbox runner
OUTBOX_POLL_INTERVAL=2       # seconds — 5s default is conservative
OUTBOX_BATCH_SIZE=50         # 50-100 is a good range
OUTBOX_MAX_ATTEMPTS=5        # Default is fine unless your SNS target has known transient outages

SQS_VISIBILITY_TIMEOUT is the most commonly under-set value. If your handler calls a slow DB query or downstream HTTP endpoint, the default 30s may expire before the handler finishes, causing the message to be delivered twice. Set it to 2–3× your p99 handler duration.

What to monitor on day one

Metric	Alert threshold	Action
outbox_pending_total	> 500 sustained for 5 min	Lower OUTBOX_POLL_INTERVAL; add runner pods
ApproximateNumberOfMessages (CloudWatch)	> 1000 sustained	Raise SQS_CONCURRENCY; add consumer pods
events_consumed_total{status="error"}	> 1% error rate	Inspect handler errors; check DLQ depth
outbox_published_total{status="failed"} / total	> 1%	Check SNS reachability; inspect outbox_dead_letters

---

Configuration reference

Variable	Default	Notes
AWS_REGION	us-east-1	Applies to both SNS and SQS clients
SNS_TOPIC_ARN	—	Required for SNS publisher
SQS_QUEUE_URL	—	Required for SQS consumer
SQS_MAX_MESSAGES	10	1–10; SQS hard limit
SQS_WAIT_SECONDS	20	Long-poll duration
SQS_VISIBILITY_TIMEOUT	30	Seconds
SQS_CONCURRENCY	1	Parallel handler tasks
SQS_MAX_RECEIVE_COUNT	0 (unset)	Pairs with a dead_letter_handler
OUTBOX_POLL_INTERVAL	5	Seconds
OUTBOX_BATCH_SIZE	50	Records per poll cycle
OUTBOX_MAX_ATTEMPTS	5	Before moving to dead-letter
OUTBOX_CLAIM_LEASE_DURATION	600	Seconds; how long a claimed record is hidden from other runners
OUTBOX_STARTUP_JITTER	0	Seconds; use 5–10 with multiple replicas
OUTBOX_PUBLISH_CONCURRENCY	1	Parallel publishes per poll cycle
OUTBOX_PUBLISH_TIMEOUT	10	Seconds; per-record publish timeout
OUTBOX_DRAIN_TIMEOUT	30	Seconds; runner.stop() wait bound
DATABASE_URL	—	Postgres DSN for the outbox runner (via pgcommon)
AWS_ENDPOINT_URL	—	Set to http://localhost:4566 for LocalStack
OTEL_SERVICE_NAME	—	OTel resource attribute
OTEL_EXPORTER_OTLP_ENDPOINT	localhost:4317
APP_ENV	—	dev/development/local enables OTel insecure mode

Wiring helpers (eventcommon.config): load_sns / load_sqs / load_outbox / load_otel · log_warnings · runner_config_from_env · sqs_consumer_options

Copy .env-example to .env via make setup for local development.

---

Error reference

Error	Raised when
eventcommon.errors.EnvelopeIDRequiredError	parse_envelope or outbox.enqueue — missing id
eventcommon.errors.EnvelopeTypeRequiredError	parse_envelope or outbox.enqueue — missing type
eventcommon.errors.EnvelopeSourceRequiredError	parse_envelope or outbox.enqueue — missing source
eventcommon.errors.KeyTooShortError	sign / sign_envelope called with a key < 32 bytes
eventcommon.errors.InvalidSignatureError	Internal parse failure in verify
eventcommon.errors.BatchTooLargeError	Internal sentinel; publish_batch splits automatically at 10 — never raised to callers
eventcommon.errors.RetryableError	Wraps a transient SNS/AWS error so the outbox runner does not count it toward max_attempts

---

Testing in consuming services

eventcommon.mock ships ready-made, thread-safe test doubles so consuming services never need to stand up LocalStack or SNS just to run a unit test.

Mock Publisher

from eventcommon.mock import MockPublisher


async def test_place_order_publishes_event():
    pub = MockPublisher()
    svc = OrderService(pub)

    await svc.place_order(Order(id="ord-1", amount=99))

    assert len(pub.published) == 1
    assert pub.published[0].type == "order.placed"
    assert pub.published[0].tenant_id == "acme"


async def test_place_order_publish_error_is_handled():
    pub = MockPublisher()
    pub.set_error(RuntimeError("sns unavailable"))

    svc = OrderService(pub)
    with pytest.raises(RuntimeError):
        await svc.place_order(Order(id="ord-2", amount=50))

    assert pub.published == []  # nothing was recorded

Method / attribute	What it does
await publish(env)	Records the envelope; raises any error set via set_error
await publish_batch(envs)	Calls publish for each envelope
published	List of all recorded envelopes (thread-safe)
set_error(err)	Makes all subsequent publish calls raise err
reset()	Clears recorded envelopes and any configured error

Mock Consumer

from eventcommon.mock import MockConsumer


async def test_notification_worker_handles_user_created():
    consumer = MockConsumer()
    worker = NotificationWorker(consumer)

    await consumer.start()  # no-op — no task needed
    worker.register_handlers()

    env = new_envelope(
        "iam.user.created",
        "platform-iam",
        {"user_id": "u-123", "email": "[e-mail]"},
        tenant_id="acme",
    )
    await consumer.inject(env)  # delivers directly — no SQS, no network

    assert worker.email_was_sent_to("[e-mail]")

Method / attribute	What it does
await start()	No-op; marks consumer as running
await stop()	No-op; marks consumer as stopped
set_handler(fn)	Registers the handler called by inject
await inject(env)	Delivers the envelope synchronously to the registered handler
is_running	Whether start has been called without a matching stop

Both mocks satisfy the same Publisher/Consumer protocols as the real implementations, so they drop in wherever the real ones are used.

---

Testing

make test-unit   # unit tests — no Docker required
make test-int    # integration tests — spins up LocalStack + Postgres (testcontainers)
make test-all    # unit + integration + e2e in one pytest run
make cover       # HTML coverage report

End-to-end tests cover the full outbox pipeline (Postgres → runner → SNS → SQS → consumer) and require Docker for LocalStack + Postgres via testcontainers. Run a single test by name:

uv run pytest tests/unit/test_envelope.py -k test_sign -v
uv run pytest tests/integration -m integration -k test_sns_publish_round_trip -v

---

CI

ci.yml — triggered on every push and pull request to main: validate (format check, lint, mypy --strict, pip-audit, full test suite) → coverage (Python 3.12/3.13/3.14 matrix) → build (wheel + sdist) → interop (cross-language compatibility against platform-events).

release.yml — triggered on v*.*.* tags: same validate/coverage/build pipeline, then publishes to PyPI (Trusted Publisher, no stored token) and creates a GitHub Release from the tagged CHANGELOG.md section.

validate.yml — reusable workflow fanning out to quality.yml (lint/type/format/audit) and test.yml (full suite), called by both ci.yml and release.yml.

---

Versioning and releases

Version bump	When
MAJOR	Breaking change in the public API (any module not under eventcommon._internal), or in the Envelope wire format / HMAC canonicalisation
MINOR	New backward-compatible capability
PATCH	Bug fix, performance improvement, documentation correction

git tag -a v0.2.0 -m "v0.2.0"
git push origin v0.2.0   # triggers the release workflow automatically

See VERSIONING.md for the full SemVer policy, the envelope wire-format compatibility rules, PyPI publishing setup, and the maintainer release checklist.

License / ownership

BCBP Solutions FZC LLC — internal platform shared library.

---

Service adoption checklist

Publisher checklist

• metrics.init(app_name, build_version) called once at startup
• Outbox schema applied via apply_schema(migrate_runner) on startup
• outbox.Runner started as a background task and stop() awaited on shutdown
• All domain-event publish call sites use outbox.enqueue inside pgcommon.run_in_tx — no bare publisher.publish for transactional events
• tenant_id, trace_id, and schema_version passed to every new_envelope call in HTTP handler context
• OUTBOX_* env vars wired via runner_config_from_env
• Readiness probe waits for runner.ready before marking the pod ready

Consumer checklist

• SNS→SQS subscription created with RawMessageDelivery=true — without it the consumer deletes messages as malformed
• SQS_* env vars wired via load_sqs + sqs_consumer_options
• SQSConsumer wired with concurrency appropriate for handler latency
• SQS_VISIBILITY_TIMEOUT set to ≥ 2× p99 handler duration
• Every handler implements idempotency via INSERT INTO processed_events ... ON CONFLICT DO NOTHING
• Handlers classify errors as transient (raise) vs. permanent (return + log)
• Handlers check env.type before parsing env.data

Observability checklist

• Handler entry logs bind event_id, event_type, trace_id, tenant_id
• OTel provider initialised before the first consumer.start or publisher.publish call
• Alerts configured on outbox_pending_total, events_consumed_total{status="error"}, and SQS ApproximateNumberOfMessages

Security checklist

• HMAC signing enabled on HTTP/webhook ingress that receives events from external systems — not on the SNS/SQS messaging path
• HMAC keys ≥ 32 bytes, stored in a secrets manager — not in source-controlled environment variables
• SYSTEM_TENANT_ID usage reviewed — no HTTP handler context uses it to avoid a tenant lookup

---

See also

Document	Description
ARCHITECTURE.md	Layer model, sequence diagrams, invariants, performance
EVENT_SCHEMA_GOVERNANCE.md	Event type naming, payload evolution rules, the event type registry, the Python unknown-field trap
VERSIONING.md	SemVer rules, envelope wire-format guarantees, PyPI publishing, release process
CHANGELOG.md	Per-version changes
CONTRIBUTING.md	Development setup, adding adapters, PR checklist
SECURITY.md	Vulnerability reporting, trust model, supported versions

---

See Publishing rules for the full decision table. The short version: publisher.publish is only valid when event loss is explicitly acceptable. For any event that drives downstream state, use outbox.enqueue inside a pgcommon.run_in_tx block — no exceptions.

eventcommon must be the only entry point for event publishing and consumption to preserve delivery guarantees, tenant isolation, and uniform observability.