core.ingress_ledger module

Atomic, content-bound publication for inbound platform events.

Platform SDKs may deliver the same immutable message callback more than once. The ingress ledger binds that logical event to one deterministic trace ID and one Redis Stream entry. The binding and XADD happen in one Lua script, so a lost gateway response can be retried without creating a second turn.

class core.ingress_ledger.IngressPublishAction(*values)

Bases: str, Enum

PUBLISHED = 'PUBLISHED'
CACHED = 'CACHED'
exception core.ingress_ledger.IngressLedgerError

Bases: RuntimeError

Base class for durable ingress publication failures.

exception core.ingress_ledger.IngressPayloadConflict

Bases: IngressLedgerError

The same immutable platform event arrived with different content.

class core.ingress_ledger.IngressPublishResult(action, trace_id, stream_id, payload_fingerprint, identity_hash, trace_identity_hash, trace_key, history_key, active_key, legacy_untracked=False, lifecycle_expired=False)

Bases: object

Parameters:
action: IngressPublishAction
trace_id: str
stream_id: str
payload_fingerprint: str
identity_hash: str
trace_identity_hash: str
trace_key: str
history_key: str
active_key: str
legacy_untracked: bool = False
lifecycle_expired: bool = False
core.ingress_ledger.ingress_event_identity(envelope)

Return the immutable logical identity for a ledger-eligible event.

A platform, channel, message ID, and event kind are all required. Callers can use is_ledger_eligible() for synthetic/batch envelopes that do not have an immutable platform message ID.

Return type:

str

Parameters:

envelope (Mapping[str, Any])

core.ingress_ledger.is_ledger_eligible(envelope)

Whether the envelope carries a complete immutable platform identity.

Return type:

bool

Parameters:

envelope (Mapping[str, Any])

core.ingress_ledger.canonical_ingress_trace_id(envelope)

Derive the one trace ID assigned to an immutable inbound event.

Return type:

str

Parameters:

envelope (Mapping[str, Any])

core.ingress_ledger.ingress_payload_fingerprint(envelope)

Hash every logical ingress field, excluding transport-only values.

Return type:

str

Parameters:

envelope (Mapping[str, Any])

class core.ingress_ledger.IngressLedger(redis, *, ttl_seconds=691200, proof_ttl_seconds=604800, namespace='sg:ingress:v1:')

Bases: object

Publish one content-bound stream entry per immutable platform event.

Parameters:

  • redis (Any)

  • ttl_seconds (int)

  • proof_ttl_seconds (int)

  • namespace (str)

property ttl_seconds: int
key_for_identity(stream, identity)
Return type:

str

Parameters:
async publish(stream, envelope)

Atomically bind envelope and append its canonical stream entry.

Return type:

IngressPublishResult

Parameters:
async publish_unkeyed(stream, envelope)

Atomically append synthetic ingress using its durable trace as identity.

Synthetic events lack an immutable platform message ID, so callers must retain the generated trace ID to retry a lost Redis response. The event bus passes a mutable envelope; storing the generated value back into that object makes a same-object retry first-writer safe.

Return type:

IngressPublishResult

Parameters: