Benzene

benzene.mesh

Make a Python Benzene service a first-class citizen of a mesh: it registers its self-description with a collector, traces every invocation, and reports heartbeats, traces and issues. Everything here is optional and additive. Distribution: benzene-mesh (depends only on benzene-core).

pip install benzene-mesh

Overview

The mesh module implements the language-neutral mesh specification. Its wire shapes — the registration (a Contract Document), the Heartbeat, the TraceEvent, and the collector topics — are the cross-language mesh contract: a Python service and a .NET/Go/TypeScript one emit the same shapes and appear in the same mesh. It gives you independent capabilities, each of which you can adopt on its own:

Every feed is independent and optional on both sides. An unreachable collector, a failing exporter, or a missing endpoint must never affect service traffic — the module is built so a mesh feed can never break, slow, or block an invocation.

Registration — the Contract Document

A service describes itself once: the Contract Document its composition root derives from its handler registry (requests) and its outbound declarations (events, sends). benzene:spec serves it, and the benzene:mesh:register body is it (mesh.md §2) — so what the mesh registers can never drift from what the service serves. As of the mesh.md declared-graph rule, the registration alone is what puts an edge in the mesh's producer/consumer graph — the graph exists before a single message has flowed (mesh.md §4).

A service that registers a handler for a topic is that topic's consumer (requests); a service that publishes an event (events) or sends to a topic (sends) is its provider — standard pub/sub terminology (Kafka/SQS/SNS/EventBridge: the receiver is the "consumer").

from benzene.core import AppDefinition, ContractInfo, OutboundRegistry
from benzene.mesh import registration_document

definition = AppDefinition(
    registry=registry,                                   # what this service consumes
    outbound=OutboundRegistry().register_send("payments:capture", CapturePayment),
    contract=ContractInfo(
        "orders",                                        # the only required field
        service_version="1.4.2",
        version_scheme="semver",
        placement={"cloud": "aws"},                      # + "region" only where documented
    ),
)

document = registration_document(definition)            # == what benzene:spec serves
payload = document.to_payload()                         # the register body (a dict)
document.registration_hash                              # "sha256:..." (contract-document §6)

OutboundRegistry

Declares what a service puts on the wire — in benzene.core (contract-document §3.3), re-exported here — as explicit records with no handler, since nothing here receives. Two kinds, split by who owns the contract: register_event(topic, message_type) for a topic this service publishes and defines, register_send(topic, request_type, response_type=None) for a message to a topic another party defines (leave response_type off for one-way). The Contract Document lists them as events[] and sends[]; a collector reads both as provider edges. A port must not infer them from call sites or any other static analysis.

from benzene.core import OutboundRegistry

outbound = (
    OutboundRegistry()
    .register_event("order:placed", OrderPlaced)
    .register_send("payments:capture", CapturePaymentRequest)
)

Declaring one (topic, version) twice — as the same kind or as both — raises DuplicateOutboundRegistrationError at start-up; there is no kind-less register. No destination address, queue name, or topic ARN is needed: that's deployment configuration, orthogonal to the contract.

The Cloud Service Profile self-check

The Cloud Service Profile names six requirements (R1–R5, R7) a service must satisfy so any tool can pull its operational surfaces from it with no collector; being meshed (mesh.md §7.1) adds M1 (registration, heartbeats, traces) and M2 (trace context propagation) — the profile's former R6 and R8, ids that are retired. evaluate_cloud_service_profile grades a composition root's AppDefinition against them at wiring time — a self-assessment of what the setup provisioned, not a runtime probe — and declare_cloud_service_profile puts its verdict on the Contract Document's optional profile field, so benzene:spec serves it and the registration carries it: the fleet can see whether a running service claims the profile and, if not, which requirements it is missing.

from benzene.mesh import declare_cloud_service_profile, evaluate_cloud_service_profile

