CloudEvents: one envelope for every event in Loams

Available Loams streams accept and serve CloudEvents in Loams 0.1.0, over HTTP and gRPC. Planned The Loams CloudEvents profile (tenant and trace attributes on every event) is a proposal that only Loams Git's WAL follows so far.

Webhooks, Dapr pub/sub, Knative, Debezium Server, Azure Event Grid and most serverless platforms already agree on what an event looks like, and the thing they agree on is CloudEvents, a CNCF specification. It does not define a transport. It defines a small set of attributes, and then, per transport, where those attributes go. That makes it the cheapest possible contract for "anything that sends events can send them to Loams".

What the protocol isCopy Anchor Link

The attributesCopy Anchor Link

An event has context attributes and optional data. Four attributes are required:

• specversion: 1.0.
• id: unique within the source. The spec says source plus id identifies an event, so a consumer may treat two events with the same pair as duplicates.
• source: a URI reference naming where the event happened, such as /github/acme/repo.
• type: what happened, typically reverse-DNS, such as com.github.pull_request.opened.

The optional ones are datacontenttype (the media type of data), dataschema, subject (what inside the source the event is about) and time (RFC 3339). Anything else is an extension attribute, whose name must be lowercase ASCII letters and digits (the spec recommends at most 20 characters). Common extensions include traceparent (distributed tracing) and partitionkey (the Kafka binding uses it as the message key).

Binary and structured modesCopy Anchor Link

Every protocol binding offers two ways to carry an event:

• Binary mode keeps the data as the message body and puts each attribute in a transport header. In HTTP, id becomes the ce-id header, type becomes ce-type, and so on, with datacontenttype mapped to the ordinary Content-Type. In Kafka, the headers are ce_id, ce_type, …, datacontenttype maps to the content-type header, and the record value is the data.
• Structured mode puts the whole event, attributes and data, into the body in an event format, such as JSON with Content-Type: application/cloudevents+json. In JSON, binary data goes in data_base64 instead of data.

HTTP also has a batched mode: a JSON array of events, Content-Type: application/cloudevents-batch+json.

Binary mode is what you want for throughput and for receivers that do not know about CloudEvents, because the body is untouched. Structured mode is what you want when an event must pass through systems that drop unknown headers.

Why Loams speaks itCopy Anchor Link

D184 made CloudEvents the envelope of the planned functions runtime's gateway for async events, and D270 followed: if events arrive as CloudEvents, streams should store and serve them natively, instead of every adapter normalizing events its own way. Two choices in D270 decide how it feels:

1. An event is an ordinary stream record, laid out as the Kafka binding's binary mode. ce_ headers, content-type, the key from partitionkey (else subject), the record timestamp from time, the value from data. Nothing in the log, the segmenter or links knows about CloudEvents, and a stream of events is also a valid Kafka topic of CloudEvents (see the Kafka post).
2. source plus id is the idempotency key. Native idempotent producers dedupe by producer sequence, which does not fit events from many independent senders. The spec already gives every event an identity; we use it.

D360, the standards charter, pins CloudEvents 1.0.2 behind one crate, so replacing or upgrading it touches one place.

How Loams implements itCopy Anchor Link

The codec: loams-cloudeventsCopy Anchor Link

A small hand-written crate, not cloudevents-sdk. We checked the SDK (0.9, Apache-2.0) and did not take it: it parses time into chrono and extension values into typed values and writes them back normalized, which breaks a byte-for-byte round trip; it had no protobuf format; and it pulled in chrono, url, uuid, hostname and snafu. Our CloudEvent keeps every attribute as the exact string it arrived with, in arrival order, plus each extension's type.

It implements the JSON format (structured and batched), HTTP binary mode, the record layout of the Kafka binding (feature record, on by default) and the protobuf format io.cloudevents.v1 (feature protobuf), which the gRPC service and Loams Git's WAL segments use.

IngestCopy Anchor Link

POST /v1/namespaces/{ns}/streams/{stream}/events takes all three HTTP modes; a request with neither a ce-specversion header nor a CloudEvents content type gets 415. StreamService.ProduceCloudEvents takes a protobuf CloudEventBatch or the batched JSON format as bytes.

Validation refuses, naming the attribute: a specversion other than 1.0 (so 0.3 events are refused), a missing or empty id, source or type, a time that is not RFC 3339, an attribute name that is not lowercase letters and digits, a context attribute of the wrong JSON type, or both data and data_base64. A batch is all or nothing.

Deduplication. The key is SHA-256(source ‖ 0x00 ‖ id), scoped to the stream, in a metastore ledger:

1. Claim every event's key in one metastore proposal. A key already done answers duplicate with the first write's partition and offset; a key pending under another request answers in_flight.
2. Append the claimed events in one append_many.
3. Complete: one proposal marks the keys done with their offsets, kept for the dedup window, 1 hour by default and at most 24 hours (--cloudevents-dedup-window).

A retry within the window never appends twice, and its answer names the original offset. HTTP answers 200 with a per-event status, or 409 with Retry-After if any event is in flight; retrying the whole request is always safe. The one duplicate case is a node that appended and died before completing: its claims lapse after 2 minutes and a later retry appends again. Nothing is lost in that case either.

Partitioning. The partition parameter if given, else Kafka's default partitioner (murmur2 of the key modulo the partition count) so a future Kafka front end agrees, else, for an event with no key, the idempotency key's hash, so a retry lands on the same partition.

ConsumeCopy Anchor Link

GET …/partitions/{p}/events takes the usual fetch parameters (offset, max_bytes, max_wait_ms) and mode=structured (the default, a batch of events) or mode=binary (one record as one binary-mode HTTP message). Events ingested as CloudEvents, or records produced with valid ce_ headers, come back with their attributes byte for byte. Any other record gets a synthesized envelope, with type io.loams.dev.stream.record and id {stream_id}-{offset}, so every stream can be read as CloudEvents. gRPC consume waits for gRPC fetch.

Where else it is usedCopy Anchor Link

• Dapr. deploy/dapr/edge passes Dapr's pub/sub and webhook CloudEvents through to ProduceCloudEvents unchanged, and relies on the source plus id dedup instead of its own invocation id, so a Dapr redelivery after a timed-out produce no longer appends twice. See the Dapr post.
• Loams Git. Each WAL record is a protobuf CloudEvent (D390, amended by D415), and committed transactions are mirrored into each namespace's reserved _git stream through the same ingest.
• Connectors. First-party sources emit CloudEvents typed io.loams.dev.flow.<connector>.<event>.v1, and sinks use ce_id as the idempotency key (D355). The Kafka connector reads and writes D270's binary layout.
• The Data API's live queries map each subscription event onto a CloudEvent (Data API post).

The Loams profile (proposed)Copy Anchor Link

D364 adds rules on top of the spec for every event Loams defines: type is io.loams.dev.<domain>.<name>.v<major> (D402), so a breaking change of data is a new type; dataschema is urn:loams:proto:<message>; time is required; tenantid (<org>/<namespace>) is required and stamped from the credential by the gateway or runner host, overwriting any client value; and traceparent is required, generated at the first hop if missing. Loams Git's WAL events carry tenantid and traceparent already. Stream ingest does not enforce the profile yet; it is still a proposal.

GotchasCopy Anchor Link

• Binary mode loses extension types. HTTP ce-* and Kafka ce_ headers are strings, so an integer extension ingested in binary mode reads back as a string. Loams records the types in a loams_ce_types header for structured JSON and protobuf ingest, so types survive those paths.
• Protobuf normalizes time. The protobuf format carries timestamps as seconds and nanos, so a protobuf read writes time in canonical UTC form, losing the original offset or trailing zeros.
• Producers must keep id stable across retries. Deduplication is only as good as the sender. A sender that mints a new id per attempt defeats it.
• The window is finite. A retry after the dedup window is a new event.
• The ledger is not free. Two small metastore proposals per request and about 80 bytes per event for the window suit webhook, trigger and agent event rates. Bulk ingest should use plain produce, which never deduplicates, even when its records carry ce_ headers (D365 describes the high-rate path).
• Percent-encoding. HTTP binary-mode header values are percent-decoded on ingest, as the HTTP binding requires.

Read nextCopy Anchor Link

• Kafka wire protocol, whose binary layout events are stored in.
• Connect RPC, which carries ProduceCloudEvents.
• Git smart protocol and remote helpers, whose WAL records are CloudEvents.