Event Streaming Architecture

How events flow through the Virtufin API: CloudEvents over Dapr pub/sub, exposed to clients as gRPC streams.

Overview

Every event on the platform is a CloudEvents v1.0 message. The API is the only component that talks to Dapr pub/sub; other services publish and subscribe through the Pubsub gRPC service.

Mechanism	Direction	Entry point
Pubsub.PublishEvent	client → Dapr topic	unary gRPC / POST /v1/pubsub/{topic}
Pubsub.Subscribe	Dapr topic → client	server-streaming gRPC
Trigger.ScheduleTrigger	cron → Dapr topic	unary gRPC / POST /v1/trigger/{name}
State.SaveState(publish_change: true)	state write → state.change topic	unary gRPC

The wire type

Events are the CloudEvent protobuf message (proto/cloudevents.proto, package io.cloudevents.v1), not a bespoke envelope:

message CloudEvent {
  string id = 1;
  string source = 2;        // MUST be an absolute URI — see validation below
  string spec_version = 3;
  string type = 4;          // "ce-type"; discriminates events within a topic
  map<string, CloudEventAttributeValue> attributes = 5;  // optional + extension attributes
  oneof data {
    bytes binary_data = 6;
    string text_data = 7;
    google.protobuf.Any proto_data = 8;
  }
}

The Pubsub service surface (proto/pubsub.proto):

service Pubsub {
  rpc PublishEvent (PublishRequest) returns (PublishResponse);
  rpc Subscribe (SubscribeRequest) returns (stream SubscribeResponse);
  rpc Unsubscribe (UnsubscribeRequest) returns (UnsubscribeResponse);
}

message PublishRequest   { string topic = 1; io.cloudevents.v1.CloudEvent cloudevent = 2; }
message SubscribeRequest { repeated string topics = 1; }
message SubscribeResponse{ string topic = 1; io.cloudevents.v1.CloudEvent cloudevent = 2; }

SubscribeRequest filters by topic only. There is no service filter and no event-type filter on the wire; a subscriber that wants to react to a subset of a topic's events inspects cloudevent.type itself.

Publish validation

PublishEvent rejects, with InvalidArgument:

• an empty topic;
• a missing cloudevent;
• an empty cloudevent.source;
• a cloudevent.source that is not an absolute URI. "demo/notebook" is rejected; "https://virtufin.com/demo/notebook" and "urn:virtufin:demo" are accepted.

Nothing else about a topic is validated — see "Topic naming" below.

Topic naming

Topic names follow the <service>.<event_category> convention:

Topic	Publisher	Contents
websocketmanager.lifecycle	WebSocketManager	connection lifecycle
workmanager.lifecycle	WorkManager	worker lifecycle
state.change	the API itself	state mutations (see below)

The name is defined as a constant in the publishing service's own Configuration/Topics.cs and is the public contract between that publisher and its consumers. Consult each service's AGENTS.md for its event taxonomy.

This convention is producer-side discipline, not an API guarantee. The API does not validate topic names against the convention, does not maintain an allow-list, and does not scope topics to callers. Any authenticated caller may publish to, and subscribe to, any topic string — including another service's lifecycle topic and state.change. Do not rely on the topic namespace for isolation.

State change events

State.SaveState, State.DeleteState and State.DeleteBulkState accept a publish_change flag. When set, the API publishes a change notification after the write succeeds. Emitting one is opt-in per request — it is not a property of every state change.

Where that notification goes depends on what kind of state you wrote.

Infrastructure state → state.change (Tier 0)

Records whose owner is a platform service and whose meaning is deployment-scoped — worker registrations, connection records, leader leases. Leave change unset and the API publishes to the global state.change topic with:

• type = com.virtufin.api.state-saved or com.virtufin.api.state-deleted, so subscribers can filter on the action without parsing the payload;
• source = urn:com.virtufin.api, the publishing service.

{
  "service": "workmanager",
  "key": "workmanager.worker.abc123",
  "action": "save",
  "value": "the-state-value"
}

action is "save" or "delete". value appears only when the request sets include_value: true. DeleteBulkState publishes a keys array in place of key, and names only the keys actually deleted.

These events carry no version, and cannot. Dapr's state API returns no etag from a write, and re-reading the key to get one would be worse than omitting it: a concurrent writer may have advanced the value in between, so the event would attribute a newer version to an older payload. A subscriber consuming value instead of reading state back is therefore relying on writes to that key being serialized. Where you need a real ordering guarantee, use the Tier 1 path below.

state.change is one shared, unscoped topic, so include_value: true publishes that value to every subscriber of it.

Domain state → your own topic (Tier 1)