report = evaluate_cloud_service_profile(definition, mesh_feeds=True, trace_propagation=True)
report.is_conformant          # True when every profile requirement (R1–R5, R7) is satisfied
report.missing                # e.g. ["R3", "R7"] — the profile ids the wiring does not satisfy, in order
report.reason("R7")           # the explanation recorded for one requirement (R or M)
report.to_profile()           # {"name": "cloud-service"} (or {..., "missing": [...]})
report.mesh_missing           # e.g. ["M2"] — the meshed-service claim, reported separately
report.is_meshed              # the profile in full, plus M1 and M2

definition = declare_cloud_service_profile(definition)
# definition.contract.profile == report.to_profile(); served on benzene:spec and registered alike

CloudServiceProfileReport carries one RequirementCheck(id, satisfied, reason) per requirement on checks (REQUIREMENT_IDS lists them: R1–R5, R7) and on mesh (MESH_REQUIREMENT_IDS: M1, M2); PROFILE_NAME is "cloud-service". See the runnable mesh_dashboard profile example.

The live-probe checker — probe_cloud_service

The outside-in counterpart: point it at a deployed service's base URL and it audits the profile over plain HTTP, speaking only the language-neutral surfaces (/benzene/spec, /benzene/health, /benzene/invoke) so it grades a Go, Node, or .NET service exactly as a Python one — it reads the Contract Document every port's /benzene/spec answers with. Each requirement gets a tri-state Verdict — satisfied / not-satisfied / inconclusive — always with a reason.

from benzene.mesh import probe_cloud_service

report = await probe_cloud_service("https://orders.example.com")   # inject http= in tests
report.not_satisfied      # profile ids positively found unmet — the actionable failures
report.inconclusive       # profile ids a black-box probe can't verify (never a failure by itself)
report.is_clean           # True when no profile requirement was positively found unmet
report.mesh               # the meshed-service claim (M1, M2), reported separately
report.to_payload()       # {"baseUrl", "requirements": [{"id", "verdict", "reason"}, ...], "mesh": [...]}

