eventcommon

Opinionated async SNS/SQS messaging layer that enforces at-least-once delivery, tenant propagation and observability by making the safe path the default, and the unsafe path visibly wrong.

This is the platform's event-driven messaging library for Python services, an asyncio library on top of aioboto3 and pgcommon. Every Python platform service imports it to publish and consume domain events.

It is the Python port of platform-events (Go) with full feature parity, currently tracking Go v2.2.1, reimplemented idiomatically for asyncio. It owns the messaging boundary:

  • the Envelope wire format, byte-identical to Go's, and its HMAC canonicalisation;
  • the SNS publisher and the SQS consumer;
  • dead-letter forwarding;
  • the transactional outbox and the idempotency inbox;
  • tenant propagation into pgcommon's RLS context;
  • the messaging telemetry every service emits: Tier 1 platform_* metrics, OTel spans and Go's log messages.

The few places where Python cannot match Go are listed in ARCHITECTURE.md § Divergences from Go.

Repository: github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon (private)

Package: eventcommon 0.4.0 (on PyPI; release status) · Apache-2.0 · Python ≥ 3.12 (CI runs 3.12, 3.13 and 3.14) · consumed as a dependency, never deployed on its own. Each release also ships the example outbox worker as a container image (ghcr.io/bcbp-solutions-fzc-llc/platform-eventcommon): a reference deployment of the runner, not a service to deploy.

Design: one src/eventcommon package of public modules on top of a private _internal/ layer, with the layer rules enforced by import-linter in CI. See ARCHITECTURE.md for design decisions, layer diagrams, the failure model and how the asyncio/contextvars model differs from Go's context.Context. This README is how to use it.

Features: typed event envelopes (CloudEvents-aligned) · async SNS publisher (single, batch, FIFO) · async SQS consumer (concurrency, FIFO groups, visibility extension, retry backoff, handler timeouts) · DLQ forwarding · transactional outbox (Postgres via pgcommon, per-key ordering, dead letters) · idempotency inbox · HMAC signing · schema-registry codec hook · Prometheus metrics · OpenTelemetry tracing · RLS tenant propagation · test doubles.

Cross-language contract: the Envelope wire format and HMAC canonicalisation must stay byte-for-byte compatible with platform-events. The two publish to and consume from the same SNS topics, SQS queues and outbox tables in a polyglot deployment. Any change to envelope.py or hmac.py must be cross-checked against the Go side. The golden-vector tests and the interop CI job exist to catch a mismatch before it reaches production; see CI.

Reading this on PyPI? Links to other files (ARCHITECTURE.md, CHANGELOG.md, docs/, …) are relative to the source repository, which is private, so they do not resolve here. Every linked document ships in the source distribution: pip download --no-binary :all: --no-deps eventcommon and unpack the .tar.gz.

Contents

Mental model

