Capability Matrix — what this port does, deliberately doesn't, and how to fill the gap
Benzene is honest about its boundaries. This page is the single place that states, for each common production concern, what the Python port provides (with the code that provides it), what it deliberately does not do (and why), and how to solve the rest outside Benzene. Package layering and install guidance live in Packages & adoption levels; the operational-requirements mapping (the profile's R1–R5 and R7, plus the meshed-service claim's M1/M2, with the test that proves each) lives in Cloud Service Profile conformance. This page consolidates both, capability-first, and adds the rows they don't state: what is absent.
The one idea behind every row
Benzene abstracts at the business-logic boundary — you write a message handler once and host it anywhere — and never at the transport or storage boundary. Wrapping SQS or Service Bus behind a generic "queue" interface would hide the transport-specific capabilities that were the reason to choose it. So Benzene's answer to "does it abstract X?" is usually a deliberate no: every binding hands you the native event/SDK object, and rolling your own middleware is a first-class, supported path, not an escape hatch.
"Deliberately not" and "not yet" are different things. A row below that gives a reason is a design decision; a row that says not implemented is simply unbuilt (usually something a sibling port ships). Conflating the two is how a document like this loses your trust, so it doesn't.
The matrix
| Capability | What this port provides | What it deliberately does NOT do (and why) | How to solve the rest |
|---|---|---|---|
| Transport features | Full, direct access to the native event/SDK object on every binding (the Lambda event dict, the confluent_kafka message, the ASGI scope) |
Hide transport capabilities behind a generic queue/transport interface — the anti-pattern above | Use the raw SDK/event directly; every decode function and host exposes it |
| Core pipeline & routing | Handler registry + @message, middleware pipeline, per-invocation DI, transport-neutral envelope — benzene-core (registry.py, pipeline.py, router.py, envelope.py, dependencies.py) over the benzene-results status vocabulary. See core, results. |
Impose the Benzene envelope on transports that already carry routing (Kafka topic, message attributes) — routing stays native where the transport has it | For envelope-less triggers (S3, Blob, Timer) the binding supplies a preset/default topic you can override |
| Message versioning | (topic, version) registration; the version is the one benzene-version header (no version/x-version fallback; renamed in one place with AppDefinition(metadata_keys=MetadataKeys(version=...)), or BenzeneMessageApplication(version_header=...)), or an HTTP /v{version} route segment that supplies it exactly as the header would. Selection is the spec's (core-concepts §2): exact match, else not-found naming topic and version; a message with no version goes to the unversioned handler, else the one declared default=True (@message / Registry.register / HttpRouter.register); a topic left with only versioned handlers and no default warns at start-up (NoDefaultHandlerWarning). highest_version is an explicit opt-in selector, never the default — benzene-core (registry.py, router.py, envelope.py). Cast reachability (selection step 2) has no candidates here: payload casting is registration-level (casting_handler registers each version it serves), so the exact match finds it |
Content-negotiation or semver-range resolution — version selection is a small injectable function, not a policy engine | Supply your own VersionSelector to message_router |
| Transport metadata (topic sources, reserved names) | Every binding takes its topic from where topicSources says (transport-bindings §1 obligation 2): the reserved topic metadata key on SQS (Lambda and the self-hosted consumer), SNS, Pub/Sub, Service Bus, Event Hub, Kafka (self-hosted and Lambda MSK) and RabbitMQ — each falling back, when a record has no topic header, to the Kafka topic it was consumed from / the AMQP routing key, so a producer that writes no header still routes (the header always wins); EventBridge's detail-type, with no topic attribute required; DynamoDB Streams' derived {tableName}:{eventName} (the table from the record's eventSourceARN, the eventName verbatim, e.g. orders:INSERT); the HTTP route; the gRPC method. The remaining metadata becomes the headers (keys lower-cased, matched case-insensitively), and benzene-version is the version. The reserved names (topic, benzene-version, EventBridge's benzene-headers) are one MetadataKeys value, configured once on the AppDefinition (metadata_keys=): application_from puts it on the application, every metadata-carrying binding decodes with it, and the SNS/SQS/EventBridge/Kinesis/Service Bus/Event Hub/Pub/Sub/Kafka/RabbitMQ senders take the same value (metadata_keys=) — benzene-core (metadata.py, envelope.py, startup.py) and each binding's decoders and clients. The conformance runner drives metadataCases and overrideCases through every host and client of each binding, and each topicSources entry from its native shape |
Read a second, legacy topic key (benzene-topic) or a fallback the fixture does not name — a port that routes what a conformant peer does not drifts unnoticed |
Override the names on the AppDefinition and hand the same MetadataKeys to your outbound clients; pin a fixed DynamoDB topic with AwsLambdaApp(dynamodb_topic=...) |
| HTTP (ASGI) | Inbound ASGI binding, Benzene↔HTTP status mapping, outbound HTTP client — benzene-http (app.py, status.py, client.py). See http. |
Be a web framework (templating, sessions, static files) — it is one ASGI app your server of choice hosts | Run under uvicorn/hypercorn, or mount alongside a framework in your ASGI stack |
| gRPC | Benzene↔gRPC status mapping with the benzene-status trailer rule, structured errors as google.rpc.BadRequest on grpc-status-details-bin (each error's code in FieldViolation.reason, read back into code), server + client transport — benzene-grpc (status.py, details.py, server.py, client.py). See grpc. |
Bundle grpcio unconditionally — it is the [transport] extra, so the mapping is usable without the SDK |
pip install "benzene-grpc[transport]" for the real transport |
| Outbound error fidelity (core-concepts §12) | Every client that can receive a problem document rebuilds the Result's errors from it — errors authoritative and ordered, each message, field and code intact, else detail as one opaque message — through the one shared reader, benzene.results.problem_errors: HttpMessageSender (benzene-http) — the failure arrives as a route-mapped application/problem+json (or application/json) body, its status from the document's benzeneStatus when it states a failure (wire-contracts §4.1, the spec 1.1 SHOULD — so an application-defined status survives the 500 it travelled as), else from the reverse HTTP table (benzene.results.problem_status); InProcessMessageSender (benzene-core) and LambdaMessageSender in RequestResponse mode (benzene-aws) — from the failure envelope's body, via decode_response, whose status is the envelope's statusCode (the document's benzeneStatus stands in only for an envelope with none). A failure body that is not a problem document (an HTML error page, an empty body, JSON with neither errors nor detail) never fails the call (wire-contracts §4.1): the peer's status stands, with no errors, and in an envelope a JSON one rides along as the payload. run_client_error_fidelity in the conformance runner drives problem-details-cases.json's envelopeCases through all three clients, and GrpcMessageSender below, against hosts built from the canonical handlers. One-way, so no reply and no problem document: the SNS, SQS, EventBridge and Kinesis senders, LambdaMessageSender in Event mode, the Service Bus, Event Hub, Queue Storage and Event Grid senders, the Pub/Sub, Kafka and RabbitMQ senders, InProcessFanOutSender and the mesh's S3 trace-inbox sender. This port has no request-reply queue client. GrpcMessageSender receives the gRPC error model (wire-contracts §4.2), not a problem document: message, field and code all survive as BadRequest.FieldViolation's description, field (the dotted path turned back into a pointer) and reason (the spec 1.1 SHOULD; the server side writes code there too), and the same envelopeCases run through it over a loopback gRPC server. The decorators (retry, correlation id, trace propagation, the resilience senders, MeshFeedSender) and generated typed clients return the inner client's Result unchanged |
Attach the received problem document (the MAY of §12): a decoded Result's problem_document stays None, so an application's own problem type and extension members do not survive the hop. Not built: an HTTP client for a peer's /benzene/invoke envelope endpoint, because HttpMessageSender posts to routes |
Read an application type from the raw response through your own HttpTransport, or carry the distinction in an error code |
| Failed calls (transport-bindings §1) | An outbound call that fails before any reply — no connection, refused or reset, the client's own deadline — returns a Result, never raises (the spec 1.1 SHOULD): timeout when the deadline elapsed, otherwise service-unavailable, so retry and the circuit breaker treat a dead peer as the transient failure it is. One classifier, benzene.core.failed_call_result (with is_timeout: a builtin TimeoutError, 3.10's asyncio.TimeoutError, an SDK exception named for a timeout — botocore's ReadTimeoutError/ConnectTimeoutError, httpx's, azure-core's, google-api-core's DeadlineExceeded — or one caused by any of those, urllib's URLError reason included). HttpMessageSender: the transport signals no reply with an OSError (ConnectionError/TimeoutError; the stdlib transport's URLError is one, and it turns a garbled or truncated reply into ConnectionError), and timeout= bounds the whole call for any transport. GrpcMessageSender: timeout= is the gRPC deadline, DeadlineExceeded→timeout and Unavailable→service-unavailable through the §4.2 reverse table, and a call on a closed channel is service-unavailable (it used to raise ValueError). LambdaMessageSender and every one-way sender (SNS, SQS, EventBridge, Kinesis, Service Bus, Event Hub, Queue Storage, Event Grid, Pub/Sub, Kafka, RabbitMQ, the S3 trace inbox) already caught every SDK error as service-unavailable; a timeout among them is now timeout. Tested over real sockets in tests/test_http_client.py (refused, a server that never answers, one that hangs up) and tests/test_grpc_transport.py (unreachable, deadline, closed channel) |
Turn a defect into a Result: a transport that raises anything but an OSError (a bug, a bad url_for) still raises, as does a ValueError from gRPC other than the closed channel. The one-way senders' catch-all also maps an SDK's own reply (an access-denied, a throttle) to service-unavailable — unchanged |
Translate an SDK-backed HttpTransport's errors to TimeoutError/ConnectionError; wrap with with_retry / with_circuit_breaker to act on the transient statuses |
| Kafka (self-hosted) | Consumer loop + produce-side client, duck-typed against confluent-kafka (SDK optional) — benzene-kafka (consumer.py, producer.py); the topic is the topic header, else the Kafka topic the record was consumed from. See kafka. |
Ship a schema-registry codec (see the Schema registry row) or hide confluent-kafka's configuration surface |
Configure the Consumer/Producer yourself and hand them in |
| RabbitMQ (self-hosted) | Consumer + publish client, SDK optional — benzene-rabbitmq (consumer.py, producer.py); the topic is the topic header, else the delivery's AMQP routing key. See rabbitmq. |
Manage exchanges/queues/topology for you — that is infrastructure, declared with your own pika/management tooling |
Declare topology in your bootstrap or IaC; pass the channel in |
| AWS | Lambda host with API Gateway, SQS, SNS, S3, EventBridge, DynamoDB Streams (topic {tableName}:{eventName}), Kinesis, Kafka/MSK, and direct-invoke inbound (app.py, events.py); SNS/SQS/EventBridge/Kinesis/Lambda outbound senders (clients.py); a self-hosted SQS consumer (sqs_consumer.py); test builders/hosts (testing.py) — benzene-aws. See aws. |
Wrap boto3 — outbound senders take your configured client; DynamoDB/S3 as storage is your handler's own code (a database is not a transport) |
Use boto3 directly in handlers for anything beyond message delivery |
| Azure | Functions host with HTTP, Service Bus, Event Hub, Queue Storage, Blob Storage, Cosmos DB change feed, Timer, and Event Grid inbound; Service Bus/Event Hub/Queue Storage/Event Grid outbound clients — benzene-azure (app.py, events.py, clients.py). (Blob inbound is a binding some sibling ports defer; this port ships it — decode_blob_created / handle_blob.) See azure. |
Carry an application payload for notification-shaped triggers (Blob, Timer) — they name a thing rather than carry a message, so the descriptor is the body under a default topic | Override default_topic and fetch the blob/document with the Azure SDK in the handler |
| Google Cloud | Cloud Functions host (HTTP + Pub/Sub inbound), Pub/Sub outbound client — benzene-gcp (functions.py, pubsub.py). See gcp. |
Cover the rest of GCP's eventing surface (Eventarc, Cloud Tasks, …) — those bindings are not implemented, plainly | Decode the trigger payload into an envelope yourself and call app.handle |
| Health checks (R3) | HealthChecks + reserved benzene:healthcheck topic, answering ok whenever the checks ran with the verdict in isHealthy (wire-contracts §5); per-check status ok/warning/failed (a warning does not flip isHealthy), optional type/data; HTTP GET /benzene/health maps the verdict to 200/503 with the report as the body either way. The topic is intercepted by health_interception in the startup's middleware, or — one declaration for both faces — by the checks declared on StandardPaths(health=...), which application_from installs in the pipeline. The self-check grades R3 as the topic intercepted plus /benzene/health only where there is an HTTP surface, so a message-only service meets it with the topic alone — benzene-core (health.py, startup.py), benzene-http (app.py, standard.py), benzene-mesh (profile.py). No liveness/readiness topics or routes |
Probe infrastructure for you — a check is your async function | Write checks against your own dependencies; see cloud-service-profile.md |
| Spec endpoint (R5) — the Contract Document | One document, derived once at the composition root (AppDefinition(contract=ContractInfo(...)), application_from) from the registry, the OutboundRegistry's event and send declarations, the router's routes and the envelope endpoint, and served on both surfaces: the reserved benzene:spec topic (spec_interception, in the pipeline, so answered wherever the transport carries a reply) and GET /benzene/spec, which copies type/format onto a benzene:spec request and runs it through the pipeline (same authentication as every topic). {type?, format?}: absent/empty/null/{} = benzene/json; anything else answers not-implemented (HTTP 501). Shape per contract-document.md (spec 3b): schemaDialect: "openapi-3.0", service/serviceVersion/versionScheme/description, the deployment-side runtime ("python")/placement/profile/degraded declared on ContractInfo and outside both hashes (placement.cloud is configured, never detected, and region is never guessed), requests/events/sends, no benzene: entries, response absent for response_type=NoPayload, no synthesized examples, contractHash + registrationHash always present. Schemas in the OpenAPI 3.0 dialect (nullable, the allOf wrapper for a nullable named type), every dataclass, pydantic v2 model and str/int enum hoisted into components.schemas; required lists every dataclass field, a defaulted one included (the marshaler always writes every field, and §4.2 makes a property it always emits required even when the type declares a default). A pydantic model (duck-typed on its v2 API; benzene-core gains no dependency) is read as model_dump(by_alias=True) writes it: a property per field and @computed_field, named by its serialization alias, required unless Field(exclude=True) (absent) or exclude_if (optional), {} for a custom serializer or a stdlib dataclass nested in a model, a RootModel inline as its root value, generic parametrizations named like generic dataclasses. Values are written in their standard JSON forms by the one marshaler (to_jsonable, which a model's python-mode model_dump also goes through, so a dataclass and a model write one form) and derived to match: UUID {"type":"string","format":"uuid"}, date format: date, datetime format: date-time (always with an offset: a naive one is taken as UTC and written +00:00), timedelta an ISO 8601 duration (format: duration), time and Path strings, and Decimal a JSON number {"type":"number"} (§4.2's floating kind, as .NET writes decimal; not pydantic's JSON-mode string — exact for an integral value or up to 15 significant digits); Literal[...] by §4.2's enum row, {"type":<the values' type>,"enum":[…]} (a str/int enum member as its value, None making it nullable), {} only when the values have no single JSON type. A name collision fails building the app, naming both types, the topics and the remedy (@schema_name("…")) — benzene-core (contract.py, contract_schema.py, declarations.py, startup.py), benzene-http (app.py, standard.py, routing.py). The same document is the mesh register body (Mesh row). Passes derivationCases (with the case's serviceInfo identity and placement) — from the canonical dataclasses and again from the same types as pydantic models, to the same hashes — namingCases, specTopicCases, specHttpCases |
Hand-authored spec documents — the document derives from the registrations so it cannot drift from what is served; other types (OpenAPI/AsyncAPI) from this topic (use benzene-openapi). Emit validation constraints (minLength, pattern, … — a §4.2 MAY) from either dataclasses or pydantic Fields; describe a plain (non-str/int) Enum, a Literal of mixed JSON types, or a type the marshaler has no JSON form for ({}). Add an offset to a naive time — it is written as isoformat() gives it, and its schema is a plain string (no RFC 3339 time format) |
Declare ContractInfo on the AppDefinition |
| Contract hashes | contract_hash / registration_hash (contract-document §6: the allow-listed, dereferenced projection, annotations and format stripped, UTF-16 ordering, RFC 8785 via the rfc8785 package, SHA-256) in benzene-core (contract_hash.py) — the one implementation a service serves and a generated client embeds. No private or provider-side hash remains: the mesh poller reads the served registrationHash. Passes every contract-hash-cases.json case |
A second, port-private hash over another projection — that was the drift check's failure mode (F4) | Compare a client's embedded CONTRACT_HASH with the served contractHash |
| Envelope endpoint (R4) | POST /benzene/invoke: HTTP 200 for every envelope it processed whatever its status (deleted, not-found, an unauthorized from the pipeline's own auth included), the response envelope as the body and any inner problem document without status; 400 only for a body that is not an envelope. Runs the same pipeline as every other transport; StandardPaths(invoke_topics=...) restricts reachable topics (any in-supporting container), a restricted topic answering not-found — benzene-http (app.py, standard.py) |
Host-enforced carrier auth before the envelope is read (the 401/403 the spec allows) — there is no such layer in BenzeneHttpApp; put auth middleware in the pipeline, or in front of the ASGI app |
An ASGI middleware ahead of BenzeneHttpApp for carrier-level auth |
| Reserved topics | benzene:healthcheck (health_interception), benzene:spec (spec_interception, installed by application_from for a declared ContractInfo), the collector's benzene:mesh:* topics (there is no bare benzene:mesh topic since spec 3b), and benzene:ping — answered ok, no payload, by the message router on every transport with no wiring, so a queue health probe is never redelivered — benzene-core (router.py) |
Reject an application topic under benzene: (a port MAY; this one does not) |
— |
| Mesh (service-side) — the meshed-service claim, M1/M2 | The self-description is the Contract Document on benzene:spec (Spec endpoint row); the register body is that same document — registration_document(definition), identical to what benzene:spec serves, runtime/placement/profile/degraded included and no instanceId (mesh.md §2). MeshFeedSender sends registration/heartbeat/traces/issues; Heartbeat.for_document carries the document's registrationHash and serviceVersion (mesh.md §5); trace_middleware emits one TraceEvent per invocation and joins inbound W3C traceparent (keeping its tracestate; an invalid traceparent starts a fresh trace and drops it); with_trace_propagation forwards traceparent + tracestate on whatever MessageSender it wraps — sends (one-way or with a reply) and published events share that one port here, so M2 covers events once the event publisher is wrapped (tests/test_mesh.py::test_a_published_event_carries_the_trace_context_to_every_subscriber). Being meshed (mesh.md §7.1) is a Cloud Service plus M1 (the three feeds — the profile's former R6) and M2 (trace context joined and forwarded — the former R8); R6/R8 are retired. The self-check (evaluate_cloud_service_profile) grades the profile's R1–R5/R7 into missing — R3 being benzene:healthcheck intercepted in the pipeline plus /benzene/health where there is an HTTP surface, R7 met trivially with no HTTP surface, R5 being benzene:spec intercepted in the pipeline by a declared ContractInfo and answered wherever the transport carries a reply (a one-way binding never costs R5), plus /benzene/spec where there is an HTTP surface — and M1/M2 separately on report.mesh (mesh_missing, is_meshed — which requires the profile too, since a meshed service is a Cloud Service plus M1/M2; M1 needs the feeds declared and a ContractInfo to register, M2 the propagation declared); M1/M2 never reach profile.missing, and declare_cloud_service_profile(definition) puts only the profile verdict on the document's profile ({"name": "cloud-service", "missing"?}). Nothing in this port interprets another service's profile.missing (the collector keeps the document verbatim, the probe does not read it), so a stale R6/R8 from an older service is carried, never acted on. probe.py audits R1–R5/R7 — R5 on both halves, GET /benzene/spec and a benzene:spec envelope over /benzene/invoke, and R3 likewise, GET /benzene/health and a benzene:healthcheck envelope over /benzene/invoke — and reports M1/M2 separately on report.mesh as inconclusive (not observable from outside), never counted toward is_clean. No ServiceDescriptor, benzene:mesh topic or descriptorHash (removed in spec 3b) — benzene-mesh (feeds.py, trace.py, profile.py, probe.py). The registration is pushed by the host loop (examples/k8s_mesh/service/reporting.py, deploy/mesh/fleet/service.py), not a library announcer; degraded is only what the composition root declares, since this port's OutboundRegistry always determines events/sends. See mesh. |
Block or fail an invocation on trace/feed export — observers are lossy by contract | Point the feeds at a collector; see joining the mesh |
| Mesh collector | MeshCollector ingest/query surface — the register body is a Contract Document: requests are consumers, events ∪ sends providers (a publisher is a provider), benzene: entries ignored; only service is required (no schemaDialect or hash check — the carried registrationHash is trusted, never recomputed), and an absent or degraded member is unknown (no edges, no undeclared-edge contract-drift on it). Declared vs. observed (mesh.md §4.2): benzene:mesh:query:topic reports per declared edge providerActivity (callers, observed via trace parentage) and consumerActivity (handlers, observed via their own spans), each {service: {"lastObservedAt": <startedAt>}} — {} when never observed (a decommission candidate, never a boolean and never removed); a trace on a topic outside a registered service's requests (as handler) or events∪sends (as caller) files one contract-drift issue in benzene:mesh:query:fleet's issues ({fingerprint, classification: "contract-drift", service, topic, status: "", count, exemplarTraceIds}, fingerprinted on undeclared-edge). A reserved benzene: topic is exempt — a traced benzene:spec/benzene:healthcheck is never undeclared drift, since no document lists one (_flag_drift_if_undeclared, is_reserved_topic; tests/test_mesh_collector.py::test_a_traced_reserved_topic_is_never_undeclared_drift). Issue error detail (mesh.md §4.1, spec 1.2): an issue's errors ([{field?, code?, count}]) merges by (field, code) identity (an empty string is an absent member) with count += delta, bounded to 10 (count descending, then field, then code, ordinal, absent first) and reported on benzene:mesh:query:fleet's issue only when it holds any; an entry that is not an object, has neither member or a non-string one, or whose count is not a positive integer (JSON true, 1.5 and 2.0 included) is skipped, the issue kept (decode_issue_errors, merge_issue_errors). The emitter side is IssueAggregator.record(..., errors=result.errors): it counts each error's (field, code) — never message; an error with neither is not counted — per issue as a delta, caps a window at 100 distinct pairs, and bounds and orders at flush(). There is no issue middleware in this port: IssueAggregator is fed by hand, so a failure's detail reaches the feed only when the caller passes its errors. Flow status (mesh.md §4): benzene:mesh:query:fleet carries traces — the newest 20 flows, {traceId, events, services, failed, startedAt, durationMs, status?} — whose status is the earliest-started failing event's when the flow failed, else the earliest-started event's (start times compared as instants; equal ones keep receipt order; an untimed event sorts last; omitted when the deciding event has no status); durationMs is (latest event end) − (earliest start) as in benzene-go's meshd, never less than the longest single event, and a flow with no readable start has startedAt: "" and sorts last — the mesh UI's live plane formats both unconditionally. The old topics/produces body is not read. MeshPoller ingests each served /benzene/spec document as the registration unchanged, with a heartbeat carrying its registrationHash and serviceVersion. Catalog store (snapshot v4, keeps each registered document, published as specJson), issue detection, inbox — benzene-mesh (collector.py, specdoc.py, poller.py, store.py, issues.py, inbox.py, artifacts.py); containerised host + Terraform in deploy/mesh/. Passes mesh-collector-cases.json and mesh-issue-cases.json |
Persist the catalog durably out of the box — the shipped store is in-memory, rebuilt by polling. Not implemented: Service Versions — the catalog is keyed by service alone, so two side-by-side releases overwrite each other's registration and a heartbeat's hash is compared with the service's one registration (mesh.md §5's non-module rule); mesh-service-version-cases.json is an explicit opt-out |
Run the poller on a timer (the deploy does); back the store yourself if you need survival across restarts |
| Mesh UI hosting | The collector serves the canonical cross-language mesh-ui.html at /benzene/mesh-ui/ (transport-bindings §1; mount= to move it), projecting its catalog into the mesh-ui artifacts — benzene-mesh (artifacts.py, s3_artifacts.py, blob_artifacts.py), wired in deploy/mesh/. See observing the mesh. |
Fork or restyle the UI per-port — the page is vendored from the main repo's one canonical UI | Publish the artifacts to static hosting (S3/Blob writers ship) if you don't run the collector's server |
| Mesh fleet | Service-discovery adapters (AWS Cloud Map/Lambda tags, Azure resource group, Kubernetes namespace) and trace mappers (Jaeger/Tempo/X-Ray), all SDK-lazy/injectable — benzene-mesh-fleet (discovery_adapters.py, mappers.py). See mesh-fleet. |
Ship or run a tracing backend — the mappers are pure field/units transforms | Wire the mapped documents into your backend's ingest API yourself |
| Conformance | The dependency-free runner (python -m tests.conformance_runner, also run by pytest as tests/test_conformance.py and granular per-fixture tests) opens status-vocabulary.json, http-status-mapping.json, grpc-status-mapping.json, envelope-cases.json, problem-details-cases.json (all three groups, with envelopeCases also read back through every reply-receiving outbound client — see the Outbound error fidelity row), transport-metadata-cases.json (every topicSources entry from the binding's native shape — the Kafka consumed-topic and RabbitMQ routing-key fallbacks, on every host, EventBridge's detail-type, DynamoDB Streams' {tableName}:{eventName}, the HTTP route, the gRPC method; every metadataCases case, headersExclude included, through every host of each metadata-sourced binding — Lambda SQS and the self-hosted SQS consumer, SNS, Pub/Sub, Service Bus, Event Hub, the self-hosted Kafka consumer and Lambda MSK, RabbitMQ — with every overrideCases case configured on the application; and every outbound client of those bindings writing the case's metadata back under the same keys (tests/conformance_bindings.py); + settlementDefaults against every host's declared constants), contract-document-cases.json (producer groups derivationCases — run twice, from the canonical dataclasses and from the same types as pydantic models — /namingCases/specTopicCases/specHttpCases and client-generator groups parseCases/topicScopeCases/schemaClosureCases), contract-hash-cases.json (every case, projections included), mesh-trace-cases.json, mesh-collector-cases.json and mesh-issue-cases.json. run_fixture_coverage fails the run if any vendored fixture goes unopened. The "requires": "validation" case and the pydantic derivationCases run are reported skipped without pydantic, the gRPC client's error-fidelity run without the [transport] extra, the gRPC method source without grpcio, and the Service Bus / Event Hub senders' write-back without the Azure SDK extras. Pin: conformance/SPEC_VERSION = spec-1.2.0, checked against the tag by .github/workflows/conformance-drift-check.yml |
Run mesh-service-version-cases.json (needs a (service, serviceVersion) catalog key) or mesh-version-order-cases.json (needs a version comparator) — both are conditional fixtures this port does not claim, listed with their reasons in UNRUN_FIXTURES in tests/conformance_runner.py. mesh-descriptor-cases.json no longer exists (spec 3b) |
Claim Service Versions in the collector to take on both opt-outs |
| Codegen / typed clients | The benzene-codegen CLI: typed client from any service's Contract Document — rejects an unknown (or missing) schemaDialect, ignores benzene: entries, reads sends, types an absent response as Result[None] and the nullable wrapper as T \| None; schema closure; the embedded CONTRACT_HASH is benzene-core's contractHash recomputed over the client's projection (whole service = the served value; include-list; topic-scoped) — benzene-codegen-client. Passes parseCases, topicScopeCases, schemaClosureCases and the projection hash cases. See codegen-client. |
Runtime client generation or remote-registry lookups — it is a build-time tool; generating reserved-topic methods (--include-reserved is gone) |
Run it in CI against the target's /benzene/spec |
| OpenAPI | OpenAPI 3.1 document derived from the handler registry — benzene-openapi (generator.py). Payload schemas are benzene.core.json_schema, which describes dataclasses, pydantic v2 models (as model_dump(by_alias=True) writes them: aliases, computed fields, exclude, a RootModel as its root value) and str/int enums; required is what the caller must supply (tests/test_openapi_pydantic.py). See openapi. |
Serve Swagger UI — document generation only | Host the JSON wherever your docs tooling wants it |
| Observability (OTel) | Export the port's existing mesh traces through the OpenTelemetry SDK (SDK optional), plus response-as-event middleware with a pluggable ResponseEventSink — benzene-otel (exporter.py, response_events.py). See otel. |
Sampling/backend configuration (Jaeger, X-Ray, …) — Benzene exports via the standard OTel API, it doesn't ship a backend | Configure the OTel SDK/exporter in your composition root |
| Caching | Cache-aside (CacheAside) over a narrow async Cache port, with in-memory and Redis backends — benzene-cache (aside.py, memory.py, redis.py). See cache. |
Invalidation strategies, distributed cache coherence — TTL + explicit eviction is the whole model | Implement Cache over your own store for anything richer |
| Validation | Pydantic request validation as an optional adapter (validated(Model) decorator → validation-error with one structured error per bad field, its field an RFC 6901 JSON Pointer into the request as serialized — /items/0/sku, using the model's aliases; omitted for a @model_validator rule) — benzene-pydantic (validation.py). gRPC carries the pointer as FieldViolation's dotted path (items[0].sku) and reads it back as a pointer — benzene-grpc (details.py). See pydantic. |
Make pydantic a core dependency — validation is an adoption choice, so the core never imports it | Or validate by hand in the handler; a plain dict request is always allowed |
| Resilience | Outbound: with_retry (status-aware, pluggable backoff) + correlation/trace decorators over the one MessageSender seam — benzene-core (outbound.py). Inbound: circuit breaker, bulkhead, rate limiting as middleware — benzene-resilience. See resilience. |
Re-implement a policy toolkit (the Polly analogue) or hide one behind its own abstraction — policies are small composable decorators/middleware you can read in one screen | Compose the decorators (with_retry(with_circuit_breaker(sender))) or write your own middleware |
| Idempotency | Dedupe middleware keyed on an idempotency header, replaying the first result; only successes are remembered, and never a missing result (a failure the binding redelivers); pluggable IdempotencyStore port + in-memory store (single-process) — benzene-resilience (idempotency.py) |
Cross-instance de-duplication — independent processes (separate Lambda invocations, containers) can't coordinate at runtime; shared state can relocate the race, not remove it | Implement IdempotencyStore over a store with an atomic conditional write (DynamoDB conditional put, Redis SET NX), and design handlers to be naturally idempotent |
| Settlement (one-way bindings) | transport-bindings §1 obligation 8 on every one-way binding — see the per-binding breakdown. A message is acknowledged only when its invocation ended isSuccessful (read from the envelope, never the status text); a failure, a missing result, an unmatched topic and an uncaught error are redelivered. Every one-way host takes settlement=Settlement(at_most_once=..., ignore_unmatched_topics=...), both off by default — benzene-core (settlement.py), the hosts in benzene-aws, benzene-azure, benzene-gcp, benzene-kafka, benzene-rabbitmq. Each host declares SETTLEMENT_DEFAULTS, checked against the fixture by the conformance runner |
A cross-transport reliability abstraction — settlement stays transport-native, and dead-lettering is each transport's own redrive configuration | Make handlers on one-way bindings idempotent (Idempotency row); configure the transport's redrive / DLQ; use Settlement(...) for the exceptions |
| Sagas / workflows | In-process compensation-based saga with reverse-order rollback, Result-speaking — benzene-resilience (saga.py) |
Durable crash-resume — the saga is in-memory callables; nothing persists across a crash | For crash-durable or long-running workflows use a real orchestrator (Step Functions, Durable Functions, Temporal) started from a handler |
| AuthN / AuthZ | Partial. Basic auth (basic.py); bearer middleware + JwtValidator over PyJWT — verification against a configured static key (HMAC secret or RSA/EC public key) with algorithm/audience/issuer constraints (bearer.py); API Gateway custom-authorizer adapter (authorizer.py) — benzene-auth. See auth. |
Not implemented: JWKS fetch / OIDC discovery — there is no key-rotation-aware validation against an IdP's metadata (the .NET and Go ports ship this; not a design decision here, simply unbuilt) | validate is a plain callable: supply one that resolves signing keys from your IdP's JWKS (e.g. PyJWT's PyJWKClient) and pass it to bearer_token_interception |
| Outbox (produce-side durability) | Not implemented. The .NET port ships Benzene.Outbox; here the only trace is a seam note — ResponseEventSink (benzene-otel, response_events.py) can feed an outbox you build, but no store-and-forward, staging, or transactional write exists |
— (nothing deliberate here; this is unbuilt, not declined) | Treat every send as at-least-once and pair the consumer with the Idempotency row; or implement store-and-forward as middleware over your own durable store |
| Oversized payloads (claim check) | Not implemented. The .NET port ships Benzene.ClaimCheck (+ S3/Blob stores); this port has no offload/hydrate middleware or benzene-claim-check header handling |
— (unbuilt, not declined) | Keep payloads under the transport limit, or write an offload/hydrate middleware pair over S3/Blob carrying a reference header |
| Schema registry / wire codecs | Not implemented. No Confluent wire-format codec or schema-registry client (the .NET port ships both); the Kafka binding sends/receives plain envelopes | — (unbuilt, not declined) | Serialize with the confluent-kafka serializers yourself before/inside the producer you hand in |
| Database / state access | (nothing — by design) | Any database or state-store abstraction — a database is not a transport; wrapping one would hide its capabilities (the core anti-pattern) | Your handler uses boto3/azure-sdk/an ORM directly; Benzene delivers the event in, persistence is yours |
| Testing | In-memory test host + test doubles — benzene-testing; per-transport builders/hosts (e.g. AwsLambdaTestHost, SqsConsumerTestHost in benzene-aws/testing.py). See testing. |
Require cloud SDKs or emulators to test a handler — bindings are duck-typed/injectable precisely so tests run SDK-free | Use the builders; integration against real infrastructure is your CI's choice |
Settlement — the per-binding breakdown
What each one-way binding does with a message, from source (2026-10-11, after the Phase 1 settlement work). "Acknowledged" means the transport may forget the message; "redeliver" means Benzene declines the acknowledgement and the transport's own retry/redrive configuration decides what happens next.
- Every binding settles through
dispatch_one_way(benzene-core,settlement.py), which decodes the native message, runsBenzeneMessageApplication.handle, decodes the response envelope withisSuccessfulauthoritative (wire-contracts §1.2), and applies the binding'sSettlement. One rule, so the columns below cannot drift binding by binding. handleturns no result intounexpected-error(encode_response) and a malformed JSON body intobad-request; the router turns an unmatched topic — no handler at any version — intonot-found(orignored, underignore_unmatched_topics), a known topic at an unknown version intonot-found, and a handler exception intoservice-unavailable. The uncaught exception column is an exception escaping the pipeline (a middleware's, or the binding's decode step).- A success —
ignoredand an application-defined status marked successful included — is acknowledged everywhere. - Overrides, on every row:
Settlement(at_most_once=True)acknowledges whatever the outcome (a failure or an exception is logged at WARNING onbenzene.settlementand dropped);Settlement(ignore_unmatched_topics=True)acknowledges a topic no handler serves asignored. The self-hosted loops keep their manual knobs too (delete=False/commit=False/ack=False+on_result; RabbitMQrequeue=Falseto dead-letter), and Kafka addsstop_on_failure. - On the envelope-less sources (S3, Kinesis, Blob, Cosmos DB, Timer; EventBridge without a
detail-type; plain-text Queue Storage) the topic is a configured convention, so with no handler for it every delivery is an unmatched topic; DynamoDB Streams derives{tableName}:{eventName}, so an unhandled table or change type is.
| Binding | Package / entry point | Fixture id · shape | Failure / no result / unmatched topic | Uncaught exception | Unit of settlement |
|---|---|---|---|---|---|
| AWS Lambda SQS | benzene-aws · AwsLambdaApp |
sqs · queue |
redeliver (batchItemFailures by messageId) |
redeliver — the whole batch (the invocation fails) | the message; every record runs; needs ReportBatchItemFailures |
| AWS Lambda SNS | benzene-aws · AwsLambdaApp |
sns · one-shot |
redeliver (raises¹) | redeliver (raises¹) | the whole invocation; stops at the first failing record |
| AWS Lambda S3 | benzene-aws · AwsLambdaApp |
s3 · one-shot |
redeliver (raises¹) | redeliver (raises¹) | the whole invocation; stops at the first failing record |
| AWS Lambda EventBridge | benzene-aws · AwsLambdaApp |
eventbridge · one-shot |
redeliver (raises¹) | redeliver (raises¹) | the one event |
| AWS Lambda Kafka / MSK | benzene-aws · AwsLambdaApp |
kafka · stream (redeliver) |
redeliver (raises¹) | redeliver (raises¹) | the whole invocation, across every topic-partition in it; stops at the first failing record |
| AWS Lambda DynamoDB Streams | benzene-aws · AwsLambdaApp |
dynamodb-streams · stream (redeliver) |
redeliver (batchItemFailures by SequenceNumber) |
redeliver — the whole batch | the record; every record runs, and Lambda resumes the shard from the lowest reported failure, so the records after it are redelivered too |
| AWS Lambda Kinesis | benzene-aws · AwsLambdaApp |
kinesis · stream (redeliver) |
redeliver (batchItemFailures by sequenceNumber) |
redeliver — the whole batch | as DynamoDB Streams |
| AWS Lambda direct invoke, asynchronous | benzene-aws · AwsLambdaApp(invocation_type="Event") |
envelope-async · one-shot |
redeliver (raises¹) | redeliver (raises¹) | the one envelope. Lambda does not reveal the invocation type, so this is a configured mode; the default ("RequestResponse") is the synchronous envelope binding |
| AWS SQS, self-hosted | benzene-aws · SqsConsumerApp + run_sqs_consumer_loop |
sqs · queue |
redeliver — not deleted | redeliver — the loop raises; this and the rest of the received batch stay undeleted | the message |
| Kafka, self-hosted | benzene-kafka · KafkaConsumerApp + run_consumer_loop |
kafka · stream (skip-and-continue) |
skip-and-continue — not committed, but the loop moves on and the next commit on the partition passes it (redelivered only if the consumer restarts or rebalances first). stop_on_failure=True: redeliver — the loop raises without committing, and a restart resumes at the record |
redeliver — the loop raises; a restart resumes at the last committed offset | the record offset, cumulative per partition |
| RabbitMQ, self-hosted | benzene-rabbitmq · RabbitMqConsumerApp + run_consumer_loop |
rabbitmq · queue |
redeliver — basic_nack(requeue=True) (or requeue=False → dead-letter exchange) |
redeliver — the loop raises; the unacked delivery is requeued when the channel closes | the delivery |
| Azure Functions Service Bus | benzene-azure · AzureFunctionsApp.handle_service_bus |
servicebus · queue |
redeliver (raises¹) | redeliver (raises¹) | the message |
| Azure Functions Queue Storage | benzene-azure · handle_queue_storage |
queue-storage · queue |
redeliver (raises¹) | redeliver (raises¹) | the message |
| Azure Functions Event Hub | benzene-azure · handle_event_hub |
eventhub · stream (redeliver) |
redeliver (raises¹) | redeliver (raises¹) | the delivered batch; stops at the first failing event, and the batch is redelivered from its checkpoint |
| Azure Functions Cosmos DB change feed | benzene-azure · handle_cosmos |
cosmosdb-changefeed · stream (redeliver) |
redeliver (raises¹) | redeliver (raises¹) | the delivered batch; stops at the first failing document |
| Azure Functions Event Grid | benzene-azure · handle_event_grid |
event-grid · one-shot |
redeliver (raises¹) | redeliver (raises¹) | the invocation's event (a batch stops at the first failing event) |
| Azure Functions Blob Storage | benzene-azure · handle_blob |
blob-storage · one-shot |
redeliver (raises¹) | redeliver (raises¹) | the notification |
| Azure Functions Timer | benzene-azure · handle_timer |
timer · one-shot |
raises¹ — the run is recorded as failed for the host's retry policy; there is no message | raises¹ | the tick |
| GCP Cloud Functions Pub/Sub | benzene-gcp · GcpFunctionsApp.handle_pubsub |
pubsub · queue |
redeliver (raises¹) | redeliver (raises¹) | the message |
Request/response bindings settle by responding (request-response, no default): http (the ASGI
binding and every cloud host's HTTP trigger), envelope (POST /benzene/invoke, and a synchronous
Lambda direct invoke), grpc.
¹ raises — the binding raises MessageHandlingError (benzene-core, errors.py; it names the
topic, the status and the problem document), or lets the exception propagate, out of the trigger
function. Whether and how often the platform then redelivers — Lambda's async retries and on-failure
destinations, the Functions host's retry, poison-queue and dead-letter handling, a Cloud Function's
retry-on-failure setting — is trigger configuration Benzene does not set and this audit cannot verify
from source. For the checkpointing Functions triggers (Event Hub, Cosmos DB) and Pub/Sub-triggered
functions in particular, confirm retries are enabled.
No binding, so nothing to settle (not implemented): a self-hosted Azure Service Bus, Event Hub or Cosmos DB worker; an Azure Functions Kafka trigger; a self-hosted Pub/Sub subscriber.
Pinned by tests: tests/test_settlement.py runs one outcome table — success, ignored, an
application-defined success, a failure, a missing result, an unmatched topic, an unknown version, a
middleware exception — through every row above, by default and under both overrides, plus Kafka's
skip-and-continue and stop_on_failure; the per-transport tests (tests/test_aws_host.py,
tests/test_aws_parity.py, tests/test_aws_sqs_consumer.py, tests/test_kafka.py,
tests/test_rabbitmq.py, examples/azure_orders/, examples/gcp_orders/) keep their own cases; the
conformance runner checks each host's SETTLEMENT_DEFAULTS against transport-metadata-cases.json.
Where this differs from the .NET reference (each allowed by the spec)
- Stream triggers redeliver an unmatched topic and a missing result. .NET accepts an
unrouted/null outcome on Lambda Kafka and the Azure Event Hub trigger so it cannot replay a partition
forever; here both are failures, as obligation 8 states, so an unhandled topic re-fails the
invocation on every retry — set
ignore_unmatched_topicsfor a shared stream. - Lambda Kafka / MSK fails the whole invocation at the first failing record; .NET defaults to per-partition partial-batch reporting.
- Lambda DynamoDB Streams does not stop at the first failure. Every record runs and every failure is reported; .NET stops at the first failed record. Lambda resumes from the lowest reported sequence number either way.
- Lambda Kinesis and the Azure Cosmos DB trigger route per record. .NET treats both as fan-in streams; here each record or document is routed under a convention topic and its result settles it.
- The self-hosted Kafka loop defaults to skip-and-continue, as .NET's worker does; its
at-least-once mode stops the loop (
stop_on_failure=True) rather than .NET'sCommitOnlyOnSuccessoffset store, with the same effect on restart.
Why "we don't do that" is a feature
Every deliberate "no" above buys you something: you keep the full power of the tool you chose,
you're never blocked by a leaky abstraction, and the surface you depend on stays small. The three
not implemented rows (outbox, claim check, schema registry) and the auth partial are stated
plainly because they are gaps, not philosophy — a port that does less, said honestly, is still
aligned with Benzene; a port that claims more than it does is not. When you need something this
port doesn't ship, a custom middleware or a MessageSender decorator is the supported, documented
path — see core.