The probe audits R1–R5 and R7. R5 has two halves and both must hold: GET /benzene/spec returns a Contract Document, and a benzene:spec envelope over /benzene/invoke answers with one (either half positively unmet → not-satisfied). R3 is graded the same way: a benzene:healthcheck envelope over /benzene/invoke answers with the health report, and — the probed service having an HTTP surface — GET /benzene/health returns the aggregate. R7 is inconclusive whenever the caller probes a non-default prefix (the service's own defaults become unknowable). The meshed-service claim is reported apart on report.mesh and is not observable from outside, so both are inconclusive by design, never silently upgraded and never counted toward is_clean: M1 (registration, heartbeat and trace delivery to a collector — the register body is the document R5 already audits) and M2 (propagation needs a second service or a collector to observe forwarded traceparent). The CLI form is python -m benzene.mesh.probe <url> [--prefix /benzene] [--json], exiting non-zero when a requirement is positively unmet.

Tracing

trace_middleware() emits exactly one TraceEvent per routed invocation — the topic, the semantic status, how long it took, and its place in a W3C trace. Install it outermost (first in the pipeline) so it times the whole invocation, including routing.

from benzene.mesh import InMemoryTraceExporter, trace_middleware

exporter = InMemoryTraceExporter()
pipeline = MiddlewarePipeline().use(
    trace_middleware(exporter, service="orders", instance_id="orders-7f9c")
)

The middleware reads the inbound traceparent header to join an existing trace (or starts a fresh one), reads x-correlation-id for the business correlation id, times the pipeline, and hands the finished event to the exporter. Export must never affect the traffic it observes — exporter errors are swallowed, so a mesh feed can never break the request.

TraceEvent

One invocation's trace record. to_payload() is its camelCase wire form, omitting (never nulling) what isn't known.

TraceEvent(
    trace_id: str,
    span_id: str,
    service: str,
    topic: str,
    status: str,                         # the Benzene status, verbatim
    parent_span_id: str | None = None,
    instance_id: str | None = None,
    topic_version: str | None = None,
    exception_type: str | None = None,
    duration_ms: float | None = None,
    started_at: str | None = None,       # RFC 3339
    correlation_id: str | None = None,
)

Wire keys: traceId, spanId, service, topic, status (always present) plus parentSpanId, instanceId, topicVersion, exceptionType, durationMs, startedAt, correlationId when known.

TraceExporter and W3C helpers

Outbound propagation

trace_middleware records the current invocation's trace in a contextvar, so an outbound call made during the invocation can forward it (mesh.md §3):

Collector feeds

MeshFeedSender pushes a service's mesh feeds to a collector over an outbound benzene.core.MessageSender (Pub/Sub, SNS/SQS, Service Bus, or an HTTP POST of the wire envelope). Sending is fire-and-report: it returns the outbound Result so a caller can log a failed feed, but it does not raise, and a failing feed must never affect service traffic.

from benzene.mesh import Heartbeat, MeshFeedSender, registration_document

feeds = MeshFeedSender(sender)                       # any benzene.core MessageSender
document = registration_document(definition)         # the document benzene:spec serves

await feeds.register(document)                       # -> benzene:mesh:register
await feeds.publish_heartbeat(Heartbeat.for_document(
    document,                                        # service, serviceVersion, registrationHash
    sent_at="2026-07-31T12:00:00Z",
    instance_id="orders-7f9c",
))                                                   # -> benzene:mesh:heartbeat
await feeds.publish_traces(exporter)                 # -> benzene:mesh:traces  {"events": [...]}
await feeds.publish_issues(aggregator.flush())       # -> benzene:mesh:issues  (see below)

The collector topic constants and their bodies (the cross-language contract):

Constant Topic Body Success response
REGISTER_TOPIC benzene:mesh:register the Contract Document {"accepted": 1}
HEARTBEAT_TOPIC benzene:mesh:heartbeat Heartbeat {"accepted": 1}
TRACES_TOPIC benzene:mesh:traces {"events": [TraceEvent, ...]} {"accepted": <count>}
ISSUES_TOPIC benzene:mesh:issues IssueBatch {"accepted": <count>}

MeshFeedSender is the sender half (a service reporting in); the receiver is MeshCollector (below).

Heartbeat

A liveness beat: identity, the registered document's hash, and the health aggregate (mesh.md §5). registration_hash is the registrationHash of the document this instance registered — a collector whose registration carries a different one knows the instance runs a contract it hasn't learned (the instance should re-register). service_version scopes that comparison to one release in a collector implementing Service Versions.

Heartbeat(
    service: str,
    sent_at: str,                                    # RFC 3339
    instance_id: str | None = None,
    registration_hash: str | None = None,
    service_version: str | None = None,
    is_healthy: bool = True,
    health_checks: Mapping[str, Any] | None = None,
)
Heartbeat.for_document(document, *, sent_at, instance_id=None, is_healthy=True, health_checks=None)

to_payload() emits service, optional serviceVersion / instanceId / registrationHash, sentAt, and a health object {"isHealthy": ..., "healthChecks": {...}}. Prefer Heartbeat.for_document, which takes service, serviceVersion and registrationHash from the registered document so they cannot disagree with it.

Issues — IssueAggregator, Issue, IssueBatch

The issues feed reports deduplicated failure signatures. Two pieces are normative so a Python service produces the same signatures a .NET one would (the collector merges by fingerprint across instances):

IssueAggregator is the pit of success — record(...) each failure, flush() to an IssueBatch:

from benzene.mesh import IssueAggregator

issues = IssueAggregator(service="orders")
issues.record(topic="order:create", status="service-unavailable", version="v2",
              transport="sqs", exception_type="HttpError", trace_id=event.trace_id)
await feeds.publish_issues(issues.flush())           # count is a DELTA; flush() resets the window

flush() drains everything seen since the previous flush and resets, so every count is a delta, never a cumulative total. Flushing an empty aggregator is valid — that batch is the feed's liveness beat.

Error detail (mesh.md §4.1). Pass the failing result's structured errors and the issue says which field, which rule, not only how often:

issues.record(topic="order:create", status=result.status, errors=result.errors)
# -> "errors": [{"field": "/orderId", "code": "required", "count": 9}, ...] on the flushed issue

errors takes anything a failure factory does (BenzeneErrors, problem-document mappings, plain strings). Only each error's field and code are counted — never message, which is free text a validator may have built from the payload — and an error with neither is not counted; an empty string is the same as an absent member. Each (field, code) pair's count is a delta, like the issue's. At flush() an issue keeps at most MAX_ISSUE_ERRORS (10) entries as IssueErrors: the highest counts, ties broken by field then code (ordinal, an absent member first), listed in that order; a window admits at most 100 distinct pairs per issue, and a pair arriving after that is ignored. An issue with no detail omits errors. Neither member participates in the fingerprint. merge_issue_errors(existing, delta) is that merge-and-bound rule — the collector applies the same one — and decode_issue_errors(raw) the collector's lenient read of the wire member.

The collector — MeshCollector

A collector is the receiving side: an ordinary Benzene service that ingests the feeds and renders the fleet. collector_registry(collector) wires a MeshCollector onto a registry, so you run it through a BenzeneMessageApplication like any other service:

from benzene.core import BenzeneMessageApplication
from benzene.mesh import MeshCollector, collector_registry

app = BenzeneMessageApplication(collector_registry(MeshCollector()))
await app.handle({"topic": "benzene:mesh:register", "headers": {}, "body": document_json})
fleet = await app.handle({"topic": "benzene:mesh:query:fleet", "headers": {}, "body": "{}"})

It ingests benzene:mesh:register / :heartbeat / :traces / :issues and answers four read models — benzene:mesh:query:fleet / :service / :topic / :trace. The catalog it derives (per mesh.md §§4–6, pinned by mesh-collector-cases.json):

service is required on register, heartbeat, and issues (→ bad-request); an unknown service / topic / trace query is not-found; the query read models are one collector's shapes (the spec pins them only as the observable surface for the ingest rules). Sender feeds live in benzene.mesh (MeshFeedSender).

The optional issues feed is supported too: benzene:mesh:issues batches merge by fingerprint (count is a delta — occurrences accrue, exemplars accumulate), a malformed entry is skipped rather than rejecting the batch, and issues appears in a service's missingFeeds only when a failing trace is unexplained. An issue's errors detail merges by (field, code) identity (count += delta) and is bounded and ordered exactly as the emitter's is (above); an entry that is not an object, has neither member (an empty string being an absent one), has a non-string one, or whose count is not a positive integer (a JSON true and 1.5 included) is skipped, never costing the issue. The fleet view's issue reports errors only when it holds any. Conformance-green against both mesh-collector-cases and mesh-issue-cases.

Persistence — CollectorStore

A MeshCollector is in-memory by default, which is exactly right for tests and single runs. A long-lived collector (the Fargate Mesh Host) should not forget the whole fleet every time its task is replaced, so pass a CollectorStore:

from benzene.mesh import JsonFileCollectorStore, MeshCollector

collector = MeshCollector(store=JsonFileCollectorStore("/data/mesh-state.json"))

The collector restores the last snapshot on construction and saves a fresh one after every mutating ingest, so a restarted host rehydrates the fleet it already knew. CollectorStore is a small two-method Protocol (load() -> dict | None, save(dict)), so any backend fits; two ship:

The snapshot is a plain JSON-able dict — collector.snapshot() / collector.restore(snap) are public, so you can persist it anywhere (S3, a database) by implementing the two-method protocol over them.

The mesh-ui artifacts — build_artifacts / write_artifacts

The canonical, cross-language Benzene Mesh UI (mesh-ui.html, one page every port vendors) is data-driven from a fixed set of static JSON artifacts an aggregator publishes. benzene.mesh.artifacts projects a collector's catalog into that read-model contract (the main repo's docs/guides/mesh-ui.md, pinned by website/demos/mesh/):