Records that are scenario- or universe-scoped — positions, portfolios, orders, risk and P&L artifacts. These do not belong on state.change: which service performed the write is an implementation detail, and subscribers route on the domain topic.

The API cannot build that event for you. It has the store, the key and the bytes, but not the scenario, the world markers, the clock type, or your service's identity. So supply the whole CloudEvent and the topic it belongs on, via the change field:

{
  "service": "workmanager",
  "key": "workmanager.position.BTCUSDT",
  "value": "{\"qty\":3}",
  "publish_change": true,
  "change": {
    "topic": "sc.LIVE.trading.position.updated",
    "cloudevent": {
      "id": "…",
      "type": "com.virtufin.position.flipped",
      "source": "urn:com.virtufin.trading-engine",
      "spec_version": "1.0",
      "text_data": "{\"symbol\":\"BTCUSDT\",\"qty\":3}"
    }
  }
}

The API publishes that event verbatim to that topic once the write lands, and does not also publish to state.change — publishing to both would double-deliver the change.

Set both topic and cloudevent. A change with only a topic falls back to the Tier 0 path: half a notification is not a Tier 1 event.

Because you control the envelope, this is where ordering lives: if subscribers consume your payload instead of reading state back, carry a monotonic token (a domain version, sequence number, or eventtime where the clock is authoritative) and have them discard anything not newer. Only you know how the entity is versioned.

Do not reach for a state.change.<state_name> topic. The Tier 0 pattern requires the category to be a domain noun rather than a state name, and per-key topics contradict the platform's cardinality rule. Domain state belongs on the Tier 1 taxonomy, which already has topics your subscribers subscribe to.

Delivery

Publication is at-most-once and is not atomic with the write. If the write succeeds and the publish fails, the event is lost — the API logs it and still reports the write as successful, because failing the call would make you repeat a write that already landed. Design subscribers to reconcile from state rather than assuming every change produced an event.

Measuring the gap

A lost event is otherwise invisible: no retry, no dead letter, and the caller sees success. Two counters make it countable:

Metric	Meaning
virtufin_api_state_change_publishes_total	notifications attempted, after a successful write
virtufin_api_state_change_publish_failures_total	of those, the ones lost

Both are tagged service, action (save/delete) and tier (tier0/tier1) — the tiers are separated because a lost Tier 1 domain event matters considerably more than a lost Tier 0 infra notification.

They are exported over OTLP only when OTEL_EXPORTER_OTLP_ENDPOINT is set. There is no collector in the deployment today, so until one exists the metrics are collected in-process and go nowhere, and the structured log is the measurement path:

State change publish failed after a successful write: service=… key=… topic=… tier=…

logged at Error from StateService. Counting those occurrences over a period is what should decide whether the at-most-once gap is worth closing with a transactional outbox — which would flip the guarantee to at-least-once and make every subscriber's idempotency a breaking concern. That trade is worth making for a frequent failure and not for a rare one, and nothing measured it before.

Subscribing

Pubsub.Subscribe is server-streaming: one call, many SubscribeResponse messages, one gRPC stream per client call regardless of how many topics it names.

using var call = pubsubClient.Subscribe(
    new SubscribeRequest { Topics = { "workmanager.lifecycle" } });

await foreach (var evt in call.ResponseStream.ReadAllAsync(cancellationToken))
{
    Console.WriteLine($"{evt.Topic}: {evt.Cloudevent.Type}");
}

Behaviors worth knowing:

• An empty topics list is not "all topics". It silently defaults to ["state.change"]. Name your topics explicitly.
• Duplicate topics in one request produce duplicate delivery of each event to that caller.
• Each subscriber buffers up to 1024 events. A consumer that falls further behind than that is treated as stuck and reaped by the health sweeper, so do slow per-event work off the read loop.
• One shared Dapr subscription per topic per process, reference-counted across callers (TopicDaprSubscriptionRegistry). This is deliberate: a Dapr subscription per caller would make daprd load-balance each event across callers instead of broadcasting it.
• Initial response headers are flushed before any event arrives. Request-reply clients depend on this to know the subscription is live before they publish; see below.
• Cancelling the call is the teardown. Disposing the streaming call releases the in-process subscriber and the shared Dapr subscription refcount.

Unsubscribe — do not use

Pubsub.Unsubscribe takes a subscription_id, but subscription ids are minted server-side and never sent to the client: SubscribeResponse carries no id field. There is no supported way to obtain a valid id, and calling it with a guessed one releases only the in-process subscriber while leaving the Dapr subscription and the server-side stream running. Cancel the Subscribe call instead.

Request-reply over pub/sub

Request-reply is implemented entirely in the client libraries. The API has no awareness of correlation ids or reply topics — it is a pure conduit, and the responder is always another service.

The requester stamps two CloudEvent extension attributes before publishing:

Attribute	Meaning
correlationid	opaque id, echoed back by the responder; the sole demux key
replytopic	topic the responder should publish its reply to

Both are lowercase with no separators, per the CloudEvents extension-naming rule. The requester holds one persistent Subscribe stream per reply topic, shared across all in-flight calls, and matches incoming events on correlationid. It waits for the stream's response headers before publishing, so a reply cannot be missed in the gap between publishing and subscribing.

Per-call subscribe/unsubscribe is not an option here: it churns the shared Dapr subscription, and daprd's Redis Streams component leaks the underlying blocking XREADGROUP connection on unsubscribe until the pool is exhausted.

Use PublishWithResultAsync (.NET), publish_with_result_async (Python), or publishWithResult (TypeScript) rather than reimplementing this.

Cron triggers

Trigger.ScheduleTrigger schedules a recurring job through Dapr's Scheduler. When it fires, DaprAppCallbackService publishes an empty CloudEvent to the trigger's target_topic (type = com.dapr.event.sent, source = urn:com.virtufin:api). The trigger carries no payload; subscribers react to the fact that it fired.

Configuration

The Dapr pub/sub component name defaults to "pubsub" and is configurable, in this precedence:

1. --pubsub-name (or the PubsubName setting);
2. pubsub.pubsubName in services.json, which helm renders from .Values.pubsub.name;
3. the "pubsub" default.

It is a gateway-wide setting despite living on each service entry. PublishEvent and Subscribe take a topic and no service, so there is nothing to resolve a per-service component against at call time. Every registered entry must therefore agree on the value — helm renders one value into all of them — and the gateway fails at startup if they disagree rather than silently picking one.

Subscriptions are created programmatically at runtime via Dapr's streaming-subscription API, not declared in YAML. The API's declarative subscription callback therefore returns an empty list by design.

Error handling

The pubsub path uses two different error channels, and this trips people up:

Failure	How it surfaces
Invalid input (empty topic, bad source)	thrown RpcException with InvalidArgument
Runtime failure (sidecar down, publish rejected)	gRPC status OK, with status.success == false

A client that only catches RpcException will treat a failed publish as a success. Always check response.Status.Success:

var response = await pubsubClient.PublishEventAsync(
    new PublishRequest { Topic = topic, Cloudevent = ce });

if (!response.Status.Success)
    throw new InvalidOperationException($"Publish failed: {response.Status.Message}");

On the subscribe side, handle stream termination as a normal condition and reconnect with backoff:

try
{
    await foreach (var evt in call.ResponseStream.ReadAllAsync(token))
        ProcessEvent(evt);
}
catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled)
{
    // Client disconnected — normal shutdown.
}

Client examples

C#

using var client = new ApiClient("localhost");

var ce = new CloudEvent
{
    Id = Guid.NewGuid().ToString(),
    Source = "urn:virtufin:my-service",
    SpecVersion = "1.0",
    Type = "com.virtufin.worker.started",
    TextData = """{"workerId":"abc123"}""",
};
await client.Gateway.PublishEventAsync("workmanager.lifecycle", ce);

using var call = client.Gateway.SubscribeToTopic("workmanager.lifecycle");
await foreach (var evt in call.ResponseStream.ReadAllAsync(token))
    Console.WriteLine($"{evt.Topic}: {evt.Cloudevent.Type}");

Python

async with ApiClient(host="localhost") as client:
    ce = cloudevents_pb2.CloudEvent(
        id=str(uuid.uuid4()),
        source="urn:virtufin:my-service",
        spec_version="1.0",
        type="com.virtufin.worker.started",
        text_data=json.dumps({"workerId": "abc123"}),
    )
    await client.pubsub.publish_async("workmanager.lifecycle", ce)

    call = await client.pubsub.subscribe_async(topics=["workmanager.lifecycle"])
    async for event in call:
        print(event.topic, event.cloudevent.type)

TypeScript

const client = new ApiClient({ host: "localhost" });

await client.publish("workmanager.lifecycle", cloudevent);

for await (const evt of client.subscribeToTopic("workmanager.lifecycle")) {
  console.log(evt.topic, evt.cloudevent?.type);
}

Best practices

1. Name your topics explicitly on Subscribe — the empty-list default is state.change, which is almost never what you want.
2. Check status.success on every publish; the gRPC status is OK even on failure.
3. Pass a cancellation token to Subscribe and cancel it to tear down; that is the only working teardown path.
4. Discriminate on cloudevent.type, not on the topic alone — one topic carries a service's whole event taxonomy.
5. Leave include_value off unless the state value is public to the whole platform.
6. Use the SDK's request-reply helper instead of hand-rolling correlation.