This library owns It does NOT own
The Envelope wire format (new_envelope, parse_envelope, to_json), byte-identical to Go's, and its HMAC canonicalisation The payload schema. data is yours; EVENT_SCHEMA_GOVERNANCE.md has the rules
Publishing: SNSPublisher (single, batch, FIFO), message attributes for filter policies, trace-context propagation, error classification, call_timeout SNS topics, SQS queues, subscriptions, filter policies and RedrivePolicies (infrastructure)
Consuming: SQSConsumer (long poll, concurrency, FIFO groups, visibility extension, retry backoff, handler and drain timeouts, malformed-body handling, dead-lettering) and SQSDLQPublisher What your handler does, and its idempotency. The library delivers at least once; your side effects must tolerate it
Tenant propagation: the consumer binds each message's tenant_id into pgcommon's GUC context, so every pool call in the handler runs under RLS Your tables and RLS policies, and the pool (pgcommon). eventcommon supplies the tenant; your policies enforce it
The transactional outbox (enqueue, Runner, outbox_events / outbox_dead_letters, golang-migrate-compatible schema shared with Go) and the inbox (processed_events) Your business transaction. enqueue joins it; it never opens one
Messaging metrics, spans and logs, registered on your CollectorRegistry, tracer provider and logger Your Prometheus exposition, OTel SDK and logging setup (fastapicommon)
The Codec hook and a decode-only GlueDecodeCodec A schema-registry client. You implement Codec against yours
Test doubles (eventcommon.mock) —
— Metric ratification and alert routing (the vendored monitoring/ bundle is Go's, unchanged) 
HTTP request (producer)
  → your handler ── await pool.run_in_tx(tx)
      → tx(conn): domain write + await outbox.enqueue(conn, new_envelope(type, source, data,
                                                     tenant_id=rc.tenant_id, trace_id=rc.trace_id))
      → COMMIT ── the event is durable with the write, or neither is
  → outbox.Runner (background) ── claim (SKIP LOCKED + lease) → SNS publish → mark published
  → SNS → SQS (RawMessageDelivery=true)
  → SQSConsumer ── receive → malformed check → decode → bind tenant / trace / source message
      → your handler(env) ── pool.with_conn / run_in_tx now scoped to env.tenant_id by RLS
      → returns: delete · raises: retry (backoff) · past max_receive_count: dead-letter + DLQ

Event flow

This diagram shows the life of an event from the domain write that produces it to the handler that consumes it. Every box is either owned by the library or governed by its configuration.

Think of eventcommon as the only door between a service and the bus: every event passes through it to gain durability, tenancy and observability guarantees.

flowchart TD
    A[HTTP request — producer service]
    B["pgcommon.run_in_tx\ndomain write + outbox.enqueue(conn, env)"]
    C[("outbox_events\nPostgreSQL")]
    D["outbox.Runner\nclaim · publish · settle"]
    E[SNSPublisher]
    F[SNS topic → SQS queue\nexternal infrastructure]
    G["SQSConsumer\nreceive · validate · decode"]
    H["handler context\ntenant GUC · trace ID · source message"]
    I["your handler(env)\nidempotent on env.id"]
    J[(PostgreSQL + RLS\nconsumer's database)]

    A --> B --> C --> D --> E --> F --> G --> H --> I --> J

    style B fill:#fff3cd,stroke:#856404
    style D fill:#d1ecf1,stroke:#0c5460
    style E fill:#d1ecf1,stroke:#0c5460
    style G fill:#d1ecf1,stroke:#0c5460
    style H fill:#d1ecf1,stroke:#0c5460
    style F fill:#e2e3e5,stroke:#6c757d
    style J fill:#d4edda,stroke:#155724

Read top → bottom: each layer adds a guarantee (durability → delivery → tenancy → observability) before the event reaches your handler.

Detailed annotated view:

HTTP request (producer)
    │
    ▼
pgcommon.run_in_tx(pool, tx)             ← caller's responsibility: one business transaction
    │  tx(conn): your domain write
    │            await outbox.enqueue(conn, env)   ← validated, ≤ 240 KB, no SNS call
    │  COMMIT: both rows durable, or neither
    │
    ▼
outbox_events                            ← PostgreSQL (platform-events' schema, shared with Go)
    │
    ▼
outbox.Runner                            ← background task: await runner.start()
    │  claim: FOR UPDATE SKIP LOCKED + a claim lease (several runners, Go or Python)
    │  publish: publish_batch (10 per request) or publish per record, publish_timeout
    │  settle: published · transient → released, no attempt · permanent → backoff
    │          · max_attempts → outbox_dead_letters
    │
    ▼
SNSPublisher                             ← attributes EventType · TenantID · Source · EventID · Subject
    │                                       + traceparent; span sns.publish(.batch)
    ▼
SNS topic → SQS queue                    (external infrastructure; RawMessageDelivery=true)
    │
    ▼
SQSConsumer                              ← await consumer.start()
    │  long poll · concurrency · visibility extension · FIFO group order
    │  malformed body → DLQ or delete (never retried) · codec decode
    │
    ▼
handler context                          ← automatic, per message, reset afterwards
    │  pgcommon.with_guc_set(GUCSet(tenant_id=env.tenant_id))
    │  trace ID · source message · span sqs.receive (linked to the producer)
    │
    ▼
your handler(env)                        ← idempotent on env.id (inbox.handler / inbox.Store)
    │  returns → message deleted · raises → retried · past max_receive_count → DLQ
    │
    ▼
    ├── Prometheus metrics               ← Tier 1 platform_* only
    ├── OTel spans                       ← no-op without the service's TracerProvider
    └── logs                             ← Go's messages, through the logger you pass

Invariant: every domain event tied to a database write goes through outbox.enqueue inside the business transaction. A direct publisher.publish after the commit is a silent data-loss bug, not a shortcut.

Key invariants:

  • Delivery is at least once on both sides, so every handler must be idempotent on Envelope.id.
  • Tenant context is bound by the consumer before your handler runs, and reset after it; your handler never sets a GUC.
  • Transient failures (throttling, outages, timeouts, shutdown) never count an outbox attempt, so an SNS outage cannot dead-letter healthy events.
  • Observability is wired once (metrics.init_metrics, the service's tracer provider and logger); there is no per-call code in services.

Ownership summary:

  • Service: the business transaction, handlers and their idempotency, RLS policies
  • eventcommon: envelope, delivery, retries, dead-lettering, tenant binding, observability
  • Infra: SNS, SQS, IAM, PostgreSQL

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. Producers do not direct consumers; consumers decide independently how to react. So consumers must be resilient to receiving the same fact more than once (Envelope.id idempotency), and 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 on an empty queue_url or visibility_timeout > 12h, and all three AWS components when no region resolves. Invalid configuration is rejected at construction, not at first use. A consumer with DLQ forwarding refuses to start against a queue without a usable RedrivePolicy.
  • Make safe usage the default. The SQS consumer propagates tenant_id into pgcommon's GUC context for RLS, deletes only after a handler returns, and drains in-flight messages on stop(). With visibility_timeout set, it extends slow handlers' messages. The default of 0 uses the queue's own timeout and extends nothing, as in Go.
  • Keep business logic free from messaging plumbing. Handlers receive an Envelope. Retry backoff, visibility extension, metric recording and OTel spans are invisible to them.
  • Centralise cross-cutting concerns. HMAC signing, tenant propagation, dead-lettering and observability live here, not in every service. A single version bump propagates to all consumers.

TL;DR

⚠️ Exactly-once delivery is not possible anywhere in this system. SNS and SQS deliver at least once: a message may arrive more than once under normal operation, through visibility timeout expiry, a consumer crash mid-handler, or 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 whether or not FIFO is used. See Implementing idempotency.

  • Use eventcommon.envelope.new_envelope(...) to create 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 on the consumer and trace continuity.
  • Use await outbox.enqueue(conn, env) inside a pgcommon.run_in_tx block for every event tied to a database write. See Publishing rules.
  • Never call publisher.publish for domain events tied to a DB write. Event loss on a process crash is silent and unrecoverable.
  • Make every handler idempotent on Envelope.id: wrap it with inbox.handler(store, handle).
  • Call eventcommon.metrics.init_metrics(MetricsIdentity(domain=..., service=...)) once at startup. The identity is mandatory, and the library emits only Tier 1 platform_* metrics.
  • OTel tracing is enabled by the consuming service (e.g. fastapicommon.tracing.init_from_env()), not by this library.

Why this library exists

Every platform service is multi-tenant, talks over SNS/SQS and keeps its state in Postgres. Each needs the same half-dozen messaging things done exactly right, and done separately in each service, each is a production incident waiting to happen:

  • No lost events. A domain write that commits while its event is lost leaves every downstream service wrong, with no error anywhere. The outbox makes the event part of the transaction and redelivers it until SNS accepts it. A full SNS outage only delays events.
  • No poison loops. A malformed message or a handler bug must not loop forever, block a FIFO group or burn capacity. Malformed bodies are dead-lettered on first receipt, failing ones after max_receive_count, and outbox records after max_attempts, each counted with its reason.
  • Tenant isolation in the consumer. A handler must run under the tenant of the event it is handling, never the previous message's. The consumer binds it into pgcommon's RLS context per message and resets it afterwards, so a handler cannot forget it.
  • One wire format across languages. Go and Python services exchange envelopes, signatures and outbox rows. A field order, time format or JSON escape that differs breaks signatures and parsing in production. eventcommon is checked byte for byte against Go in CI.
  • Uniform telemetry. Every service exports the same messaging metrics with the same names, labels and semantics as the Go library. One dashboard and one alert set cover both languages (Tier 1 of the Enterprise Platform Observability Standard).

A fix lands in every service with a version bump. This is enforced structurally: eventcommon._internal is private (a leading underscore, never re-exported, outside the SemVer surface), and the import-linter contracts in pyproject.toml encode the dependency-direction rules CI checks on every PR.

Why eventcommon instead of raw aioboto3?

Concern Raw aioboto3 SNS/SQS eventcommon
Event loss after a DB commit A publish after COMMIT is lost on a crash; nobody notices outbox.enqueue in the transaction; the runner redelivers until SNS accepts
Envelope format Each service invents one; Go and Python drift One Envelope, byte-identical to Go's, checked against golden vectors
Retry classification Every error retried or none; throttling dead-letters healthy events Go's classification: transient errors never count an attempt; permanent ones back off and dead-letter
Batch publishing Manual 10-entry / 256 KiB splitting, partial-failure bookkeeping publish_batch splits, attempts every request, raises one BatchError keyed by envelope ID
Visibility Manual ChangeMessageVisibility loops; slow handlers redelivered while running visibility_timeout extends every message from receipt until settled
Poison messages Loop until SQS's redrive, without a reason Malformed → DLQ at once; max_receive_count → dead-letter handler + DLQ with DLQReason, OriginalQueue, FailedAt
Tenant context in handlers Each handler must set the RLS GUC itself, and reset it Bound per message from env.tenant_id and reset afterwards
Shutdown In-flight handlers killed, or waited on forever stop() drains up to drain_timeout, then cancels; unsent outbox records released at once
Metrics and traces Hand-written counters; no producer → consumer trace link Tier 1 platform_* metrics and linked spans, identical to Go's
Idempotency Every team writes its own table inbox.handler / inbox.Store.process on a shared schema
Testing An AWS emulator for every unit test MockPublisher, MockConsumer, MockDLQPublisher with the real dispatch rules

The single-sentence justification: every row above is a bug we have seen in services that used raw SNS/SQS calls: lost events, duplicate side effects, poison loops, tenant leaks, invisible failures. eventcommon makes the correct path the default path.

This is not an abstraction over aioboto3. It is a guardrail layer that enforces delivery and tenancy guarantees by construction.

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 for:

  • One-off scripts or CLIs. A plain aioboto3 SNS call is simpler, without the pool setup.
  • Services that consume a single event type, where filtering, retries and concurrency are handled entirely by the queue configuration. A thin wrapper is enough.
  • Non-AWS message brokers. The library is SNS/SQS-specific; EventBridge, Kafka and other backends are not supported without a new adapter (CONTRIBUTING.md).
  • Services without PostgreSQL. pgcommon is a hard dependency regardless, because the SQS consumer binds the tenant GUC context for every message, not just for the outbox. A service with no database can still use the publisher and consumer and accept at-most-once publishing; it simply never calls outbox.enqueue or outbox.Runner.

API overview

The public modules are listed in VERSIONING.md § What counts as public API. Import only those, never eventcommon._internal. The source of truth is each module's __all__, docstrings and type hints; the package ships py.typed and is checked with mypy --strict. ARCHITECTURE.md § Public API reference has the full tables.

eventcommon.envelope — the wire format

Symbol Purpose
new_envelope(event_type, source, payload, *, tenant_id, system_tenant, trace_id, correlation_id, schema_version, subject, actor, ip_address, user_agent, schema_id) Build an envelope: UUID v7 id, time = now(UTC)
parse_envelope(data, payload_type=…, require_time=True) Parse and validate, as Go's ParseEnvelope; payload_type builds a typed payload, ignoring unknown fields
Envelope.to_json() · Envelope.data_json_bytes() What Go's json.Marshal writes, byte for byte · data's bytes as received
decode_payload(data, YourDataclass) Typed payload from env.data inside a handler, dropping undeclared fields
SYSTEM_TENANT_ID · format_time · parse_time · ZERO_TIME · LongInt "system" · Go-compatible time helpers · integers beyond Python's 4,300-digit limit
trace_id_from_context() The envelope's trace ID inside a handler

eventcommon.publisher — SNS

Symbol Purpose
SNSPublisher(SNSConfig(topic_arn, region, endpoint_url, logger, call_timeout), *options, message_group_id, message_deduplication_id, attributes, codec, client) The publisher. Validates at construction; one client per event loop
await publisher.publish(env) · await publisher.publish_batch(envs) One event · many, split at 10 entries / 256 KiB, one BatchError for any failure
await publisher.close() · async with publisher: Close every loop's client (never an injected client=)
Publisher (protocol) · with_codec, with_attributes, with_message_group_id, with_message_deduplication_id What the outbox runner accepts · Go's functional options

eventcommon.consumer — SQS and DLQ

Symbol Purpose
SQSConsumer(SQSConfig(queue_url, region, endpoint_url, max_messages, wait_seconds, logger), handler, *options, concurrency, visibility_timeout, handler_timeout, retry_backoff, max_retry_backoff, max_receive_count, dead_letter_handler, dlq_publisher, drain_timeout, codec, malformed_body_logging, queue_depth_interval, client) The consumer
await consumer.start() (as a task) · await consumer.stop() Long-poll until stopped · drain, then cancel at drain_timeout
Handler · AcceptedHandler · DeadLetterHandler · EventData Handler types: middleware and routers use Handler; what the consumer accepts is AcceptedHandler
SQSDLQPublisher(DLQConfig(region, endpoint_url, consumer_name, logger, cache_ttl, strict_attributes)) Forwards to the source queue's RedrivePolicy DLQ: send_to_dlq, resolve_dlq, close
forward_source_message(dlq, reason) Dead-letter the message the current handler is processing
handler_context · validate_dlq_message · parse_redrive_policy · parse_sqs_queue_arn · dlq_event_type · DLQ_ATTR_* Helpers, as Go exports them
with_concurrency, with_visibility_timeout, with_handler_timeout, with_retry_backoff, with_max_receive_count, with_dead_letter_handler, with_dlq_forwarding, with_drain_timeout, with_consumer_codec, with_malformed_body_logging, with_queue_depth_metrics Go's functional options

eventcommon.outbox and eventcommon.inbox — Postgres

Symbol Purpose
await outbox.enqueue(conn, env) · await outbox.enqueue_ordered(conn, env, key) Insert inside the caller's transaction · with per-key publish order
outbox.Runner(RunnerConfig(...)) · start() · stop() · ready · healthy(max_age) The poll runner; see Transactional outbox
runner.list_dead_letters · reprocess_dead_letters(_with) · discard_dead_letters · prune_published Operator tools (DLQFilter, DeadLetterRecord)
await outbox.apply_schema(pool) · await inbox.apply_schema(pool) Apply platform-events' migrations, golang-migrate-compatible
outbox.OutboxStore · PostgresStore · OutboxRecord The store port and its adapter
inbox.Store(pool, consumer) · inbox.handler(ledger, handle) is_processed, mark_processed, process(env, fn), prune(retention) · dedup wrapper

eventcommon.metrics, hmac, codec, context, mock, config

Symbol Purpose
metrics.init_metrics(MetricsIdentity(domain, service, environment), registry=None, *, event_type_limit, event_types) · metrics_identity_from_env(domain) The only init: registers the Tier 1 platform_* metrics; returns list[RegistrationWarning]
hmac.sign · verify · sign_envelope · verify_envelope HMAC-SHA256 over Go's canonical JSON
codec.Codec (protocol) · NoopCodec · GlueDecodeCodec · wrap_codec_payload · unwrap_codec_payload Schema-registry hook
context.source_message_from_context() · record_batch_published(ids) The raw SQS message in a handler · progress reporting for custom publishers
mock.MockPublisher · MockConsumer · MockDLQPublisher Test doubles
config.load_sns · load_sqs · load_outbox · log_warnings · sns_config_from_env · sqs_config_from_env · sqs_consumer_options · runner_config_from_env · mask_dsn Environment wiring with Go's names, defaults and warning texts

eventcommon-config CLI

Command Purpose
eventcommon-config Print the effective configuration (credentials masked), configuration warnings and problems, and production wiring reminders (Go: cmd/platform-events)
eventcommon-config --strict The same; exits 1 when a required variable is missing — an init-container or CI smoke check

Input validation

eventcommon validates everything it can before a network call or a database write, so misconfiguration fails at startup and a bad envelope fails at the call site, not in a consumer three services away. Environment variables are the exception to "fail fast": the load_* loaders never raise for a bad value. They apply the default and add a warning (Go's exact text) to .warnings, which you must log.

Input Rule Checked by Raises
SNSConfig.topic_arn Non-empty; starts with arn:aws:sns:, arn:aws-cn:sns: or arn:aws-us-gov:sns:. A .fifo topic needs message_group_id SNSPublisher(...) ValueError
SQSConfig.queue_url Non-empty; visibility_timeout ≤ 12 h. (max_messages outside 1–10 and wait_seconds outside 1–20 are not errors: a warning, then 10 / 20, as Go) SQSConsumer(...) ValueError
AWS region region=, else AWS_REGION, else boto3's chain (AWS_DEFAULT_REGION, the profile) — the Go SDK's order SNSPublisher, SQSConsumer, SQSDLQPublisher ValueError("…: no AWS region configured — set AWS_REGION (or the config's region)")
Envelope to publish Non-empty id / type / source; valid JSON (NaN, ±Infinity, naive datetime rejected); ≤ 10 message attributes after trace context publish, publish_batch (per entry), MockPublisher ValueError; per entry BatchFailure (InvalidEnvelope, MarshalError, TooManyAttributes)
Envelope to enqueue As above, plus non-zero time, a canonical lowercase UUID id, no NUL anywhere, serialised ≤ 240 KB, an open transaction outbox.enqueue / enqueue_ordered ValueError, Envelope*RequiredError
Ordering key Non-empty, ≤ 256 bytes (MAX_ORDERING_KEY_LEN) enqueue_ordered ValueError
Received body An object with id, type, source; no NUL in tenant_id / trace_id; parsed as Go's decoder parses it SQSConsumer, MockConsumer.inject Not raised: the message is malformed (DLQ or delete) · MalformedEnvelopeError from the mock
Envelope to parse As Go's ParseEnvelope; time required unless require_time=False. A null body is an envelope with no fields (Go's no-op unmarshal), so it is missing its ID parse_envelope EnvelopeParseError, Envelope{ID,Type,Source,Time}RequiredError
apply_schema(lock_timeout=…) Finite, > 0, at most 2,147,483,647 ms; rounded up to whole milliseconds ('0ms' would mean no timeout in PostgreSQL) outbox.apply_schema, inbox.apply_schema ValueError, before any SQL
HMAC key ≥ 32 bytes sign, sign_envelope, verify_envelope KeyTooShortError
Metric identity domain, service, environment required and well-formed; the message names APP_ENV / ENVIRONMENT when the bad value came from there init_metrics (nothing changes on failure) InvalidMetricsIdentityError
RunnerConfig publisher, and pool or store, required; claim_lease_duration ≥ batch_size × publish_timeout + 60s (30 s when the timeout is disabled). Zero values mean the default Runner(...) ValueError
DLQ forward Source URL, body, reason, SQS-legal characters, attribute names, ≤ 1 MiB, ≤ 10 attributes with strict_attributes send_to_dlq DLQInvalidMessageError, before any AWS call
Inbox event ID A UUID inbox.Store InvalidEventIDError
Environment variables Go's time.ParseDuration and integer rules: a bare 30 is not a duration load_sns / load_sqs / load_outbox Never raises: default + warning; .validate() raises ConfigValidationError for a missing required variable

Notable validation rules

Rule Enforcement
An event that could never publish is never persisted enqueue rejects envelopes over 240 KB (under SNS's 256 KB), NUL characters (jsonb cannot store \u0000, so the insert would roll back the business transaction on every retry) and non-UUID IDs
A malformed message never reaches a handler, and is never retried The consumer forwards it to the DLQ or deletes it on first receipt; its log line carries size + SHA-256, never the body
A NUL never reaches a GUC A NUL in tenant_id / trace_id makes a received message malformed
A bare number is not a duration SQS_VISIBILITY_TIMEOUT=60 is rejected (warning, default applied), exactly as Go rejects it — write 60s
Metric labels are bounded event_type capped at 200 values (__other__) and 128 bytes (__oversized__); queue / topic are names, never URLs or ARNs
A misconfigured DLQ fails at startup With dlq_publisher, start() raises when the queue has no usable RedrivePolicy, instead of looping poison messages

Error reference

Error Raised by When
EnvelopeIDRequiredError · EnvelopeTypeRequiredError · EnvelopeSourceRequiredError parse_envelope, outbox.enqueue A missing id / type / source
EnvelopeTimeRequiredError parse_envelope A missing or zero time (unless require_time=False)
EnvelopeParseError (a ValueError) parse_envelope Invalid JSON, a body that is not an object (an array, number, string or boolean; null is missing its ID instead, as in Go), a field of the wrong type, a non-RFC 3339 time
KeyTooShortError sign, sign_envelope, verify_envelope A key under 32 bytes
InvalidSignatureError — Exported for parity; never raised (verification returns False)
CodecEncodeError SNSPublisher The codec failed to encode (wrapped in RetryableError when the codec's error was retryable)
BatchError publish_batch One or more entries failed; .failures lists BatchFailure(id, code, message, retryable)
RetryableError SNSPublisher Wraps a transient SNS/AWS error so the outbox does not count an attempt; str() is the cause's text. Test with errors.is_retryable(exc), which follows __cause__
DLQNotConfiguredError · DLQInvalidRedrivePolicyError SQSDLQPublisher, SQSConsumer.start() The source queue has no RedrivePolicy · it is unparseable or not an SQS ARN
DLQUnresolvedError · DLQSendFailedError send_to_dlq The DLQ lookup failed or timed out (30 s) · the send failed (.retryable for throttling, 5xx and timeouts)
DLQInvalidMessageError send_to_dlq The message is not valid for SQS (before any AWS call)
InvalidMetricsIdentityError (a ValueError) init_metrics An invalid metric identity
InvalidEventIDError · TxEndedInCallbackError · TxCommitRollbackError (eventcommon.inbox) inbox.Store The envelope ID is not a UUID · process's callback ended the transaction · it swallowed a failed statement, so the transaction rolled back. The message is retried
MigrationDirtyError · SchemaVersionNotInSourceError (eventcommon.outbox) apply_schema A Go migration failed half-way · the database was migrated by a newer release
ConfigValidationError (eventcommon.config) .validate() A required environment variable is missing

All errors in eventcommon.errors subclass EventCommonError; check with isinstance(exc, EventCommonError), the Python equivalent of Go's errors.Is against the package's sentinels.

Architecture

The Go module's Clean Architecture layers become one flat package in four layers: composition roots (cli, config, mock), transports (publisher, consumer, outbox, inbox), a shared core (envelope, hmac, codec, context, metrics) and two sinks (errors, _internal). Full layer diagrams, flows, invariants, the failure model and the asyncio translation notes are in ARCHITECTURE.md; the standalone Mermaid sources are in docs/architecture/.

platform-eventcommon/
├── src/eventcommon/                 # The package (the wheel ships only this)
│   ├── __init__.py                  # Re-exports of the shared core
│   ├── envelope.py · errors.py      # Envelope, new_envelope, parse_envelope · EventCommonError and every sentinel
│   ├── hmac.py · codec.py           # sign / verify over Go's canonical JSON · Codec, NoopCodec, GlueDecodeCodec
│   ├── context.py                   # trace ID, source message, batch progress and DLQ attribution (contextvars)
│   ├── metrics.py · metrics_registry.py  # init_metrics, MetricsIdentity · the governed Tier 1 platform_* inventory
│   ├── publisher/sns.py             # SNSPublisher, SNSConfig, options
│   ├── consumer/sqs.py · dlq.py     # SQSConsumer, SQSConfig, options · SQSDLQPublisher, DLQConfig
│   ├── outbox/                      # enqueue, Runner, PostgresStore, dead-letter tools, apply_schema, migrations/ (001–011)
│   ├── inbox/                       # Store, handler, apply_schema, migrations/ (001)
│   ├── config/ · cli.py             # Environment loaders and wiring · eventcommon-config
│   ├── mock/                        # MockPublisher, MockConsumer, MockDLQPublisher
│   └── _internal/                   # PRIVATE: AWS clients, canonical JSON, Go durations and formatting, logging, migrations, UUID v7, …
├── examples/outbox_worker/          # The example worker (the container image)
├── tests/                           # unit (no Docker) · integration (Postgres + floci) · e2e (full pipeline); fixtures/
├── monitoring/                      # Go's Prometheus rules + tests, Grafana dashboard, KEDA example (vendored, byte-identical)
├── docs/                            # architecture/ (Mermaid sources), guides/, observability/
├── docker/ · docker-compose.yaml    # floci (SNS/SQS), Postgres and the observability stack for local work
├── .github/workflows/ · scripts/    # ci, validate-test, validate-quality, release, docs, changelog-check · their job logic
├── Dockerfile · build-constraints.txt  # The example-worker image (digest-pinned bases) · the hash-locked build backend
└── pyproject.toml · uv.lock · Makefile · .env-example

Dependency rules (enforced by import-linter in CI)

The contracts live in pyproject.toml ([tool.importlinter]); make arch-lint runs them, as validate-test's test job does.

Component May depend on
eventcommon.errors Nothing in the package (a leaf)
eventcommon._internal.* errors only (one function-level exception: canonical_json → envelope.format_time)
The shared core: envelope, hmac, codec, context, metrics, metrics_registry Each other and the sinks — never a transport, config, cli or mock
The transports: publisher, consumer, outbox, inbox The core and the sinks — never each other, config, cli or mock
config, cli, mock Anything
The package root (eventcommon/__init__.py) The shared core only

Plus the RLS guard, a grep in validate-quality: no session-level SET app.tenant_id|user_id|tenant_roles in src/. The tenant is bound only through pgcommon (with_guc_set), which the pool sets with set_config and resets.

Runtime dependencies (what consumers inherit)

Library Range (pyproject.toml) Locked in uv.lock Purpose
pgcommon >=0.4.0,<0.5.0 0.4.0 Pool and transactions for the outbox / inbox; GUC context for tenant binding
aioboto3 >=15.5.0 15.5.0 Async SNS / SQS clients
boto3 >=1.40.61 1.40.61 Error classes and the credential / region chain
prometheus-client >=0.26.0 0.26.0 Metrics
opentelemetry-api >=1.45.1 1.45.1 Span API (the SDK is the service's)
structlog >=26.1.0 26.1.0 Structured logging in the example worker and the default logger adapter

No OTel SDK, no web framework and no test tool: those live in the dev dependency group, so they never enter a consumer's environment. CI's lowest direct dependencies job runs every suite with each runtime dependency at its floor.

Integrating with a consumer service

Everything a consuming service does with eventcommon, in roughly the order it does it:

  1. Install.
  2. Wire metrics, the pool and the schema once at startup.
  3. Enqueue events inside business transactions.
  4. Run the outbox runner and the consumer as background tasks.
  5. Make handlers idempotent.
  6. Expose readiness and liveness.
  7. Drain at shutdown.

The two checklists are the short version; the subsections after them are the reference.

Installing

Published on PyPI. The source repository stays private, but the released package is public; VERSIONING.md § Publishing to PyPI explains why.

pip install eventcommon
# or with uv:
uv add eventcommon

Pin an exact version in production; VERSIONING.md § Consume a release has the recommended pin styles.

Import only the public modules (eventcommon.envelope, eventcommon.publisher, eventcommon.consumer, eventcommon.outbox, …). Never import eventcommon._internal: it is private by lint rule and not part of the supported API.

Quick start

import asyncio
from typing import Any

import pgcommon
import structlog

from eventcommon import inbox, metrics, outbox
from eventcommon.config import (
    load_outbox,
    load_sns,
    load_sqs,
    log_warnings,
    runner_config_from_env,
    sns_config_from_env,
    sqs_config_from_env,
    sqs_consumer_options,
)
from eventcommon.consumer import DLQConfig, SQSConsumer, SQSDLQPublisher
from eventcommon.envelope import Envelope
from eventcommon.publisher import SNSPublisher

logger = structlog.get_logger()


async def main() -> None:
    # 1. Tier 1 platform_* metrics, once — same registry and identity as pgcommon's.
    #    environment ← APP_ENV / ENVIRONMENT / "dev"; service ← APP_NAME.
    for warning in metrics.init_metrics(metrics.metrics_identity_from_env("iam")):
        logger.warning("metric_disabled", warning=str(warning))  # one metric off, not fatal

    # 2. Load and validate configuration (warnings: invalid values, defaults applied).
    sns_env, sqs_env, outbox_env = load_sns(), load_sqs(), load_outbox()
    log_warnings(sqs_env.warnings, logger=logger)
    log_warnings(outbox_env.warnings, logger=logger)
    for env in (sns_env, sqs_env, outbox_env):
        env.validate()  # ConfigValidationError when a required variable is missing

    # 3. Pool (DATABASE_URL or PG_*, via pgcommon.config_from_env) and schemas.
    pool = await pgcommon.Pool.create(outbox_env.pool_config())  # validated, non-optional
    await outbox.apply_schema(pool)
    await inbox.apply_schema(pool)

    # 4. Publisher + outbox runner (delivers enqueued events asynchronously). The
    #    publisher keeps one SNS client: close it at shutdown.
    publisher = SNSPublisher(sns_config_from_env(sns_env, logger=logger))
    runner = outbox.Runner(runner_config_from_env(outbox_env, pool, publisher, logger=logger))

    # 5. Consumer: inbox dedup + DLQ forwarding for poison messages.
    ledger = inbox.Store(pool, "iam-user-created")
    dlq = SQSDLQPublisher(DLQConfig(consumer_name="iam", logger=logger))

    async def handle(env: Envelope[dict[str, Any]]) -> None: ...  # idempotent business logic

    consumer = SQSConsumer(
        sqs_config_from_env(sqs_env, logger=logger),
        inbox.handler(ledger, handle),
        dlq_publisher=dlq,
        **sqs_consumer_options(sqs_env),
    )

    tasks = [asyncio.create_task(runner.start()), asyncio.create_task(consumer.start())]
    await asyncio.wait_for(runner.ready.wait(), timeout=30)  # gate the readiness probe
    try:
        # Returns when either stops; a crashed runner or consumer ends the service
        # (the orchestrator restarts it) instead of leaving it half alive.
        done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        for task in done:
            task.result()
    finally:
        # Stop both at once: their drains then fit one terminationGracePeriodSeconds.
        await asyncio.gather(consumer.stop(), runner.stop(), return_exceptions=True)
        await publisher.close()
        await dlq.close()
        await pool.drain_and_close()


asyncio.run(main())

examples/outbox_worker/main.py is the same wiring, production-shaped: startup retries, signals, a heartbeat for liveness, /metrics and bounded shutdown.

Integration checklist

Run through this before wiring the library into a new service. Each item links to the relevant section.

  • metrics.init_metrics: call once at startup, before building any publisher, consumer or runner, with the service's domain / service / environment; log the returned warnings (Observability)
  • Outbox for transactional events: every event tied to a database write goes through outbox.enqueue inside pgcommon.run_in_tx; no bare publisher.publish for them (Publishing rules)
  • Request context on every envelope: tenant_id=rc.tenant_id, trace_id=rc.trace_id from the HTTP request; system_tenant=True only for genuine cross-tenant jobs (Event envelope)
  • Idempotent handlers: inbox.handler(store, handle) or store.process(env, fn); external calls carry Envelope.id as their idempotency key (Implementing idempotency)
  • Error classification in handlers: raise for transient errors, return and log for permanent ones (Error classification)
  • Configuration from the environment: load_sns / load_sqs / load_outbox, log_warnings, .validate(); durations carry a unit (Configuration reference)
  • Dead-lettering: dlq_publisher (and/or a dead-letter handler) with max_receive_count, and the queue's maxReceiveCount higher than it (Dead-letter handling and DLQ forwarding)
  • Shutdown: consumer.stop(), runner.stop(), publisher.close(), dlq.close(), then pool.drain_and_close() (Health, readiness and shutdown)
  • Tests: unit tests use eventcommon.mock; nothing calls AWS (Testing in consuming services)

Production checklist

The deployment-time complement of the Integration checklist: what to have in place before a service takes production traffic.

  • RawMessageDelivery=true on every SNS → SQS subscription. Without it the consumer treats every message as malformed
  • AWS region and IAM: AWS_REGION set for the config loaders, or region= / AWS_DEFAULT_REGION for configs built in code (construction fails without one); the role may Publish / PublishBatch to the topic, ReceiveMessage / DeleteMessage / ChangeMessageVisibility / GetQueueAttributes on the queue, and GetQueueUrl / SendMessage on the DLQ (plus KMS for SSE-KMS queues — see the IAM table in Dead-letter handling and DLQ forwarding)
  • Visibility and timeouts: SQS_VISIBILITY_TIMEOUT ≥ 2× p99 handler duration; SQS_HANDLER_TIMEOUT set when handlers call slow dependencies; SQS_RETRY_BACKOFF set so a dependency outage does not dead-letter messages (Suggested defaults)
  • Schema as a deploy step: outbox.apply_schema / inbox.apply_schema once per deploy with a DDL-privileged role (MIGRATION_DATABASE_URL) (Migration safety)
  • Probes: readiness on runner.ready; liveness on runner.healthy(max_age) (or a heartbeat driven by it, as the example worker does), not on a database query (Health, readiness and shutdown)
  • Drain budget: SQS_DRAIN_TIMEOUT and OUTBOX_DRAIN_TIMEOUT (run concurrently) below terminationGracePeriodSeconds
  • Retention: prune_published and inbox.Store.prune scheduled; dead-letter retention decided (Pruning published records)
  • Alerts: the vendored monitoring/ rules loaded, routed to the owning team (Alerts, rules and dashboard)
  • HMAC only where the channel needs it (webhook / HTTP ingress), with keys ≥ 32 bytes from a secrets manager (HMAC helpers)

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 until published or dead-lettered
       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, and its claim lease expires. The next runner instance (or the restarted process) picks it up.

Detecting misuse in code review

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

# ✅ Correct — both writes in one transaction
async def save_and_announce(conn: Any) -> None:
    await repo.save_user(conn, user)
    await outbox.enqueue(conn, env)


await pool.run_in_tx(save_and_announce)


# ❌ Wrong — SNS call outside the transaction; event lost on crash
async def save(conn: Any) -> None:
    await repo.save_user(conn, user)


await pool.run_in_tx(save)
await publisher.publish(env)  # ← data inconsistency risk

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]. This document uses these terms consistently:

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",            // omitted when empty (new_envelope sets none by default)
  "tenant_id":      "acme",
  "trace_id":       "4bf92f3577b34da6a3ce929d0e0e4736",
  "correlation_id": "...",
  "subject":        "users/01926e4f-...",
  "actor":          "admin@acme.com",
  "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:00.1234Z",  // RFC 3339, Go RFC3339Nano form (trailing zeros trimmed)
  "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 handler's trace_id_from_context() and its log lines carry no trace ID, which breaks the log-to-trace pivot structured logging provides;
  • an incident that starts as an HTTP 500 in service A and shows up as a data anomaly in service B can no longer be correlated in Tempo or Grafana without manual log searching. (The sqs.receive span is still linked to the producer's span through the traceparent attribute when a publish ran inside an active span.)

For background jobs with no inbound HTTP trace, generate a trace ID at job startup and pass it to every event the job run emits, so at least the events of one run are grouped together.

System-level events (background jobs, scheduled tasks, cross-tenant operations) that are not scoped to any single tenant use SYSTEM_TENANT_ID (or new_envelope(..., system_tenant=True), Go's WithSystemTenant()):

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 it should be fixed, not papered over.

"system" does not disable RLS: the consumer sets app.tenant_id = 'system' like any other tenant, so a global-event handler needs an RLS-bypassing pool or role. Misuse on a tenant-scoped event runs the handler's queries under the wrong tenant, so RLS policies either reject them or return the wrong rows. 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.MetricsIdentity 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 the 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 build typed payloads with parse_envelope(data, payload_type=YourDataclass) or decode_payload(env.data, YourDataclass), never YourDataclass(**raw_dict). The former filter 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). new_envelope sets none, as Go's NewEnvelope does, so specversion is omitted from the wire and parses as ""; pass schema_version="1" to start versioning a type. Increment it only when both conditions hold: (1) a new optional field was added, and (2) at least one consumer needs to branch on whether that field is present. A consumer that receives an unrecognised specversion should log a warning and drop the message rather than misparse it. Treat an empty specversion as "1" in consumer logic, but never rewrite it on the envelope: the HMAC covers the field as sent.

Wire compatibility with Go. Fields are serialised in Go's order, and time uses Go's RFC3339Nano form (up to 9 fractional digits, trailing zeros trimmed, the original UTC offset kept).

  • What parsing keeps. It keeps the nanoseconds (Envelope.time_ns_remainder) and data's received bytes. While data is unchanged, to_json() emits those bytes (compacted and HTML-escaped, as Go's json.Marshal emits a json.RawMessage) and HMAC signs them, so a forwarded or re-signed envelope verifies in both languages (Envelope.data_json_bytes() returns them). Changing data (any value, key order, or the sign of a zero) re-encodes it.
  • What to_json() writes. What Go's json.Marshal of the envelope writes, byte for byte: compact, with &, <, > and U+2028 / U+2029 escaped in every string. The one exception is invalid UTF-8 in data, forwarded as U+FFFD.
  • How parsing reads. It follows Go's decoder. Keys match case-insensitively. A duplicate member resolves as encoding/json resolves it (a later null keeps the earlier value of a string or time field). Nesting is capped at 10,000 levels, and integers of any length are kept.
  • Very long integers. An integer longer than 4,300 digits (Python's int conversion limit) arrives as eventcommon.envelope.LongInt, a Decimal. Compare it, or convert it with int(value) (exact) or str(value). Arithmetic on it raises rather than rounding, and json.dumps cannot encode it until it is converted.

tests/unit/test_interop.py checks all of this against golden vectors generated by platform-events v2.1.0.

Serialisation

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

# Unmarshal and validate (Go: ParseEnvelope). Raises EnvelopeParseError (a ValueError) for
# invalid JSON, a non-object body, a field of the wrong type or a non-RFC 3339 time;
# EnvelopeIDRequiredError / EnvelopeTypeRequiredError / EnvelopeSourceRequiredError for a
# missing field; EnvelopeTimeRequiredError for a missing or zero time.
env = parse_envelope(data, payload_type=UserCreatedPayload)

The SQS consumer is deliberately less strict: it accepts an envelope without time (parse_envelope(..., require_time=False)), as Go's consumer does.

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. Use decode_payload(env.data, YourDataclass) (from eventcommon.envelope), never YourDataclass(**env.data). It drops fields your dataclass does not declare, so a producer's new optional field cannot break the consumer, and it raises TypeError for a payload that cannot be your type. Treat that as permanent: log and return, or dead-letter.

# ✅ 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 = decode_payload(env.data, UserCreatedPayload)
    # payload is now typed; env.id / env.tenant_id / env.trace_id are always available directly
    await repo.create_user(payload)

Routing handlers that inspect the event type and forward without parsing 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, and friendly to Postgres B-tree indexes. Use it as the idempotency key on the consumer side. eventcommon._internal generates UUID v7 itself, for portability down to Python 3.12; the stdlib uuid module only gained uuid7() 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 without affecting existing consumers. See ARCHITECTURE.md § Envelope compatibility guarantees for the per-field stability classes.

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=None,  # default: AWS_REGION, then AWS_DEFAULT_REGION / profile; none → ValueError
        endpoint_url=os.environ.get(
            "AWS_ENDPOINT_URL"
        ),  # set to http://localhost:4575 for the local floci (make docker-up)
        logger=logger,
    )
)

Region: SNSPublisher, SQSConsumer and SQSDLQPublisher resolve the region at construction: region=, else AWS_REGION, else boto3's chain (AWS_DEFAULT_REGION, the profile) — the Go SDK's order. (botocore alone does not read AWS_REGION; the library does.) When there is none they raise ValueError("…: no AWS region configured — set AWS_REGION (or the config's region)") instead of failing on the first call.

Deadline: SNSConfig(call_timeout=…) bounds each Publish / PublishBatch request, retries included (default 120 s; 0 disables). aiobotocore does not bound DNS resolution or the wait for a pooled connection. A timeout is a retryable error (per entry in publish_batch).

Client lifecycle:

  • One client per event loop. The publisher opens one SNS client on first use (one per event loop) and reuses it, so a publish does not pay a TCP + TLS handshake. Publishers may be used from event loops on several threads.
  • Close it on shutdown. Call await publisher.close(), or use async with publisher:, after the outbox runner has stopped, on the event loop that published. A client left open when its loop ends is reported with a warning. SQSDLQPublisher works the same way (close() / async with).
  • Injected clients. A client passed as client= (Go: NewWithClient) belongs to the caller: it is used as-is and never closed.
  • Credentials that were not ready. A client opened without credentials (NoCredentialsError / PartialCredentialsError, e.g. at pod start, before IRSA or the container-credentials endpoint is ready) is replaced, at most once per 5 s per event loop, so the publisher recovers without a restart. It is closed at once: those errors are raised while signing, so it never has a request in flight. The failure is retryable, so the outbox releases the record without counting an attempt.

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 into requests of at most 10 entries and 256 KiB, and attempts
# every request even when one fails. Any failed entry raises one BatchError.
await publisher.publish_batch(envelopes)

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

Callers must inspect the error to tell a partial failure from a total one:

from eventcommon.errors import BatchError

try:
    await publisher.publish_batch(envelopes)
except BatchError as exc:
    # str(exc): "events: N message(s) failed in batch (first: <id> <code>: <message>)".
    # exc.failures is list[BatchFailure(id, code, message, retryable)] — id is the
    # envelope ID. Successfully published messages must NOT be retried.
    by_id = {env.id: env for env in envelopes}
    retry = [by_id[f.id] for f in exc.failures if f.is_transient]
    for f in exc.failures:
        logger.warning("batch publish failure", event_id=f.id, code=f.code, error=f.message)
    # Retry transient failures, or hand off to the outbox for durable retry.
    await publisher.publish_batch(retry)

Failure codes (Go parity) are InvalidEnvelope, CodecEncodeError, MarshalError, TooManyAttributes, TransportError (the whole request failed, retryable), PublishError, or the AWS error code SNS returned. retryable is true for throttling / 5xx codes, or when SNS reports no sender fault; is_transient also covers TransportError.

Key rules:

Rule Reason
Never retry the full batch on a BatchError Messages that succeeded would be published a second time, creating duplicates
Always map failures by BatchFailure.id The envelope ID is the 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, which requires content-based deduplication to be disabled on the topic.

⚠️ 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 (per tenant, per aggregate), and Envelope.id with 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. SQS subscription filter policies can then scope queues to event types, tenants or subjects without deserialising the body. Actor is an audit-trail field and is not forwarded as an attribute.

  • Static attributes. SNSPublisher(attributes={...}) adds them. A key that collides with one of the five routing attributes is ignored with a warning, and 5 or more extra attributes log a warning at construction.
  • Trace context. The OTel propagator's traceparent, tracestate and baggage are injected as attributes, so the consumer can link its span to the producer's.
  • The 10-attribute limit. SNS allows 10 attributes. baggage, then tracestate, are dropped first (logged once per publisher); if the message still does not fit, publish raises before calling SNS.

Schema registry codec

eventcommon adds no registry SDK dependency. For consumers of Glue-framed payloads, it ships the decode-only eventcommon.codec.GlueDecodeCodec (Go: events.GlueDecodeCodec). It strips Glue's 18-byte header and returns the JSON payload, with no registry client or AWS permission needed, and it rejects compressed and non-JSON (e.g. Avro) payloads. To encode, implement eventcommon.codec.Codec against your own registry client and pass it to SNSPublisher / SQSConsumer:

from eventcommon.codec import Codec


class GlueCodec:
    async def encode(
        self, event_type: str, payload: bytes
    ) -> tuple[bytes, str]: ...  # returns (registry-encoded bytes, schema ID)

    async def decode(
        self, schema_id: str, encoded: bytes
    ) -> bytes: ...  # JSON → parsed data; else raw bytes


publisher = SNSPublisher(SNSConfig(topic_arn=...), codec=GlueCodec())
consumer = SQSConsumer(SQSConfig(queue_url=...), handle_event, codec=GlueCodec())
  • No codec (the default, None). data stays plain JSON and schema_id stays empty.
  • Publishing with a codec. Publisher.publish base64-wraps the codec's encoded bytes into data and stamps schema_id. An encode failure raises CodecEncodeError, wrapped in RetryableError when the codec's error was retryable.
  • Consuming with a codec. SQSConsumer decodes a message whose schema_id is non-empty and whose data is a JSON string (the codec wire format); a dataschema on plain-JSON data is passed through undecoded. A decode failure is logged and the message is left visible for redelivery; the handler is never called with an undecoded payload.

See codec.py's docstrings for the full contract, and eventcommon.codec.NoopCodec for a reference identity implementation.

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=60,  # seconds; 0 (default) = the queue's timeout, no extension
    # Optional (Go option in brackets; all off by default):
    handler_timeout=300,  # [WithHandlerTimeout] bounds decode + DLH + handler
    retry_backoff=60,  # [WithRetryBackoff] failed delivery n hidden 60s·2^(n-1)…
    max_retry_backoff=900,  # …capped at 15m (+ up to 10% jitter)
    dlq_publisher=SQSDLQPublisher(DLQConfig(consumer_name="billing")),  # [WithDLQForwarding]
    max_receive_count=5,  # with a DLQ publisher / dead-letter handler (default 5)
    queue_depth_interval=60,  # [WithQueueDepthMetrics] platform_queue_depth / _dlq_depth
    # To load from env: **sqs_consumer_options(load_sqs())
)

task = asyncio.create_task(consumer.start())  # runs until stop() is called
...
await consumer.stop()  # drain — waits up to drain_timeout (30s), then cancels handlers

Behaviour worth knowing (all Go v2.1.0 parity):

  • Concurrency is enforced at dispatch: a received message waits for a free worker, and its visibility is extended from receipt while it waits. With visibility_timeout=0, a receive asks only for as many messages as there are free workers.
  • FIFO queues (URL ending in .fifo) are processed per MessageGroupId, in receive order, one message per group at a time. A message that is not settled hands the group's later messages of that batch back unprocessed, so order holds. Groups run in parallel up to concurrency, and each receive carries a fresh ReceiveRequestAttemptId.
  • Retry backoff applies to every unsettled message: a handler, decode or dead-letter-handler error, a handler_panic, or a failed DLQ forward. The message is hidden for retry_backoff·2^(n-1) after its n-th delivery, through ChangeMessageVisibility. Without it, a failed message reappears after the visibility timeout, so a dependency outage longer than about maxReceiveCount × VisibilityTimeout dead-letters messages that would have succeeded.
  • Handler timeout cancels the handler's task, counts platform_message_timeouts_total{operation} and stops extending visibility, so the message is redelivered. A FIFO group's waiting messages share the running message's budget.
  • Timeouts cancel; the handler decides (Go: the deadline or drain timeout cancels the handler's ctx). At handler_timeout or drain_timeout, the handler task is cancelled, which interrupts the await it is in, as Go's ctx-aware I/O returns ctx.Err(). The outcome is then the handler's:
    • If it raises, the message is a handler_error (plus platform_message_timeouts_total past the deadline) and is retried.
    • If it catches CancelledError (Python's way of ignoring the cancellation) and returns normally, the message is deleted and counted processed, like a Go handler returning nil after its deadline.
    • The same holds for codec decode and the dead-letter handler. After the drain timeout start() returns, but the SQS client stays open until such a handler finishes, so its late success still deletes the message.
  • Lifecycle.
    • start() is a plain method that returns the coroutine (Go: Start(ctx)). A stop() that runs after create_task(consumer.start()) but before the task first runs is therefore not lost: that start returns at once. A start() that was cancelled or never ran leaves no trace.
    • start() raises RuntimeError if the consumer is already running, and can be called again after stop(). It opens its own SQS client (unless one was injected with client=) and closes it when it returns, or, after a drain timeout, once a handler that ignored the cancellation has finished.
    • stop() is safe before start() and more than once. It waits until start() has returned, and raises TimeoutError if that takes longer than drain_timeout + 5 s.
    • Called from a handler (or the start() task), stop() only signals, because waiting there would deadlock; start() returns once that handler has finished. Any other task waits, including one a handler spawned (e.g. a shutdown coroutine that then closes the DB pool), as Go's Stop does, so a handler must not await such a task.
    • Cancelling the start() task drains, then raises CancelledError as usual.
  • Handler typing. Where a handler is accepted (SQSConsumer, dead_letter_handler, with_dead_letter_handler, MockConsumer.set_handler, inbox.handler), it is typed AcceptedHandler: an Envelope of what the consumer delivers, which is parsed JSON (Envelope[dict[str, Any]], Envelope[Any]) or bytes from a non-JSON codec. A handler annotated Envelope[YourModel] is a mypy error there, because env.data is never YourModel at runtime: parse it in the handler (parse_envelope(…, payload_type=YourModel) is typed Envelope[YourModel]). Annotate middleware and route tables with Handler (Callable[[Envelope[Any]], Awaitable[None]]), which can be called with any envelope.
  • Synchronous handlers (def, not async def) are supported but block the event loop while they run; the consumer logs sqs: handler is synchronous — it blocks the event loop; use async def once. Their return value is final, exactly as for an async def handler: a plain return deletes the message and a raise retries it. The same applies to a synchronous dead-letter handler.
  • Tracing: the sqs.receive span (tracer platform-events, CONSUMER kind) starts a new trace linked to the producer's span through the message's traceparent. It is rooted in an empty context, never in the span that was current when start() was called, and it carries the producer's baggage into the handler.
  • Context: inside a handler, eventcommon.context.source_message_from_context() returns the raw SQS message (queue URL, message ID, body, attributes, receive count), and trace_id_from_context() returns the envelope's trace ID.

Handler contract

Handler outcome Consumer behaviour
Returns normally SQS message deleted from the queue
Raises an exception, or times out SQS message left visible; retried after retry_backoff, or the visibility timeout
Raises a "panic" (a BaseException that is not an Exception, other than CancelledError, KeyboardInterrupt and SystemExit) Recovered, logged with its stack (sqs: handler panic recovered), counted as reason="handler_panic"; retried
Malformed body (not an envelope: invalid JSON, missing id / type / source, …) Forwarded to the DLQ (with dlq_publisher) or deleted immediately; counted as platform_messages_failed_total{reason="malformed"} — never retried

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 concurrency. On stop(), the consumer stops receiving new messages and awaits in-flight handler tasks for up to drain_timeout, then cancels them. Use handler_timeout to bound a single handler call.

Error classification

The handler's outcome (return normally, or raise) is the only signal the consumer uses to decide between retrying and deleting. Getting this backwards decides whether a broken message sits in SQS forever or is 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 platform_messages_failed_total{reason="handler_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 makes the handler raise, SQS retries it MaxReceiveCount times and then pushes it to the DLQ, where it sits indefinitely. Return normally (and log) for any message that cannot succeed on retry, however many times it is delivered.

Classification pattern:

async def handle_user_created(env: Envelope[dict[str, Any]]) -> None:
    try:
        payload = decode_payload(env.data, UserCreatedPayload)
    except TypeError as exc:
        # Permanent: a payload that cannot be this type will never parse on retry.
        # (decode_payload drops fields the dataclass does not declare, so a producer's
        # new optional field is not an error.)
        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

Dead-letter handling and DLQ forwarding

Messages that can never succeed reach dead-letter storage three ways:

Path Trigger What happens
max_receive_count ApproximateReceiveCount > max_receive_count, with a dead_letter_handler and/or dlq_publisher set The dead-letter handler runs (it receives the parsed, decoded Envelope; the raw message is in source_message_from_context()). Then the message is forwarded to the DLQ (DLQReason receive count N exceeded consumer max receive count M) and deleted
Malformed body / decode failure With dlq_publisher, a malformed body is forwarded at once, and a decode failure past the threshold is forwarded Without a DLQ publisher, a malformed body is logged (size + SHA-256; malformed_body_logging=True adds a 512-byte excerpt) and deleted
Explicit The handler calls await forward_source_message(dlq, reason) and returns normally Counted as platform_dlq_messages_total{reason="explicit"}, not as processed; deleted even if the handler raises after the forward; kept for retry if it returns after a failed forward

Handler-initiated dead-letters. forward_source_message then a normal return is the supported route for a permanent failure (e.g. a downstream 400 / 413 / 422); there is no non-retryable handler exception. Forward from inside the handler, so its context carries the dead-letter attribution the consumer settles on:

After the handler Consumer Metrics (one per message)
Forward succeeded, handler returned Deletes the original platform_dlq_messages_total{operation="consume",reason="explicit"} +1; not processed, not failed
Forward succeeded, handler then raised (or panicked) Deletes the original (WARNING sqs: handler failed after forwarding the message to the DLQ — deleting it, not retrying) — retrying would forward a second copy Same as above
Forward failed (and no later one succeeded), handler returned anyway Leaves it visible for retry (retry_backoff applies) — deleting it would lose it platform_messages_failed_total{reason="dead_letter_error"} + platform_retry_total
Forward failed, handler re-raised the DLQError Leaves it visible for retry failed_total{reason="handler_error"} + retry_total
Delete fails after a successful forward (any path) Retries the delete 3 times, then remembers the message ID for 12 h: this replica deletes the redelivery without dispatching or counting it Nothing more

A redelivery received by another replica runs the handler again; DLQ consumers dedupe on the envelope ID. Your own reason goes in the DLQReason attribute, never a metric label. max_receive_count is read only with dead_letter_handler / dlq_publisher: a handler that dead-letters on its own retry limit (source_message_from_context().receive_count) must keep that limit below the queue's RedrivePolicy maxReceiveCount (e.g. 4 below 5), or SQS moves the message first, without a DLQReason.

SQSDLQPublisher (Go: NewSQSDLQPublisher) works like this:

  • Which DLQ. It resolves the DLQ from the source queue's RedrivePolicy, cached for DLQConfig.cache_ttl (default 15m). Concurrent forwards share one lookup per queue, bounded by 30 s. With forwarding enabled, start() fails when the queue has none.
  • What it sends. Each forwarded message keeps the original body and String/Number attributes, and gains EventType, DLQReason, OriginalQueue, FailedAt and (when configured) ConsumerName. Lower-priority caller attributes are dropped to fit SQS's 10-attribute limit (strict_attributes=True raises instead).
  • FIFO DLQs. MessageGroupId is the envelope ID, and MessageDeduplicationId is unique per forward.
  • Failures. They raise a DLQError subclass: DLQNotConfiguredError, DLQInvalidRedrivePolicyError, DLQUnresolvedError, DLQSendFailedError or DLQInvalidMessageError, with .retryable set for throttling and 5xx. The original is deleted only after the forward succeeded.

IAM — the complete set for a consumer with DLQ forwarding (automatic or from a handler):

Resource Actions Used by
Source queue sqs:ReceiveMessage, sqs:DeleteMessage, sqs:ChangeMessageVisibility Receive loop, visibility extension, hand-back, retry_backoff
Source queue sqs:GetQueueAttributes RedrivePolicy lookup (resolve_dlq, start()); queue_depth_interval
DLQ sqs:GetQueueUrl, sqs:SendMessage send_to_dlq
DLQ sqs:GetQueueAttributes Only with queue_depth_interval (platform_dlq_depth)
Source queue's customer-managed KMS key kms:Decrypt ReceiveMessage of an SSE-KMS queue
DLQ's customer-managed KMS key kms:GenerateDataKey, kms:Decrypt SendMessage to an SSE-KMS queue

SSE-SQS and the AWS-managed alias/aws/sqs key need no KMS statement in the role. For a customer-managed key the key policy must also allow the role, and the source queue's key must allow sns.amazonaws.com (kms:GenerateDataKey, kms:Decrypt) so SNS can deliver — infrastructure, not this library. The DLQ is in the source queue's account and region (an SQS RedrivePolicy rule).

A dead-letter handler that raises (or panics) leaves the message visible, and it is not forwarded.

Breaking change in 0.3.0: dead_letter_handler used to receive the raw SQS message dict before parsing; it now receives the Envelope, as Go's does.

Poison messages

A poison message is structurally valid JSON (it parses without error) but cannot be processed successfully however many times it is retried. It is distinct from a transient error (which may succeed on retry) and from a malformed message (which fails 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 = decode_payload(env.data, OrderShippedPayload)
    except TypeError:
        return  # cannot be an OrderShippedPayload — 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: dead-letter it for later investigation instead of dropping it.
        await forward_source_message(dlq, "illegal state transition: order cancelled")
        return

    await repo.mark_shipped(order.id)

Decision rules:

  1. Return normally; don't raise. Raising retries the message, and a semantically invalid message never becomes 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, and the log is the only record that something was discarded.
  3. Optionally dead-letter it with forward_source_message(dlq, reason) when the business impact warrants investigation or replay.
  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, because of 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 0 — eventcommon.inbox (recommended)

The inbox (Go: pkg/inbox) records each handled envelope ID per consumer in processed_events:

from eventcommon import inbox

await inbox.apply_schema(pool)  # own tracking table: inbox_migrations
store = inbox.Store(pool, "billing-invoice-settled")  # stable consumer name

# Either wrap the handler — a redelivery is acknowledged without calling it
# (counted in platform_duplicate_messages_total):
consumer = SQSConsumer(config, inbox.handler(store, handle_invoice_settled))


# …or, when the side effects are Postgres writes, claim the ID in the same transaction
# (exactly-once writes; concurrent copies serialise on the claim):
async def handle_invoice_settled(env: Envelope[dict[str, Any]]) -> None:
    async def write(conn: Any) -> None:
        await repo.mark_settled(conn, env.data["invoice_id"])

    await store.process(env, write)


# Periodically: keep at least the queue's retention plus any replay window.
await store.prune(timedelta(days=7))

The envelope ID must be a UUID: otherwise InvalidEventIDError is raised, so the message reaches the DLQ instead of being processed unguarded.

These are retried, because the transaction is aborted and its writes and the claim roll back:

  • a write that ends its own transaction (TxEndedInCallbackError), including one that ends it and begins another (ROLLBACK; BEGIN);
  • a write that catches a failed statement and returns normally (TxCommitRollbackError). On pgcommon 0.2, that COMMIT would have reported success and the message would have been deleted unprocessed.

To tolerate an expected failure (e.g. a unique violation), run that statement in a savepoint (conn.transaction()). A message the handler dead-lettered is not recorded, so redriving the DLQ after a fix processes it.

Pattern 1 — Postgres unique constraint (hand-rolled)

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 def tx(conn: Any) -> None:
        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 = decode_payload(env.data, UserCreatedPayload)
        await repo.create_user(conn, payload)

    await pool.run_in_tx(tx)

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, which is correct. 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

When the side effect is a row update rather than an insert, record the last applied event on the row itself and make the update conditional on it:

UPDATE invoices
   SET status = 'settled', last_event_id = $2
 WHERE id = $1 AND last_event_id IS DISTINCT FROM $2;
-- 0 rows: this event was already applied (or the invoice does not exist — check separately)

This only works for updates that are safe to apply once per event regardless of order. For anything else, use Pattern 0 or 1.

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 0 or 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 0 or 1
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 inbox.Store.prune(retention) on a schedule

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 the inbox claim (store.process) or 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, at least once. The outbox does not guarantee global ordering; use enqueue_ordered, and a FIFO topic with a stable message_group_id, when ordering is required.

Wiring

from eventcommon.outbox import Runner, RunnerConfig, apply_schema

# 1. Apply outbox schema migrations 001-011 — platform-events' files and tracking, so Go
#    and Python services can share the schema. See Migration safety below.
await apply_schema(pool)  # or apply_schema(pool, lock_timeout=60.0); eventcommon.inbox too

# 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; negative disables it
        retry_backoff=1.0,  # a record's n-th failed publish waits retry_backoff·2^(n-1)…
        max_retry_backoff=300.0,  # …capped at 5m (equal jitter)
        gauge_interval=15.0,  # backlog gauges + ordered-record promotion sweep
        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)

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 · retry_backoff 1s · max_retry_backoff 5m · gauge_interval 15s.

claim_lease_duration must be at least batch_size × publish_timeout + 60s (30s when the timeout is disabled), else RunnerConfig raises ValueError. The runner re-polls immediately while batches keep publishing (for at most poll_interval). On shutdown, claimed records that were not yet sent are released at once instead of staying hidden for the lease.

A stalled database (a hung primary, a half-open connection after a failover) makes each poll fail within the store's 5 s query timeout. The failure is counted in platform_outbox_errors_total{operation="poll"} and logged, and the runner recovers by itself. This relies on pgcommon ≥ 0.3, whose timed-out queries clean up without waiting on the frozen server. Drive a liveness probe from runner.healthy(...), not from whether the task is alive; see Health, readiness and shutdown.

Enqueueing inside a transaction

async def tx(conn: Any) -> None:  # pgcommon runs it in one transaction (Go: RunInTx)
    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)


await pool.run_in_tx(tx)  # or pgcommon.run_in_tx(pool, tx)

enqueue requires an open transaction and validates the envelope as Go does:

  • non-empty id / type / source, a non-zero time, and a canonical lowercase UUID id;
  • no NUL character in id / type / source / tenant_id / trace_id or anywhere in the JSON. Postgres jsonb cannot store \u0000, so the INSERT would roll back the business transaction on every retry; strip NULs from user input first;
  • a serialised size of at most 240 KB, under the SNS 256 KB hard limit, so an outbox record that could never publish is never persisted.

Ordered outbox

async def tx(conn: Any) -> None:
    await conn.execute(
        "SELECT 1 FROM orders WHERE id = $1 FOR UPDATE", order.id
    )  # serialise per key
    await outbox.enqueue_ordered(conn, env, f"order/{order.id}")


await pool.run_in_tx(tx)

Records with the same ordering key (≤ 256 bytes) are published one at a time, in enqueue order, across all runner replicas.

  • A record enqueued while an earlier one of its key is unpublished waits (scheduled_at = 'infinity'), and is promoted when that one is published or dead-lettered.
  • A failing head record holds its key until it is dead-lettered; platform_outbox_ordering_blocked_events counts the waiting records.
  • A replayed dead letter joins the back of its key.

On a FIFO topic, derive message_group_id from the same key. Requires migration 0010.

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. A row written around enqueue that SNS rejects is dead-lettered after max_attempts; one over 1 MiB as stored is not even loaded by the runner (a memory guard)
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 side-by-side timeline and the 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.

Transient failures never count toward max_attempts. They are throttling and 5xx codes, network errors, timeouts, and BatchFailures that are retryable or TransportError. The record's lease is released with a shared backoff that advances at most once per poll and resets on the next successful publish, so a period of SNS unavailability does not dead-letter healthy records. Permanent failures count an attempt and wait retry_backoff·2^(n-1) (capped at max_retry_backoff). A publisher "panic" (a BaseException that is not an Exception) is recovered, logged with its stack, and counts an attempt, so a poison batch cannot hold its lease.

Replay inserts into outbox_events first and deletes only the dead letters it inserted. A dead letter whose ID is already back in outbox_events is left in place (logged). List, replay and discard order by (failed_at, id), and limit must be positive.

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 every consumer has processed the event before the row is deleted; 7 days covers most SLA windows. outbox_dead_letters has no automatic TTL; schedule your own cleanup (see ARCHITECTURE.md § Dead-letter table retention).

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, or of their ordering key;
  • the same Envelope.id is presented again, so idempotency guards must be in place;
  • 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 holding the shared key, and was not tampered with in transit. It does not provide confidentiality or replay protection: a captured signed message verifies again. Where replay matters (e.g. webhook ingress), the receiver must reject it: check that time is recent (allowing for clock skew), and that Envelope.id has not been seen within that window (the inbox, or any ID store).

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 it with string ==. Key length: sign raises KeyTooShortError for keys under 32 bytes.

Rules:

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

Health, readiness and shutdown

Signal Use How
Readiness Take traffic only once the database is reachable and the schema exists await asyncio.wait_for(runner.ready.wait(), timeout=…) — ready is set after the first successful poll
Liveness Restart a worker whose database work has stalled runner.healthy(max_age), e.g. 3 * poll_interval + 30: True while a publish bounded by publish_timeout is in flight, or a store round trip succeeded within max_age. It tracks the database, not SNS speed (Python-only; Go has Ready() only). runner.last_poll_succeeded is the monotonic time behind it
Consumer health Detect a stuck or saturated consumer platform_messages_in_flight against concurrency, platform_queue_depth (with queue_depth_interval), and the start() task: it only returns on stop() or a fatal error
Shutdown Fit inside terminationGracePeriodSeconds await asyncio.gather(consumer.stop(), runner.stop()), then publisher.close(), dlq.close(), pool.drain_and_close()

The example worker drives a heartbeat file from runner.healthy(...) (the image's HEALTHCHECK reads it), retries the first database connection until STARTUP_TIMEOUT, and bounds the whole SIGTERM shutdown by SHUTDOWN_TIMEOUT. Use it as the template for a service's entrypoint.

Failure model

Failure Behaviour
Invalid configuration (topic ARN, queue URL, region, RunnerConfig, metric identity) Raised at construction; the process should not start
A queue with DLQ forwarding but no usable RedrivePolicy consumer.start() raises DLQNotConfiguredError / DLQInvalidRedrivePolicyError
SNS throttling, 5xx, network or DNS failure, call_timeout RetryableError; the outbox releases the record without counting an attempt (shared backoff)
Permanent SNS error (authorization, not found, invalid parameter) The outbox counts an attempt and backs off; dead letter at max_attempts
Credentials not yet available (pod start) The client is replaced (≤ 1 per 5 s per loop); the error is retryable; recovers without a restart
SQS receive failing Logged; backoff 1 s → 30 s with jitter; the loop continues
Malformed body DLQ (with dlq_publisher) or deleted on first receipt; never retried; logged as size + SHA-256
Codec decode failure decode_error; left visible and retried; DLQ past the threshold
Handler raises / times out / panics Left visible; retry_backoff when set; dead-letter handler + DLQ past max_receive_count
Dead-letter handler raises The message stays visible; it is not forwarded
DLQ lookup or send failure The original stays on the source queue (retryable for throttling, 5xx and timeouts)
Postgres unreachable or stalled Each poll fails within 5 s, counted; backoff 1 s → 30 s; healthy(max_age) turns False; recovers
mark_published fails after SNS accepted the event The record is published again after its lease: a duplicate, absorbed by idempotent consumers
Process killed mid-batch Claimed records reappear after claim_lease_duration; in-flight SQS messages after their visibility timeout
stop() while handlers run Waits drain_timeout, then cancels; cancelled messages are redelivered without delay
A payload over 1 MiB in outbox_events (written around enqueue) Not loaded; counts an attempt; dead-lettered at max_attempts
A metric name already registered with another shape That metric is disabled; a RegistrationWarning is returned; everything else works

Errors are never swallowed: every path either propagates the original exception or chains it (raise … from exc). ARCHITECTURE.md has the full degradation matrix.

Backpressure considerations

Under sustained load, the SQS backlog and the outbox_events table both grow if consumers or the outbox runner cannot keep up.

Consumer-side (SQS)

Tuning order:

  1. Raise concurrency; it 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.

monitoring/kubernetes/keda-scaledobject.example.yaml scales a consumer on platform_queue_depth.

Outbox-side (Postgres → SNS)

Tuning order:

  1. Decrease OUTBOX_POLL_INTERVAL.
  2. Increase OUTBOX_BATCH_SIZE, keeping it below 100 to avoid long-held Postgres locks.
  3. Scale runner pods horizontally; SKIP LOCKED distributes the work with no coordination.
  4. Check SNS publish errors: a rising platform_outbox_publish_attempts_total{outcome="error"} rate usually means 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 or receive size.

Performance considerations

Overhead

eventcommon is a thin layer over aioboto3 and pgcommon. Network latency to SNS, SQS and Postgres dominates every path:

Mechanism When it runs Approximate cost Notes
new_envelope Every event Microseconds UUID v7, datetime.now(UTC), object construction
outbox.enqueue Every transactional event One INSERT in the caller's transaction Plus serialisation and validation; no AWS call
Runner poll, empty Every poll_interval One claim query Gauge queries every gauge_interval (15 s), capped at MAX_COUNTED_ROWS
Runner poll, full batch While there is a backlog ceil(batch_size / 10) PublishBatch calls (publish_concurrency=1), or one publish per record Successes settled with one mark_published_many write
Consumer, per message Every delivery One delete, plus one visibility extension every max(visibility_timeout/2, 1 s) while it runs Dispatch is a semaphore acquire and a task
Metrics, spans Every operation Microseconds, in-process Export is the OTel SDK's, off the path
verify (HMAC) Per call Microseconds Canonical JSON + HMAC-SHA256 + constant-time compare

See ARCHITECTURE.md § Performance characteristics.

Tuning

  • Keep handlers short and async. A worker slot is held for the whole handler; a synchronous or CPU-bound handler blocks every other message on the loop. Use asyncio.to_thread for blocking work.
  • Keep external I/O out of the outbox transaction. enqueue adds one insert; anything slow you add to the business transaction holds its locks longer.
  • Use publish_concurrency=1 (batch publishing) unless SNS latency is the bottleneck. It needs a tenth of the API calls.
  • Raise concurrency before max_messages. More workers use the long poll better than larger receives.

Common pitfalls

❌ Publishing directly after a database write

If the process dies after the commit but before publish returns, the event is silently dropped: no retry, no recovery.

# ❌ WRONG: the event is lost if the process dies between the commit and the publish
await pool.run_in_tx(save_order)
await publisher.publish(order_placed)


# ✅ CORRECT: the event commits with the order, or neither does
async def tx(conn):
    await save_order(conn)
    await outbox.enqueue(conn, order_placed)


await pool.run_in_tx(tx)

❌ Ignoring Envelope.id in the handler

The outbox and SQS deliver at least once; without an idempotency check, a redelivered event causes duplicate side effects.

# ❌ WRONG: a redelivery charges the customer twice
consumer = SQSConsumer(config, charge_customer)

# ✅ CORRECT: redeliveries of the same Envelope.id are acknowledged without running it again
consumer = SQSConsumer(config, inbox.handler(inbox.Store(pool, "billing-charge"), charge_customer))

❌ Raising for a message that can never succeed

Raising retries the message max_receive_count times and then dead-letters it. A validation failure never fixes itself.

# ❌ WRONG: a permanently invalid payload is retried, then sits in the DLQ
payload = decode_payload(env.data, Order)  # TypeError propagates

# ✅ CORRECT: log and return, or dead-letter it explicitly
try:
    payload = decode_payload(env.data, Order)
except TypeError:
    await forward_source_message(dlq, "payload is not an Order")
    return

❌ Branching on env.data without checking env.type

A handler subscribed to several event types silently misparses the wrong payload. Check env.type first, and route to typed handlers by event type.

❌ A non-stable FIFO MessageGroupId

A random group ID per message puts every message in its own group, so SNS delivers them concurrently and ordering is lost. Use a deterministic, stable ID (tenant_id, user_id, an aggregate ID): a value that must be ordered relative to itself.

❌ SYSTEM_TENANT_ID in an HTTP handler

The consumer's queries then run as app.tenant_id = 'system', which RLS policies reject or answer with the wrong rows, silently. Pass tenant_id=rc.tenant_id; use SYSTEM_TENANT_ID only for genuine cross-tenant jobs.

❌ Strict payload schemas

Parsing payloads with a schema that rejects unknown fields (extra="forbid" in Pydantic, a strict model) turns every non-breaking producer addition into a consumer runtime error. Parse permissively (extra="ignore", decode_payload).

❌ enqueue outside a transaction

The event is then not durably linked to the domain write. enqueue raises on a connection with no open transaction; always call it inside pgcommon.run_in_tx.

❌ Swallowing asyncio.CancelledError in a handler

A handler that catches the cancellation keeps running past the drain deadline and the handler timeout. Let it propagate (clean up in finally).

❌ Bare numbers in duration variables

SQS_VISIBILITY_TIMEOUT=60 is rejected with a warning and the default (30 s) is used, exactly as Go does. Write 60s, and log .warnings at startup so the mistake is visible.

❌ Forgetting RawMessageDelivery=true

Without it SNS wraps the envelope in its own JSON notification, and the consumer treats every message as malformed and dead-letters it.

Configuration reference

The components take dataclass configs plus keyword options. Every field but topic_arn / queue_url / pool + publisher is optional. Each can also come from the environment through eventcommon.config (see Environment variables).

Field Env var Default Notes
SNSConfig.topic_arn SNS_TOPIC_ARN — Required
SNSConfig.region / SQSConfig.region / DLQConfig.region AWS_REGION None: AWS_REGION, then boto3's chain (AWS_DEFAULT_REGION, profile) No region → ValueError at construction
SNSConfig.endpoint_url (and SQS / DLQ) AWS_ENDPOINT_URL None http://localhost:4575 for the local floci
SNSConfig.call_timeout — 120 s Bounds each SNS request, retries included; 0 disables (Python-only)
SQSConfig.queue_url SQS_QUEUE_URL — Required
SQSConfig.max_messages / wait_seconds SQS_MAX_MESSAGES / SQS_WAIT_SECONDS 10 / 20 1–10 / the long poll
concurrency SQS_CONCURRENCY 1 Parallel handlers (FIFO: message groups)
visibility_timeout SQS_VISIBILITY_TIMEOUT 0 (env loader: 30s) 0: the queue's timeout, nothing extended; ≤ 12 h
handler_timeout SQS_HANDLER_TIMEOUT 0 (off) Bounds decode + dead-letter handler + handler
retry_backoff / max_retry_backoff SQS_RETRY_BACKOFF / SQS_MAX_RETRY_BACKOFF off / 15m Failed delivery n hidden base·2^(n-1), +≤10% jitter
max_receive_count SQS_MAX_RECEIVE_COUNT 0 (5 with a DLQ publisher or dead-letter handler) Above it: dead-letter handler + DLQ
dead_letter_handler / dlq_publisher — None See Dead-letter handling
drain_timeout SQS_DRAIN_TIMEOUT 30 s stop() wait before cancelling handlers
queue_depth_interval SQS_QUEUE_DEPTH_INTERVAL 0 (off) Samples platform_queue_depth / platform_dlq_depth; minimum 10 s
malformed_body_logging — False Adds a 512-byte excerpt of malformed bodies to their log line
codec — None See Schema registry codec
DLQConfig.consumer_name / cache_ttl / strict_attributes — "" / 0 (15m) / False ConsumerName attribute · DLQ lookup cache (negative: forever) · raise instead of dropping attributes
RunnerConfig.poll_interval OUTBOX_POLL_INTERVAL 5s
RunnerConfig.batch_size / max_attempts OUTBOX_BATCH_SIZE / OUTBOX_MAX_ATTEMPTS 50 / 5
RunnerConfig.claim_lease_duration OUTBOX_CLAIM_LEASE_DURATION 10m ≥ batch_size × publish_timeout + 60s
RunnerConfig.publish_concurrency / publish_timeout OUTBOX_PUBLISH_CONCURRENCY / OUTBOX_PUBLISH_TIMEOUT 1 / 10s 1 uses PublishBatch; negative timeout disables it
RunnerConfig.retry_backoff / max_retry_backoff OUTBOX_RETRY_BACKOFF / OUTBOX_MAX_RETRY_BACKOFF 1s / 5m Equal jitter
RunnerConfig.drain_timeout / startup_jitter / gauge_interval OUTBOX_DRAIN_TIMEOUT / OUTBOX_STARTUP_JITTER / OUTBOX_GAUGE_INTERVAL 30s / 0 / 15s
RunnerConfig.logger (and every logger) — None structlog or a stdlib logging.Logger; None logs nothing

Suggested defaults (production)

# SQS Consumer
SQS_MAX_MESSAGES=10          # Always set to max (SQS hard limit)
SQS_WAIT_SECONDS=20          # Long-poll at max
SQS_VISIBILITY_TIMEOUT=60s   # comfortably above your p99 handler latency; 30s default is tight
SQS_HANDLER_TIMEOUT=5m       # release hung handlers for redelivery
SQS_RETRY_BACKOFF=60s        # ride out dependency outages instead of dead-lettering
SQS_CONCURRENCY=5            # Start here; scale up to ~10 before considering horizontal pod scaling

# Outbox runner
OUTBOX_POLL_INTERVAL=2s      # 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
OUTBOX_STARTUP_JITTER=5s     # desync replicas on a rolling restart

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, and the message is delivered twice. Set it to 2–3× your p99 handler duration.

What to monitor on day one:

Metric Alert threshold Action
platform_outbox_oldest_pending_age > 5 min Delivery is stalled: check SNS reachability and the runner
platform_outbox_pending_events > 500 sustained for 5 min Lower OUTBOX_POLL_INTERVAL; add runner pods
platform_queue_depth (or CloudWatch ApproximateNumberOfMessages) > 1000 sustained Raise SQS_CONCURRENCY; add consumer pods
platform_messages_failed_total / platform_messages_received_total > 1% error rate Inspect handler errors; check DLQ depth
platform_dlq_messages_total rate > 0 for 15 min Something is being dead-lettered; see the runbook

The vendored monitoring/prometheus/platform-events.rules.yml encodes these and more.

Upgrading from 0.2.0

0.3.0 brings the port in line with Go platform-events v2.1.0, and is breaking in several places (while the package is 0.y.z, a MINOR bump may break the public API). The authoritative list is CHANGELOG § Upgrade notes; this is the checklist.

Change What to check
The outbox schema cannot be upgraded in place (platform-events' migrations 001–011, golang-migrate tracking) Drain the 0.2.0 outbox, drop outbox_dead_letters, outbox_events, outbox_migrations, then apply_schema with 0.3.0
Only Tier 1 platform_* metrics remain; metrics.init / init_with_registry are gone Call init_metrics(MetricsIdentity(domain, service, environment)); update dashboards to the vendored monitoring/ bundle
BatchFailure is (id, code, message, retryable) envs[f.index] → map by f.id; custom publishers raise BatchError and call record_batch_published
dead_letter_handler receives the Envelope; a DLQ handler/publisher without max_receive_count uses 5 Handlers that read the raw SQS dict now use source_message_from_context()
Env durations need a unit; integers must be clean SQS_VISIBILITY_TIMEOUT=60 → 60s; log load_*().warnings
SQSConsumer(visibility_timeout=) defaults to 0 Pass it explicitly when constructing in code (sqs_consumer_options is unaffected)
A missing AWS region fails at construction Pass region= (the config loaders pass AWS_REGION), or set AWS_DEFAULT_REGION
parse_envelope requires time; specversion is no longer sent by default require_time=False for leniency; schema_version="1" to keep sending it
Payloads must be valid JSON (no NaN, no naive datetime) Use aware datetimes; check for float edge cases
Publishers hold a long-lived client per loop await publisher.close() / async with at shutdown
The drain timeout cancels handlers; stop() waits for start() and can raise Handlers tolerate CancelledError; a handler must not await a task that calls stop()
start() returns the coroutine; sync handlers' results are final inspect.iscoroutinefunction(x.start) is False
pgcommon>=0.4.0,<0.5.0 Your lock file; pgcommon's own 0.3 upgrade notes (0.4.0 changes only its FastAPI integration's error bodies)
The outbox backs off failed records by default; transient failures count no attempt Dead-lettering takes ~8–15 s of retries instead of 5 polls
Tracer name platform-events; sqs.receive rooted in an empty context Span filters and sampling rules
Handler / AcceptedHandler typing Handlers annotated Envelope[YourModel] parse inside the handler instead

Local development

Prerequisites

  • Python ≥ 3.12 (CI runs 3.12, 3.13 and 3.14; uv python install 3.12 if you have none)
  • uv: the environment is uv sync from uv.lock
  • Docker: testcontainers for the integration and e2e suites, and make docker-up for the local stack
  • Optional: hadolint and trivy for the image gates (brew install hadolint trivy); promtool for make rules-check (else it uses Docker); gh and cosign ≥ 2.4 to verify a release; a sibling platform-events checkout for make go-parity

Setup

git clone https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon
cd platform-eventcommon
make setup       # uv sync + copy .env-example → .env + git hooks
make docker-up   # floci (demo topic + queue + DLQ), Postgres on :5539, the observability stack
make ci          # everything CI's validate jobs run, in the same order (Docker required for the test suites)

Common commands

Command Description
make setup / make install uv sync, copy .env-example to .env if missing (never overwrites) and make install-hooks / uv sync only
make install-hooks Point git at .githooks/ (core.hooksPath): the pre-commit hook runs lock-check, format-check, lint and typecheck (git commit --no-verify skips it)
make pin-base-images / make docs-serve Re-resolve the Dockerfile's base images to their current @sha256: digests / browse the API docs with pydoc
make format / make format-check ruff format + ruff check --fix (rewrites files) / ruff format --check (CI). Aliases fmt / fmt-check
make lint / make typecheck ruff check / mypy --strict
make lock-check / make pip-check / make vulncheck uv lock --check / uv pip check (alias deps-check) / pip-audit (alias vuln-check)
make arch-lint The six import-linter contracts (the same script CI runs)
make metrics-lint / make metrics-doc / make rules-check Observability-standard conformance + inventory + monitoring/ assets / regenerate docs/observability/metrics-registry.md / promtool rules and alert tests
make go-parity Vendored migrations, monitoring/ and log texts against a platform-events checkout (PLATFORM_EVENTS_DIR=…, else a sibling checkout)
make docs-check Every docs/architecture/mermaid/*.mmd embedded byte-identically in ARCHITECTURE.md
make ci-scripts-test Regression tests for the CI shell scripts (detect-changes.sh)
make test-unit · make test-int · make test-e2e · make test Unit (no Docker) · integration (Postgres + floci; alias test-integration) · e2e (the full pipeline) · all three in one run (alias test-all)
make test-ci / make cover-check / make cover / make cover-func / make test-lowest Every suite with coverage, no threshold (CI's test gate) / the same + the ≥ 95% gate / HTML report in htmlcov/ / per-file missing lines / every suite at the dependency floors (replaces .venv)
make build / make build-constraints Wheel + sdist into dist/, twine check --strict, an install check / re-lock the build backend with hashes
make api-compat / make release-guard TAG=vX.Y.Z griffe public-API diff against the last release tag / pre-tag check (package version = tag, a dated CHANGELOG section)
make docker-build / make lint-docker / make scan / make smoke-test Build the example-worker image / hadolint / Trivy, CI's severities / build + CI's smoke-tests.sh
make docker-up / make run / make docker-down / make docker-reset / make docker-logs / make docker-ps The local stack / + the example worker / stop and remove it, volumes included / down + up / tail logs / list containers
make psql / make show-config psql in the local Postgres / eventcommon-config with .env loaded (credentials masked)
make ci lock-check + pip-check + format-check + lint + typecheck + metrics-lint + rules-check + docs-check + ci-scripts-test + vulncheck → test-ci → coverage gate → arch-lint → build
make clean / make help Remove .venv, caches, build artefacts and coverage output / list every target

Always run uv commands from a shell without another project's virtualenv active (env -u VIRTUAL_ENV make … if one is), or uv warns and may use the wrong environment.

Running a single test

uv run pytest tests/unit/test_envelope.py -k test_sign -v --no-cov
uv run pytest tests/integration/test_sns_publisher_floci.py -k test_publish_delivers_a_real_message_with_attributes -v --no-cov
uv run pytest tests/e2e/test_full_pipeline.py -v --no-cov

--no-cov matters: pyproject.toml's pytest addopts collect coverage, and the 95% fail_under would fail a subset of the suite on coverage even when its tests pass. Do not pass -q; it clashes with the configured addopts.

Developer tools

make docker-up
uv run eventcommon-config --strict   # what the loaders resolve from .env, with problems and reminders
make run                             # the example worker against the local stack (floci + Postgres)

uv run python gives a REPL with the package installed in editable mode, for trying the API against the same stack. The floci UI at http://localhost:4506 shows topics, queues and messages.

Testing SNS/SQS locally

The fastest way to see the delivery contract working, or to debug a consumer, is to drive the local stack by hand. make docker-up provisions what the library expects (docker/floci/init-floci.sh): topic eventcommon-demo, queue eventcommon-demo-q subscribed with RawMessageDelivery=true, and its DLQ eventcommon-demo-q-dlq through a RedrivePolicy with maxReceiveCount 5.

How it works

publish  ── SNSPublisher → floci SNS (eventcommon-demo) → SQS eventcommon-demo-q (raw delivery)
consume  ── SQSConsumer receives → handler returns → deleted
fail     ── handler raises → visible again → … → past max_receive_count → DLQ eventcommon-demo-q-dlq

Step 1 — Start the stack

make docker-up
set -a; . ./.env; set +a      # AWS_ENDPOINT_URL=http://localhost:4575, SNS_TOPIC_ARN, SQS_QUEUE_URL, test credentials

Step 2 — Publish an event

Save as publish_demo.py and run uv run python publish_demo.py:

import asyncio
import os

from eventcommon.envelope import new_envelope
from eventcommon.publisher import SNSConfig, SNSPublisher


async def main() -> None:
    config = SNSConfig(
        topic_arn=os.environ["SNS_TOPIC_ARN"],
        region=os.environ["AWS_REGION"],
        endpoint_url=os.environ["AWS_ENDPOINT_URL"],
    )
    async with SNSPublisher(config) as publisher:
        await publisher.publish(
            new_envelope("demo.thing.happened", "demo", {"n": 1}, tenant_id="acme")
        )


asyncio.run(main())

Step 3 — Look at the raw message

aws --endpoint-url "$AWS_ENDPOINT_URL" sqs receive-message --queue-url "$SQS_QUEUE_URL" \
  --message-attribute-names All --visibility-timeout 0

The body is the envelope itself (raw delivery), and the attributes include EventType, TenantID, Source and EventID.

Step 4 — Consume it, then fail it into the DLQ

import asyncio
import os

from eventcommon.consumer import DLQConfig, SQSConfig, SQSConsumer, SQSDLQPublisher


async def handle(env):
    print("got", env.type, env.tenant_id)
    raise RuntimeError("simulated failure")  # remove to see the message deleted


async def main() -> None:
    region, endpoint = os.environ["AWS_REGION"], os.environ["AWS_ENDPOINT_URL"]
    async with SQSDLQPublisher(DLQConfig(region=region, endpoint_url=endpoint)) as dlq:
        consumer = SQSConsumer(
            SQSConfig(
                queue_url=os.environ["SQS_QUEUE_URL"],
                region=region,
                endpoint_url=endpoint,
                wait_seconds=1,
            ),
            handle,
            visibility_timeout=1,
            max_receive_count=2,
            dlq_publisher=dlq,
        )
        task = asyncio.create_task(consumer.start())
        await asyncio.sleep(10)
        await consumer.stop()
        await task


asyncio.run(main())

After the third delivery the message is forwarded to eventcommon-demo-q-dlq with DLQReason, OriginalQueue and FailedAt attributes:

aws --endpoint-url "$AWS_ENDPOINT_URL" sqs receive-message \
  --queue-url "${SQS_QUEUE_URL}-dlq" --message-attribute-names All

Troubleshooting

Symptom Likely cause Fix
Every message counted malformed The subscription lacks RawMessageDelivery=true, so the body is SNS's notification JSON Set the subscription attribute
ValueError: … no AWS region configured No region=, no AWS_REGION / AWS_DEFAULT_REGION, and no profile region Export AWS_REGION (the .env sets us-east-1) or pass region=
start() raises DLQNotConfiguredError dlq_publisher set on a queue without a RedrivePolicy Add the RedrivePolicy, or drop DLQ forwarding
Messages go to the DLQ without the library's attributes The queue's maxReceiveCount is at or below max_receive_count, so SQS moved them itself Keep the queue's maxReceiveCount higher
A handler runs twice for one message The handler outlived the visibility timeout with visibility_timeout=0 Set visibility_timeout (it is then extended)
A handler's queries return no rows The pool lacks guc_provider=guc_set_from_context, or the policy does not handle the tenant Wire the provider; check the policy against app.tenant_id
connection refused on 4575 The stack is down, or another project holds the port make docker-up; set FLOCI_PORT and AWS_ENDPOINT_URL together

Testing

make test-unit   # unit tests — no Docker required
make test-int    # integration tests — floci + Postgres (testcontainers)
make test-e2e    # the full pipeline: Postgres → runner → SNS → SQS → consumer
make cover       # full suite + HTML coverage report (threshold: ≥ 95%)

Skip Docker-dependent tests:

uv run pytest -m "not integration and not e2e"

Integration and e2e tests start their own containers (testcontainers; Docker must be running) and never use the docker-compose.yaml stack. Every test has a 300 s timeout (pytest-timeout), and faulthandler dumps every stack at 240 s.

Status (0.3.0 branch): 1700 tests pass (plus 1 expected failure pinning the year-0000 divergence from Go), with 99.97% branch coverage of src/eventcommon. That covers 58 unit files (the test_cov_*.py suites close every coverage gap with behaviour tests, Go-confirmed where they assert Go parity), 6 integration files and the e2e pipeline. The cross-language interop harness passes 53/53 checks against platform-events v2.1.0.

Canonical tests (do not break)

  • tests/e2e/test_full_pipeline.py: no dual write, at-least-once delivery and idempotent redelivery together, against real Postgres and real SNS/SQS semantics.
  • tests/unit/test_interop.py, test_go_wire_corpus.py, test_wire_go_parity.py: the wire format and HMAC against golden vectors and messages generated by platform-events v2.1.0, including the Go consumer's decisions on malformed and unusual bodies.
  • tests/unit/test_schema_parity.py, test_monitoring.py, test_log_parity.py (make go-parity): the vendored migrations, monitoring/ assets and every log message identical to platform-events.
  • tests/unit/test_metrics_standard.py, test_metrics_inventory.py (make metrics-lint): every metric conforms to the registry and to Go's, and docs/observability/metrics-registry.md is current.
  • tests/integration/test_outbox_schema_postgres.py: concurrent and Go-compatible migrations on real Postgres.
  • tests/integration/test_outbox_db_stall.py: a database that stalls mid-poll fails each poll within 5 s, turns healthy() False, and recovers.
  • tests/unit/test_sqs_consumer_lifecycle.py, test_outbox_runner_execution.py, test_cov_sqs_consumer.py, test_cov_outbox_runner.py: drain, cancellation, panic recovery and settlement rules.
  • tests/unit/test_migrations_lock_timeout.py: lock_timeout never weakens to '0ms' and is validated before any SQL.
  • tests/unit/test_null_body_parity.py, test_cov_envelope.py: Go's decoder behaviour on edge-case bodies (including null), with the one remaining divergence (year 0000) pinned as a strict expected failure.

Coverage

pyproject.toml sets fail_under = 95 (branch coverage of src/eventcommon). make test-ci runs every suite once with coverage, and .github/scripts/coverage-gate.sh checks the threshold separately, so a failing test and a coverage drop are told apart. make cover-check runs both, and make cover-func lists missing lines per file from the last run. On Python 3.14, coverage's default sys.monitoring core does not record the entry / exit arcs of a plain with block, so two arcs in outbox/runner.py show as partial locally; COVERAGE_CORE=ctrace (and CI's 3.12 gate) measure them as covered. No reachable code is excluded: the two # pragma: no cover lines are the CLI's __main__ guard and the uninstalled-package version fallback.

Testing in consuming services

eventcommon.mock ships ready-made, thread-safe test doubles, so consuming services never need an AWS emulator or SNS 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) Validates (id/type/source) and records the envelope; raises any error set via set_error
await publish_batch(envs) Records the valid envelopes; raises one BatchError for invalid ones (InvalidEnvelope) and any IDs in set_batch_error
published List of all recorded envelopes (thread-safe)
set_error(err) Makes all subsequent publish calls raise err
set_batch_error(err) Makes publish_batch fail the BatchError's IDs as given
reset() Clears recorded envelopes and any configured errors

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": "alice@example.com"},
        tenant_id="acme",
    )
    await consumer.inject(env)  # delivers directly — no SQS, no network

    assert worker.email_was_sent_to("alice@example.com")
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 as the SQS consumer does: raises MalformedEnvelopeError (without calling the handler) for a missing id / type / source or a NUL in tenant_id / trace_id; round-trips it through JSON; decodes a codec payload with MockConsumer(codec=...); and binds the tenant GUC, trace ID, source_message_from_context() (queue MOCK_QUEUE_URL unless queue_url= is given) and DLQ attribution. Settles a handler's forward like production: returns normally once send_to_dlq succeeded whatever the handler raised, and raises DeadLetterNotSentError when the handler returned after a failed one. Returns without a handler
is_running Whether start has been called without a matching stop

MockDLQPublisher(dlq_url="mock://dlq") records send_to_dlq calls in .sent (validated like SQS, counted as dead-lettered), with set_error / reset; a failed send marks the attribution as SQSDLQPublisher does.

The mocks satisfy the same Publisher / Consumer protocols as the real implementations, so they drop in wherever the real ones are used. For metrics in tests, pass a fresh CollectorRegistry to init_metrics and call metrics.reset() between tests.

Environment variables

Read only when the service calls a load_* loader, metrics_identity_from_env, or runs eventcommon-config. Nothing is read at import time, and the library reads no OTEL_* or log-level variable itself (the deprecated load_otel aside). .env-example documents the local defaults (make setup copies it to .env).

Durations use Go's syntax (5s, 250ms, 1m30s), so a shared ConfigMap means the same in both languages. Anything Go rejects is rejected too: a bare number (300; write 300s), surrounding whitespace, nan / inf, exponents, _ separators, non-ASCII digits, and integers outside int64. An invalid integer or duration never raises: the default is applied and a warning (Go's exact text) is added to .warnings. A missing required variable is reported by .validate() (ConfigValidationError).

AWS, SNS and SQS (load_sns, load_sqs)

Variable Default Notes
AWS_REGION AWS_DEFAULT_REGION, else us-east-1 Read by the loaders, and by the components themselves when region=None (before boto3's own chain)
AWS_ENDPOINT_URL — http://localhost:4575 for the local floci (FLOCI_PORT moves it)
SNS_TOPIC_ARN — Required for the publisher
SQS_QUEUE_URL — Required for the consumer
SQS_MAX_MESSAGES 10 1–10; SQS hard limit
SQS_WAIT_SECONDS 20 Long-poll duration
SQS_VISIBILITY_TIMEOUT 30s Duration
SQS_CONCURRENCY 1 Parallel handler tasks (FIFO: message groups)
SQS_MAX_RECEIVE_COUNT 0 (unset) Applied only when > 0; pairs with dead_letter_handler / dlq_publisher
SQS_HANDLER_TIMEOUT 0 (off) Bounds each handler call
SQS_DRAIN_TIMEOUT 0 (consumer default 30s) stop() wait before cancelling handlers; keep below terminationGracePeriodSeconds
SQS_RETRY_BACKOFF 0 (off) Failed delivery n is hidden base·2^(n-1)
SQS_MAX_RETRY_BACKOFF 0 (15m) Cap for SQS_RETRY_BACKOFF; at most 12h
SQS_QUEUE_DEPTH_INTERVAL 0 (off) Samples platform_queue_depth / platform_dlq_depth; minimum 10s

Outbox and database (load_outbox)

Variable Default Notes
OUTBOX_POLL_INTERVAL 5s
OUTBOX_BATCH_SIZE 50 Records per poll cycle
OUTBOX_MAX_ATTEMPTS 5 Before moving to dead-letter
OUTBOX_CLAIM_LEASE_DURATION 0 (runner default 10m) How long a claimed record is hidden from other runners
OUTBOX_STARTUP_JITTER 0 Use 5s–10s with multiple replicas
OUTBOX_PUBLISH_CONCURRENCY 1 Parallel publishes per poll cycle
OUTBOX_PUBLISH_TIMEOUT 10s Per-record publish timeout
OUTBOX_DRAIN_TIMEOUT 30s runner.stop() wait bound
OUTBOX_RETRY_BACKOFF 1s A record's n-th failed publish waits base·2^(n-1)
OUTBOX_MAX_RETRY_BACKOFF 5m Cap for OUTBOX_RETRY_BACKOFF
OUTBOX_GAUGE_INTERVAL 15s Backlog gauge refresh
DATABASE_URL / PG_* — Postgres pool config via pgcommon.config_from_env (OutboxEnv.db); its warnings are merged into OutboxEnv.warnings
MIGRATION_DATABASE_URL DATABASE_URL DDL-privileged DSN for apply_schema (OutboxEnv.migration_database_url)

str(OutboxEnv) / repr(OutboxEnv) mask credentials (URL userinfo, and password / passwd / sslpassword in URL and libpq forms); never log .db / .database_url directly.

Metrics identity

Variable Default Notes
APP_ENV (then ENVIRONMENT) dev The environment label when MetricsIdentity.environment is empty (trimmed, lower-cased)
APP_NAME — (eventcommon-config: eventcommon) The service label in metrics.metrics_identity_from_env (trimmed, lower-cased)
BUILD_VERSION the installed eventcommon version Shown by eventcommon-config; not a metric label (platform_library_info reports the installed library)
OTEL_SERVICE_NAME, OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_INSECURE — Read only by the deprecated load_otel; tracing is configured by the service

Example worker only

Read only by examples/outbox_worker (the image CI builds), a template for a service's own entrypoint, not library settings:

Variable Default Notes
METRICS_PORT 9464 /metrics (the bundled Prometheus scrapes worker:9464)
HEARTBEAT_PATH /tmp/heartbeat Touched while Runner.healthy(...) holds; the image's HEALTHCHECK reads it
LOG_LEVEL / LOG_FORMAT info / json (console on a terminal) structlog output
STARTUP_TIMEOUT 60s How long the database is retried at start (1→10 s backoff) before exiting 1 with one line
SHUTDOWN_TIMEOUT 25s Bound on the whole SIGTERM shutdown; keep it below terminationGracePeriodSeconds

Security

Topic Guidance
Who may publish SNS topic policies and IAM decide; the library does not authenticate producers. Within one account and IAM boundary that is the control. Across trust boundaries (webhooks, HTTP ingress), sign with sign_envelope and verify in the receiver (HMAC helpers)
Tenant isolation in handlers The consumer binds each message's tenant_id into pgcommon's GUC context and resets it afterwards; each message runs in its own task with its own context copy. RLS policies (yours) do the enforcement. A NUL in tenant_id / trace_id makes the message malformed, so it never reaches a GUC
"system" events A real GUC value, not an RLS bypass. Handle them with a pool or policy built for cross-tenant work, never by widening the app role
Message bodies and PII Malformed bodies are logged as size + SHA-256 only (malformed_body_logging=True opts into a 512-byte excerpt). Payloads are never metric labels or span attributes. Do not log env.data in handlers
Credentials AWS credentials come from boto3's chain (IRSA in production); nothing is logged. OutboxEnv, eventcommon-config and make show-config mask database passwords; pgcommon ≥ 0.3.1 masks the DSN in reprs and error chains
HMAC keys ≥ 32 bytes (KeyTooShortError), constant-time verification, keys from a secrets manager, rotation by accepting two keys for a window
Poison and oversized input Malformed bodies dead-lettered on first receipt; max_receive_count and max_attempts bound retries; the claim does not load payloads over 1 MiB; event_type labels are capped
Example-worker image python:3.14-slim, digest-pinned, numeric non-root USER 10001:10001, pip stripped, dependencies from uv.lock. Trivy-gated (CRITICAL / HIGH / UNKNOWN), with a CycloneDX SBOM, SLSA provenance and a Cosign keyless signature
Supply chain PyPI uploads only through Trusted Publishing (OIDC, no stored token) from release.yml; the release's SHA256SUMS is Cosign-signed (Deployment). The build backend is hash-locked (build-constraints.txt), pip-audit runs in CI, and every third-party action is pinned by commit SHA

SECURITY.md has the trust model, the scope of vulnerability reports and how to report one privately. ARCHITECTURE.md § Threat model has the STRIDE analysis.

Observability — Prometheus and OTel

Both are optional. The publisher, consumer and outbox runner work without metrics.init_metrics or a registered OTel provider.

Prometheus metrics

eventcommon implements the Enterprise Platform Observability Standard exactly as platform-events v2.1.0 does. It emits only Tier 1 platform_* metrics, carrying the required domain / service / environment labels from a mandatory MetricsIdentity. Full details (tiers, semantics, CI enforcement and every difference from Go) are in docs/observability/README.md; the generated inventory is docs/observability/metrics-registry.md.

from eventcommon import metrics

# Once at startup. environment empty → APP_ENV, then ENVIRONMENT, else "dev".
identity = metrics.MetricsIdentity(domain="iam", service="event-consumer")
# Or: metrics.metrics_identity_from_env("iam")  # service from APP_NAME
for warning in metrics.init_metrics(identity):  # registry=None → prometheus_client.REGISTRY
    logger.warning(str(warning))  # a platform_* metric the registry refused — disabled, not fatal
# Raises InvalidMetricsIdentityError (a ValueError) on an invalid identity, changing nothing.
# Options: event_type_limit=200, event_types=[...] (pre-register the event types you handle).

# For isolated test registries:
metrics.init_metrics(identity, CollectorRegistry())
  • Required labels are bound centrally. domain, service and environment come from one identity, so no recording site can omit or misspell them. Pass the same values (and registry) pgcommon's init_with_identity gets.
  • Fail-soft registration. A platform_* metric whose name is already registered with another shape is disabled and returned as a RegistrationWarning; everything else keeps recording.
  • Go's exposition. No *_created series, and a family appears once it has a series. The interop job compares the exposition with Go's.
  • Removed metrics. The pre-standard metrics, the legacy init path (metrics.init / metrics.init_with_registry), the legacy= toggle and MetricsIdentity.version were removed without a compatibility period, because nothing had been deployed. See docs/observability/README.md § Removed pre-standard metrics.

Tier 1 metrics. Every one also carries domain, service and environment. Canonical ones may back alerts and SLOs; Proposed ones are shadow-emitted until ratified.

Metric Type Labels Status Description
platform_messages_received_total Counter queue canonical SQS deliveries, redeliveries included
platform_messages_processed_total Counter queue, event_type canonical Handler succeeded, message deleted
platform_messages_failed_total Counter queue, event_type, reason canonical malformed / decode_error / handler_error / handler_panic / dead_letter_error
platform_retry_total Counter operation, event_type canonical Failed attempt left for retry (consume, outbox_publish)
platform_dlq_messages_total Counter operation, event_type, reason canonical Dead-lettered, counted once (consume: max_receive_count / malformed / decode_error / explicit; outbox_publish: max_attempts)
platform_dependency_request_seconds Histogram dependency, operation, outcome proposed SNS / SQS / codec call latency and outcome
platform_event_propagation_seconds Histogram queue, event_type proposed Envelope time → first receipt
platform_messages_in_flight Gauge queue proposed Messages being processed, per replica
platform_messages_published_total Counter topic, event_type, outcome proposed Publish attempts per event
platform_message_processing_duration_seconds Histogram queue, event_type proposed Handler / dead-letter handler wall time
platform_outbox_pending_events Gauge — proposed Due unpublished outbox rows (capped at 100000)
platform_outbox_leased_events Gauge — proposed Leased / backing-off outbox rows (capped at 100000)
platform_outbox_oldest_pending_age Gauge — proposed Age (s) of the oldest unpublished outbox row — the delivery-stall signal
platform_outbox_publish_attempts_total Counter event_type, outcome proposed Outbox runner publish attempts
platform_outbox_errors_total Counter operation proposed poll / unmarshal / mark_published / gauge-query failures
platform_outbox_dead_letter_operations_total Counter operation proposed Dead-letter records reprocessed / discarded
platform_telemetry_label_overflow_total Counter label proposed event_type values replaced (__oversized__, __other__)
platform_library_info Gauge library, library_version proposed Always 1; library="platform-eventcommon"
platform_duplicate_messages_total Counter queue, event_type proposed Inbox duplicates skipped (inbox.handler / Store.process)
platform_queue_depth Gauge queue proposed Visible messages, sampled with queue_depth_interval (opt-in)
platform_dlq_depth Gauge queue proposed Visible messages in the RedrivePolicy DLQ, sampled likewise
platform_message_timeouts_total Counter queue, event_type, operation proposed handler_timeout expiries by stage (decode, dead_letter_handler, handler)
platform_outbox_ordering_blocked_events Gauge — proposed Ordered records waiting behind their key's head

queue / topic are the queue / topic name, never the URL / ARN. event_type is bounded in-process: at most 200 distinct values (__other__ beyond), 128 bytes (__oversized__ beyond), and unknown for an empty or unparseable type.

OpenTelemetry

OTel is always initialised by the consuming service (e.g. fastapicommon.tracing.init_from_env()). eventcommon calls trace.get_tracer("platform-events"), the same instrumentation scope as Go, and produces no-op spans when no provider is registered.

Span Kind Attributes
sns.publish PRODUCER messaging.system=aws_sns, messaging.destination (topic ARN), messaging.message_id, events.event_type, events.event_id
sns.publish.batch PRODUCER messaging.system=aws_sns, messaging.destination, messaging.batch.message_count
sqs.receive CONSUMER messaging.system=aws_sqs, messaging.destination (queue URL), messaging.message_id, messaging.operation=process, events.event_type, events.event_id
sqs.dlq_forward PRODUCER messaging.system=aws_sqs, messaging.operation=publish, messaging.destination (DLQ URL), messaging.message_id, events.event_type, events.dlq.source_queue

The publisher injects traceparent / tracestate / baggage as message attributes. sqs.receive is rooted in an empty context and linked to the producer's span through them, and the producer's baggage is carried into the handler.

Logging correlation

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()) and 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 = decode_payload(env.data, UserCreatedPayload)
    await repo.create_user(payload)

    log.info("event processed")

The library's own log lines use Go's messages verbatim (tests/unit/test_log_parity.py), so one log query matches both languages. They go through the logger you pass (logger= on each config): structlog, or a stdlib logging.Logger with the fields appended as key=value.

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

Alerts, rules and dashboard

monitoring/ vendors platform-events' reference assets, byte-identical (make go-parity):

  • prometheus/platform-events.rules.yml: recording rules and alerts, with promtool unit tests in platform-events.rules.test.yml (make rules-check);
  • grafana/platform-events.json: the dashboard;
  • kubernetes/keda-scaledobject.example.yaml: queue-depth autoscaling.

Because the metric families are identical, the same assets serve Go and Python services, and a single query aggregates both. Group every query by (domain, service, environment), never by service alone. The runbook is docs/observability/runbook.md.

Deployment

eventcommon itself is released to PyPI, and services adopt it by bumping their pin; it is never deployed on its own. The example-worker image is a reference, built and scanned by every release.

Installing a release

Pin an exact version in production (VERSIONING.md § Consume a release has the pin styles):

pip install "eventcommon==0.4.0"
uv add "eventcommon==0.4.0"

The files on PyPI are byte-identical to the GitHub Release's wheel and sdist: the release builds them once and uploads the same files to both.

Container image — the example worker

Executable In the image Purpose
python -m examples.outbox_worker.main ENTRYPOINT, EXPOSE 9464, a heartbeat HEALTHCHECK Applies the outbox schema, runs a Runner and an SQSConsumer, serves /metrics, shuts down within SHUTDOWN_TIMEOUT

The Dockerfile has a build and a runtime stage:

  • Build stage. ghcr.io/astral-sh/uv and python:3.14-slim, both digest-pinned. It installs the runtime dependencies from uv.lock and the wheel built with the hash-locked backend (build-constraints.txt).
  • Runtime stage. python:3.14-slim, digest-pinned, with pinned Debian fixes. It removes pip and runs as the numeric USER 10001:10001, so runAsNonRoot is enforceable.
  • Size. The smoke gate fails above 250 MB.
  • Tags. They have no v: X.Y.Z always, plus X.Y and X only when the release is the newest stable version of that line, and never latest, because it is a demonstration, not a deployable service. A pre-release gets only its own tag.

Verifying a release

Needs cosign ≥ 2.4. A tag-triggered release is signed by release.yml@refs/tags/<tag>, and a re-run dispatched from main by release.yml@refs/heads/main; the regular expression below accepts both.

TAG=vX.Y.Z   # the release to verify
REPO=BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon
IDENTITY="^https://github\.com/${REPO}/\.github/workflows/release\.yml@refs/(tags/${TAG}|heads/main)$"

# Wheel + sdist: verify the signed checksums, then the files against them
gh release download "$TAG" -R "$REPO" -p SHA256SUMS -p SHA256SUMS.sigstore.json -p '*.whl' -p '*.tar.gz'
cosign verify-blob --bundle SHA256SUMS.sigstore.json \
  --certificate-identity-regexp "$IDENTITY" \
  --certificate-oidc-issuer https://token.actions.githubusercontent.com \
  SHA256SUMS
sha256sum --check SHA256SUMS   # macOS: shasum -a 256 -c SHA256SUMS

# Example-worker image (tag without the v)
cosign verify --certificate-identity-regexp "$IDENTITY" \
  --certificate-oidc-issuer https://token.actions.githubusercontent.com \
  "ghcr.io/bcbp-solutions-fzc-llc/platform-eventcommon:${TAG#v}"

Each GitHub Release carries the wheel, the sdist, SHA256SUMS, SHA256SUMS.sigstore.json, the image's SLSA provenance and its CycloneDX SBOM; PyPI also shows the Trusted Publishing attestations. Releases up to v0.2.0 predate signing and SHA256SUMS. VERSIONING.md § Verifying a release is the canonical copy of these commands.

Migration safety

outbox.apply_schema(pool) and inbox.apply_schema(pool) apply platform-events' own migration files, tracked as golang-migrate tracks them, so Go and Python services can share and migrate one database:

  • Locking. The run takes every advisory-lock key Go's runner may use (/db, db, /) as transaction-level locks, so concurrent Go and Python runners serialise. It runs in one transaction with lock_timeout (30 s by default) set first, so a migration blocked behind a long transaction on the outbox tables fails with asyncpg.LockNotAvailableError instead of queueing every query behind it. Because the locks are transaction-level, it also works through PgBouncer.
  • Refused states. A dirty version (a Go run that failed half-way) raises MigrationDirtyError. A version this release does not ship raises SchemaVersionNotInSourceError; migrate down with the newer release before rolling back.

Run it once per deploy, at startup or as a pre-deploy job, with a DDL-privileged role (MIGRATION_DATABASE_URL). Concurrent runs are safe, and the loser finds nothing to do. Go and Python services sharing a database must run releases that ship the same migration set. See ARCHITECTURE.md § Schema migration flow.

Versioning and releases

Release status: the newest release is 0.4.0 (2026-10-22). It brings platform-events v2.2.1 parity (handler-initiated dead-letters are settled once) and dependency updates, and is compatible with 0.3.0. 0.3.0 (2026-10-16) brought platform-events v2.1.0 parity and production hardening, and is breaking against 0.2.0; read Upgrading from 0.2.0 first.

Earlier releases were 0.2.0 (2026-08-07), and 0.1.1 and 0.1.0 (2026-07-27).

Go parity: this port tracks platform-events v2.2.1: the same wire format, HMAC, schema, metrics, configuration, log messages and behaviour, except for the documented divergences. The Python package keeps its own version line (0.x); the CHANGELOG names the Go version each release corresponds to.

See VERSIONING.md for the full release process, the public-API surface SemVer applies to, the envelope wire-format rules, and how Trusted Publishing to PyPI is set up. Summary:

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

While the package is 0.y.z, a MINOR bump may break the public API (SemVer §4) and a PATCH may not. The release's api-compat gate (griffe) enforces exactly that.

# Tag a commit on main; pushing the tag triggers the release workflow.
make release-guard TAG=vX.Y.Z          # pyproject version == tag, a dated CHANGELOG [X.Y.Z]
git tag -a vX.Y.Z -m "vX.Y.Z"
git push origin vX.Y.Z

release.yml runs these stages in order:

  1. verify checks the dispatch ref, that the tag is at HEAD and on origin/main, that the version equals the tag, that the CHANGELOG section is dated, and which floating image tags may move.
  2. The CI gates run again at the tag: tests, quality (Go parity required), api-compat (blocking), Hadolint, Trivy, smoke and interop.
  3. dist builds the wheel and sdist once, reproducibly.
  4. push, only after every gate: publishes the fingerprint-checked image with SBOM, SLSA provenance and a Cosign signature.
  5. pypi-publish uploads the same files via Trusted Publishing.
  6. publish creates the GitHub Release.

Interop gates push, so a red cross-language check stops the whole release.

CI

The pipeline is platform-pgcommon-py's (itself the Go repos', translated to Python tooling): the same jobs, gates and scripts (.github/scripts/), adapted to this library and its example-worker image. Every third-party action is pinned by commit SHA.

  • ci.yml runs on every push and pull request to main. changes (detect-changes.sh) decides whether the change is documentation only; if so, the build and test jobs are skipped through if:, so the required checks still report. Otherwise validate-test and validate-quality run in parallel with build-image. The image then goes to trivy and smoke, then interop runs, and pr-summary comments on the PR. On a push to main, push publishes the image (main / sha-<short>), but only one whose content fingerprint matches the image Trivy and the smoke tests judged, and Cosign signs it.
  • validate-test.yml (reusable) runs make test-ci (unit + integration + e2e on floci + Postgres, with coverage), the coverage gate (≥ 95%) and make arch-lint on Python 3.12. It also runs the suite on 3.13 / 3.14 and at the lowest direct dependencies.
  • validate-quality.yml (reusable) runs:
    • uv lock --check and uv pip check;
    • the RLS session-SET app.* guard;
    • ruff format / lint and mypy --strict;
    • make metrics-lint, make rules-check (promtool), make docs-check and make ci-scripts-test;
    • Go parity against a platform-events checkout: required on main and for releases, a warning on PRs without access;
    • pip-audit, digest-pinned FROM lines, and make build.
  • release.yml runs on vX.Y.Z / vX.Y.Z-pre.N tags; see Versioning and releases.
  • docs.yml checks the diagram sync on documentation-only changes (not a required check).
  • changelog-check.yml fails a PR that changes src/, pyproject.toml, uv.lock, build-constraints.txt or the Dockerfile without updating CHANGELOG.md.
Job (ci.yml) Check name What it gates Required
changes Detect changes Docs-only decision + the shared SOURCE_DATE_EPOCH —
validate-test → test Validate / Test / test Full suite on Python 3.12 + coverage ≥ 95% + import-linter ✅
validate-test → python / lowest-deps Validate / Test / python 3.13, … / python 3.14, … / lowest direct dependencies The suite on the other Pythons and at the dependency floors —
validate-quality → quality Validate / Quality / quality Lock, format, lint, types, RLS guard, metrics standard, rules, docs, CI-script tests, Go parity, pip-audit, digest pins, packaging ✅
api-compat API compatibility griffe vs the last release tag; warning only here, blocking in release.yml —
build-image Build image (cache) Hadolint + .dockerignore, then a cached build —
trivy Trivy CVE scan CRITICAL / HIGH / UNKNOWN → fail; SARIF to the Security tab; CycloneDX SBOM artifact —
smoke Smoke tests smoke-tests.sh: size ≤ 250 MB; a bare run exits 1 naming SNS_TOPIC_ARN; the installed version; eventcommon-config --strict; HEALTHCHECK; no pip; non-root —
interop Cross-language compatibility platform-interop-tests (SHA-pinned). Not required in CI; gates the release's push —
pr-summary PR summary PRs only: one status comment —
push Push image → GHCR Pushes only: fingerprint-checked push, SBOM + provenance, Cosign sign + verify —

Private repository: checkouts use CI_REPO_READ_TOKEN || github.token without persisting credentials. CI_REPO_READ_TOKEN also checks out platform-events for the Go parity step, and reusable workflows get explicit secrets: maps.

Run the same gates locally before pushing:

make ci            # lock-check + pip-check + format-check + lint + typecheck + metrics-lint + rules-check
                   # + docs-check + ci-scripts-test + vulncheck + test-ci + coverage gate + arch-lint + build
make go-parity     # against a sibling platform-events checkout (or PLATFORM_EVENTS_DIR=…)
make api-compat    # griffe vs the last release tag
make lint-docker   # hadolint (CI: first step of build-image)
make scan          # Trivy, same severities as CI
make smoke-test    # build the image + .github/scripts/smoke-tests.sh

make docs-check checks that every docs/architecture/mermaid/*.mmd is embedded byte-identically in ARCHITECTURE.md; edit a .mmd file and its embedded block together.

Docker

What the bundled docker-compose.yaml starts

Every port is bound to 127.0.0.1 and can be moved with the variable in brackets, so the stack can run beside other projects' stacks.

Container Image Host port Purpose
floci floci/floci (digest-pinned) 4575 (FLOCI_PORT) AWS emulator (SNS, SQS); provisions the demo topic, queue and DLQ on start
floci-ui floci/floci-ui 4506 (FLOCI_UI_PORT) Web console for floci's resources and messages
postgres postgres:16-alpine 5539 (POSTGRES_PORT) Database eventcommon (user and password eventcommon)
otel-collector otel/opentelemetry-collector-contrib 4317 / 4318 OTLP receiver; logs trace batches (no trace backend)
loki · promtail Loki · Promtail 3100 · — Logs · log shipping from the worker
prometheus · grafana Prometheus · Grafana 9090 · 3000 Scrapes worker:9464 · provisions the vendored monitoring/grafana dashboard
worker (profile demo) built from the Dockerfile — The example worker; make run starts it

make docker-up starts everything but the worker and waits for floci; make run adds the worker; make docker-down removes the stack with its volumes. The integration and e2e suites don't use it: they start their own containers.

Building the example-worker image

make docker-build   # the example worker image, tagged :local
make smoke-test     # build + CI's image gates (.github/scripts/smoke-tests.sh)
make lint-docker    # hadolint, as the first step of CI's build-image job
make scan           # Trivy, CI's severities

Running the worker with the image

docker run --rm -p 9464:9464 \
  -e AWS_REGION=eu-west-1 -e SNS_TOPIC_ARN=arn:aws:sns:eu-west-1:123456789012:orders \
  -e SQS_QUEUE_URL=https://sqs.eu-west-1.amazonaws.com/123456789012/orders-q \
  -e DATABASE_URL='postgres://app:…@db:5432/app?sslmode=verify-full' \
  -e APP_NAME=orders-worker -e APP_ENV=production \
  ghcr.io/bcbp-solutions-fzc-llc/platform-eventcommon:<version>

A bare docker run exits 1 naming the first missing variable, with no traceback. The worker exits 1 when the runner or the consumer dies, so the orchestrator restarts it.

Minimum required environment variables

# Publisher + outbox runner
AWS_REGION=eu-west-1
SNS_TOPIC_ARN=arn:aws:sns:eu-west-1:123456789012:orders
DATABASE_URL=postgres://app_user:…@pgbouncer:6432/app?sslmode=verify-full
MIGRATION_DATABASE_URL=postgres://migrator:…@postgres:5432/app?sslmode=verify-full   # apply_schema
APP_ENV=production              # the environment label — use the platform's agreed spelling

# Consumer
SQS_QUEUE_URL=https://sqs.eu-west-1.amazonaws.com/123456789012/orders-q
SQS_VISIBILITY_TIMEOUT=60s

Parity with platform-events

eventcommon is the Python port of platform-events v2.2.1 and has its features:

  • the envelope, HMAC and the SNS publisher;
  • the codec hook (GlueDecodeCodec);
  • the SQS consumer: FIFO group ordering, handler and drain timeouts, retry backoff, queue-depth sampling and DLQ forwarding;
  • the DLQ publisher;
  • the transactional outbox: per-key ordering, retry backoff and dead-letter replay;
  • the inbox, the metrics registry and the reference monitoring.

Go is the source of truth. The wire format, metric names, labels, buckets, help text, log and warning texts, environment variables and defaults are the same. So are the outbox and inbox schemas, down to the migration files and golang-migrate tracking, so Go and Python services can share one outbox or inbox database.

Wire interop is verified end to end: every envelope and payload shape, single and batch publishing, FIFO group order, the Glue codec, HMAC signatures, trace links, baggage, tenant / trace context and a shared outbox, published by either language over SNS/SQS and consumed by the other. tests/unit/test_go_wire_corpus.py replays messages platform-events published, and the Go consumer's decisions on malformed and unusual bodies, so any drift fails this repository's CI.

The behavioural differences left are ones Python cannot express. An envelope time in year 0000 is rejected (Python's datetime starts at year 1; Go accepts it), and data containing invalid UTF-8 is forwarded with U+FFFD in its place (Go forwards the raw bytes). The full list, including the Python-only additions, is ARCHITECTURE.md § Divergences from Go. Where the two languages differ in style, both forms are available:

  • Options. Every Go functional option has a Python counterpart with the same name, usable positionally (SQSConsumer(cfg, handler, with_concurrency(4), with_retry_backoff(timedelta(minutes=1), timedelta(minutes=30))), SNSPublisher(cfg, with_codec(codec))) or as the constructor keyword it stands for (concurrency=4, codec=codec). Options apply in order, after the keywords. Durations take seconds or a timedelta (Go's time.Duration).
  • Clients. SNSPublisher(cfg, client=…) and SQSConsumer(cfg, handler, client=…) take a caller-owned aioboto3 client (Go: NewWithClient / NewSQSConsumerWithClient), as SQSDLQPublisher(client=…) does. The library never closes it; without one, they open their own.
  • Durations use Go's time.ParseDuration syntax and nothing else: SQS_HANDLER_TIMEOUT=300 is rejected (a warning, default applied) by both. Write 300s.

Metric-level differences are listed in docs/observability/README.md § Parity with platform-events.

Dependencies

1. Consumers (projects that import this library)

Consumer eventcommon pin Uses
Python platform services (FastAPI, on fastapicommon + pgcommon) their own Publisher, consumer, outbox, inbox, metrics, mocks
platform-interop-tests (Python probe) editable path to this checkout Cross-language compatibility with platform-events (CI's interop job)

2. Upstream libraries

Runtime dependencies lists the ranges. pgcommon is the load-bearing one: the outbox and inbox use its Pool and run_in_tx, the consumer binds tenants through its with_guc_set, and its floor is the latest release, 0.4.0 (0.3.1 brought the bounded queries on a stalled database and the DSN masking). aioboto3 and boto3 provide the clients and the error classes the retry classification reads. CI's lowest direct dependencies job runs every suite at each floor.

3. Platform repos

Repo Relationship
platform-events (Go) The source of this port: the parity target (v2.2.1), and the origin of the vendored migrations and monitoring/ assets
platform-pgcommon-py Runtime dependency (above); the model for this repository's pipeline, Makefile and documentation
platform-fastapicommon No dependency either way. A service using both passes the same CollectorRegistry and identity values, and sets up tracing there
platform-interop-tests Cross-language compatibility suite run by CI's interop job (pinned to a commit SHA)

4. Infrastructure

System Role
AWS SNS / SQS Topics, queues, subscriptions (RawMessageDelivery=true) and RedrivePolicies, provisioned by infrastructure
PostgreSQL The outbox and inbox tables, in each service's own database; RLS policies are the service's
Prometheus / OTel collector Scrape and collect through the service's own endpoints; the library opens no listener
PyPI · GHCR Host the package (Trusted Publishing) · the signed example-worker image

5. Development tooling

The dev dependency group (uv sync installs it by default) has pytest, pytest-asyncio, pytest-cov, pytest-timeout, testcontainers[postgres], opentelemetry-sdk, ruff, mypy, types-aioboto3, pip-audit, pyyaml, griffe (make api-compat) and import-linter (make arch-lint). The build backend is pinned and hash-locked in build-constraints.txt.

Out of scope

Concern Where it lives
Topics, queues, subscriptions, filter policies, RedrivePolicies, IAM, KMS Platform infrastructure
Authenticating producers and deciding their tenant Each service's auth middleware (fastapicommon) and IAM
RLS policies, database roles, the pool Each consuming service · platform-pgcommon-py
Payload schemas and their registry The producing service · EVENT_SCHEMA_GOVERNANCE.md · your Codec
HTTP app factory, request middleware, logging and tracing setup platform-fastapicommon / the service
Alert routing, dashboard hosting The platform's monitoring stack (the rules and dashboard are vendored here)
Metric naming ratification Observability governance
Brokers other than SNS / SQS Not supported without a new adapter (CONTRIBUTING.md)
Synchronous (non-asyncio) services Not supported

Contributing

See CONTRIBUTING.md for the development setup, project layout, adding event types and adapters, changing the outbox or inbox schema, adding a metric, testing requirements, the PR checklist and the release process. Before pushing:

  • run make ci;
  • update CHANGELOG.md for any change to src/, pyproject.toml, uv.lock, build-constraints.txt or the Dockerfile (changelog-check.yml enforces it);
  • keep this README's API, configuration and error tables in step with the code;
  • edit a docs/architecture/mermaid/*.mmd file together with its embedded copy in ARCHITECTURE.md (make docs-check);
  • change wire-format, schema, metric or log behaviour in platform-events first, then here (make go-parity).

Glossary

Term Meaning in this library
Envelope The typed event wrapper (Envelope[T]) and its JSON wire format, identical to Go's. id, type, source, time are always present
Event / message / outbox record The domain occurrence and its envelope · the SNS/SQS unit the broker delivers · the Postgres row in outbox_events / outbox_dead_letters
Outbox The outbox_events table and the Runner that publishes from it; enqueue writes to it inside the business transaction
Claim lease The time a runner hides a claimed record from other runners (claim_lease_duration, 10m), by moving scheduled_at forward
Inbox The processed_events table that records which envelope IDs a consumer has handled (inbox.Store)
DLQ The SQS dead-letter queue named by the source queue's RedrivePolicy. Not the same as outbox dead letters (Postgres)
Malformed A body that is not an envelope by Go's rules; dead-lettered or deleted on first receipt, never retried
Poison message A valid envelope that can never be processed successfully; return (and log) or dead-letter it, never raise
Transient / permanent failure A failure that may succeed on retry (throttling, 5xx, timeouts; no outbox attempt counted) · one that will not (counts an attempt, backs off, dead-letters)
GUC A PostgreSQL runtime parameter. The consumer binds app.tenant_id through pgcommon for RLS policies to read
SYSTEM_TENANT_ID "system": the tenant of genuinely cross-tenant events. A real GUC value, not an RLS bypass
Panic Go's term; here a BaseException that is not an Exception (other than CancelledError, KeyboardInterrupt, SystemExit), recovered as Go recovers a panic

License / ownership

BCBP Solutions FZC LLC — internal platform shared library. Licensed Apache-2.0 (LICENSE).

See also

Document Description
ARCHITECTURE.md Layer model, flows, failure model, invariants, threat model, divergences from Go, design decisions
docs/lld/platform-lld-eventcommon-py.md Low-level design: API contract, algorithms, failure handling, invariant IDs (EVPY-*, mapped to the Go LLD's), sign-off register
docs/architecture/README.md The standalone Mermaid sources of every ARCHITECTURE.md diagram
docs/guides/ Task guides: quick start, publishing, consuming, outbox, envelope, codec, HMAC, observability, operations, testing in services
docs/observability/README.md · metrics-registry.md · runbook.md The observability standard as implemented here · the generated metric inventory · the alert runbook
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, verifying a release
CHANGELOG.md Per-version changes, with upgrade notes
CONTRIBUTING.md Development setup, adding event types and adapters, PR checklist
SECURITY.md Vulnerability reporting, trust model, supported versions (the root SECURITY.md points here)
.claude/CLAUDE.md Agent reference set for Claude Code (with architecture.md, api-reference.md, database-schema.md, request-flows.md, operations.md, settings.json): commands, critical rules and condensed contracts

eventcommon must be the only entry point for event publishing and consumption, to preserve delivery guarantees, tenant isolation and uniform observability. For any event that drives downstream state, use outbox.enqueue inside a pgcommon.run_in_tx block, with no exceptions.

Project links

Data verified by PyPI on Oct 22, 2026
Data provided by the project maintainers, verified at the time the release was uploaded to PyPI.

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Latest release

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for bcbp_admin from gravatar.com bcbp_admin

Credits

Author: BCBP Solutions FZC LLC

GitHub Statistics

Data verified by PyPI on Oct 22, 2026
The GitHub source repository was provided by the project maintainers and verified by PyPI at the time of upload. Stars, forks, and open issues/PRs are derived from that repository and have not been independently verified.

Provenance

Data verified by PyPI on Oct 22, 2026
Data provided by the project maintainers and verified by PyPI at the time of upload.

Full provenance

All 2 files in this release have provenance metadata.

View provenance details

License expression

Apache-2.0
View SPDX License List

Requires

Python >=3.12

Classifiers

Metadata

Project links

Data verified by PyPI on Oct 22, 2026
Data provided by the project maintainers, verified at the time the release was uploaded to PyPI.

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Latest release

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for bcbp_admin from gravatar.com bcbp_admin

Credits

Author: BCBP Solutions FZC LLC

GitHub Statistics

Data verified by PyPI on Oct 22, 2026
The GitHub source repository was provided by the project maintainers and verified by PyPI at the time of upload. Stars, forks, and open issues/PRs are derived from that repository and have not been independently verified.

Provenance

Data verified by PyPI on Oct 22, 2026
Data provided by the project maintainers and verified by PyPI at the time of upload.

Full provenance

All 2 files in this release have provenance metadata.

View provenance details

License expression

Apache-2.0
View SPDX License List

Requires

Python >=3.12

Classifiers

Release files for eventcommon 0.4.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for eventcommon 0.4.0
File Size Uploaded Provenance
eventcommon-0.4.0.tar.gz 799.1 kB With provenance Details

Built distribution (wheel)

Table of built distributions (wheels) for eventcommon 0.4.0
File Interpreter ABI Platform
eventcommon-0.4.0-py3-none-any.whl 235.5 kB With provenance Python 3 none any Details

Total release size: 1.0 MB

Release files / eventcommon-0.4.0.tar.gz

Download URL eventcommon-0.4.0.tar.gz
Size 799.1 kB
Tags Source
SHA-256 checksum
How to use checksums 		3969b4261011323cc88178e11a28b6355cf883702154d372246c9be3dd44b670
BLAKE2b-256 checksum
How to use checksums 		76a05215a414c044a4f20af7c157a5da7e5a1556f273fba02b0fe2db0d653e38
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing? 		Yes
Uploaded via twine/7.0.0 CPython/3.13.16

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Oct 22, 2026.

Transparency log
Identity
Publishing platform
GitHub Actions
Publishing repository
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon
Publishing commit
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon/tree/4073cb55ee384454371f6524f401c339e0ac61e7
Workflow
Publishing configuration
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon/blob/4073cb55ee384454371f6524f401c339e0ac61e7/.github/workflows/release.yml
Publishing logs
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon/actions/runs/39762984527/attempts/1
Artifact
Subject
eventcommon-0.4.0.tar.gz
SHA-256 checksum
3969b4261011323cc88178e11a28b6355cf883702154d372246c9be3dd44b670 Verifying attestations

Release files / eventcommon-0.4.0-py3-none-any.whl

Download URL eventcommon-0.4.0-py3-none-any.whl
Size 235.5 kB
Tags Python 3
SHA-256 checksum
How to use checksums 		a7e6586fb37a5e022c87b24b3d7bd15c99c771ab2c14bb9564a40c2d2d1dcd0c
BLAKE2b-256 checksum
How to use checksums 		fdfe42cb44cb7a879ad77f5cbb3dd61292e3bfe69dfce45e9d13766f90bbd6aa
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing? 		Yes
Uploaded via twine/7.0.0 CPython/3.13.16

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Oct 22, 2026.

Transparency log
Identity
Publishing platform
GitHub Actions
Publishing repository
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon
Publishing commit
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon/tree/4073cb55ee384454371f6524f401c339e0ac61e7
Workflow
Publishing configuration
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon/blob/4073cb55ee384454371f6524f401c339e0ac61e7/.github/workflows/release.yml
Publishing logs
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon/actions/runs/39762984527/attempts/1
Artifact
Subject
eventcommon-0.4.0-py3-none-any.whl
SHA-256 checksum
a7e6586fb37a5e022c87b24b3d7bd15c99c771ab2c14bb9564a40c2d2d1dcd0c Verifying attestations

Security for eventcommon 0.4.0

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

Full provenance

All 2 files in this release have been uploaded with provenance metadata.

Select from 2 files

Attestation for eventcommon-0.4.0.tar.gz

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Oct 22, 2026.

Transparency log
Identity
Publishing platform
GitHub Actions
Publishing repository
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon
Publishing commit
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon/tree/4073cb55ee384454371f6524f401c339e0ac61e7
Workflow
Publishing configuration
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon/blob/4073cb55ee384454371f6524f401c339e0ac61e7/.github/workflows/release.yml
Publishing logs
github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon/actions/runs/39762984527/attempts/1
Artifact
Subject
eventcommon-0.4.0.tar.gz
SHA-256 checksum
3969b4261011323cc88178e11a28b6355cf883702154d372246c9be3dd44b670 Verifying attestations

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page