from benzene.mesh import write_artifacts

write_artifacts("/data/mesh-ui", collector, sources=poller_sources, generated_at=now_iso)
# manifest.json, topology.json, topics.json, usage.json, asyncapi.json, annotations.json,
# and services/{name}.json — the UI fetches all of these by relative path.

The projection honours the contract's "must not invent fields, degrade when absent" rule. From the registration feed it derives the estate (health mapped to healthy/unhealthy/unreachable, contract-drift + previousSpecHash history), the functional map (topics with declared consumers/producers, benzene:* flagged reserved, request/response schemas, version, schemaMismatch when two consumers (handlers) disagree — schema authority stays with the handler registration — changes[] when a consumer re-registers a topic with a new schema, and removedTopics for a topic no longer consumed), per-service spec (the registered Contract Document, as specJson) + per-check health, and an AsyncAPI 3.0 export of the domain topics; the topology's client→server edges come from that same declared graph, and the trace feed layers each edge's error rate plus usage (exercise counts per topic/service/status) on top. annotations.json is an honest empty read-model (writing is a backend-gated live-plane feature). Only what genuinely needs feeds the collector doesn't have — latency/rate metrics, a usage time window, and transports — is emitted as null. The field set is pinned by tests/test_mesh_artifact_contract.py. See deploy/mesh for the host that serves them.

The poller — MeshPoller (pull aggregator)

MeshFeedSender is the push side (a service reports in). MeshPoller is the pull side, mirroring the .NET Mesh Host: it reaches out to a configured fleet on a timer, reads each service's /benzene/spec + /benzene/health (the StandardPaths surfaces), and folds the result into the same MeshCollector — so a service appears in the mesh with no egress wiring, just by being pollable.

from benzene.mesh import MeshCollector, MeshPoller, HttpServiceSource

collector = MeshCollector()
poller = MeshPoller(collector, [
    HttpServiceSource("orders", "https://orders.svc"),
    HttpServiceSource("inventory", "https://inventory.svc"),
])
await poller.poll_once()               # one sweep (call on a timer); collector now reflects the fleet
collector.query_fleet({})

Exports

registration_document, OutboundRegistry, EventDeclaration, SendDeclaration, DuplicateOutboundRegistrationError, MeshPoller, HttpServiceSource, CallableServiceSource, ServiceSource, PollResult, PollError, trace_middleware, TraceEvent, TraceExporter, InMemoryTraceExporter, QueueTraceExporter, parse_traceparent, new_trace_id, new_span_id, current_traceparent, current_tracestate, with_trace_propagation, TracePropagatingMessageSender, MeshFeedSender, Heartbeat, Issue, IssueBatch, IssueAggregator, IssueError, classify, issue_fingerprint, merge_issue_errors, decode_issue_errors, MAX_ISSUE_ERRORS, CLASSIFICATIONS, is_contract_document, spec_service, spec_service_version, spec_registration_hash, spec_entries, spec_topic_ids, known_members, resolve_schema, MeshCollector, collector_registry, CollectorError, CollectorBadRequest, CollectorNotFound, REGISTER_TOPIC, HEARTBEAT_TOPIC, TRACES_TOPIC, ISSUES_TOPIC, QUERY_FLEET_TOPIC, QUERY_SERVICE_TOPIC, QUERY_TOPIC_TOPIC, QUERY_TRACE_TOPIC, evaluate_cloud_service_profile, declare_cloud_service_profile, CloudServiceProfileReport, RequirementCheck, REQUIREMENT_IDS, PROFILE_NAME, probe_cloud_service, CloudServiceProbeReport, RequirementProbe, Verdict, build_artifacts, write_artifacts, CollectorStore, NullCollectorStore, JsonFileCollectorStore; and, for the cloud mesh deployments, S3CollectorStore, S3ArtifactStore / write_artifacts_to_s3, BlobArtifactStore / write_artifacts_to_blob, and S3TraceInbox.

Gone in 3b (pre-1.0): ServiceDescriptor, ServiceInfo, TopicDescriptor, MESH_TOPIC, mesh_interception, DescriptorSource, OutboundDefinition (and OutboundRegistry.definitions()), spec_topics, spec_produces, and the json_schema / Schema re-export (use benzene.core.json_schema).

See also

Edit this page on GitHub