ChangeStreamService
CE Change Data Capture: tail a shard's Raft log via gRPC server-side streaming.
Client reconnects with a ResumeToken to continue from a position it chose. At-least-once delivery. History is released only up to a position the client acknowledges with AcknowledgeSubscription.
Proto source
Methods
Subscribe
Open a long-lived server-side streaming RPC that tails the Raft log.
The server streams a ChangeEvent for each log entry once the serving node has applied it; entries that are not committed yet are never sent. When caught up, the stream waits until the node applies another entry. The stream stays open until the client cancels or the connection drops.
On reconnect, pass the last received ResumeToken to resume without gaps. Omit resume_token (or set all fields to zero) to start from the oldest retained entry.
Every stream reads as a registered consumer: the request names the consumer and either registers it with an explicit retention policy (incarnation 0) or resumes a live registration (incarnation from the first stream). The registration outlives the connection: a STRICT one until CancelSubscription, a BOUNDED one until a declared bound is crossed. The response's initial metadata carries the registration's incarnation as x-coordinode-consumer-incarnation.
A token whose position the log no longer holds is refused with FAILED_PRECONDITION and ErrorInfo reason RETENTION_LOST (metadata requested_index, first_retained_index): the stream never resumes past a gap. A registration that has ended is refused with FAILED_PRECONDITION and reason CONSUMER_TERMINATED (metadata incarnation, reason, checkpoint); registering the id again starts a new incarnation.
Sending an event is not delivery: the registration keeps every entry from its acknowledged position on, so a stream resumed without a token starts there and re-sends what the client never acknowledged. A resume token chooses where reading starts and releases nothing.
Transport: Server-streaming gRPC only (no HTTP transcoding)
Request: SubscribeRequest
| Field | Type | Description |
|---|---|---|
resume_token | ResumeToken | Token from the last received ChangeEvent.position. Omit (all-zero) to start from the oldest retained entry. |
filters | CdcFilters | Optional filter to reduce bandwidth. |
consumer_id | string | The consumer this stream reads as: globally unique, chosen by the client. Required. |
incarnation | uint64 | 0 registers the consumer anew with retention, starting at resume_token (or the oldest retained entry without one); refused with ALREADY_EXISTS while another incarnation of the id is live. Any other value resumes that live incarnation, and retention is ignored. |
retention | ConsumerRetention | Required when registering (incarnation 0). |
Response: stream ChangeEvent
| Field | Type | Description |
|---|---|---|
ts | uint64 | HLC commit timestamp of the entry, in wall-clock microseconds since the Unix epoch with logical increments inside one microsecond: the latest commit timestamp of the transactions it carries. 0 for an entry that carries none (membership, leader blank). |
term | uint64 | Term of the leader that wrote this entry. |
log_index | uint64 | Raft log index. Monotonically increasing per shard. |
shard_id | uint32 | Shard that produced this entry. |
is_migration | bool | True if this entry was written during a shard migration (e.g. bulk data transfer). Most CDC consumers should skip migration entries. |
ops | ChangeOp[] | The mutations carried by this log entry. Empty for an entry that carries none, and for a progress event: entries up to position held nothing the stream's filters keep, so the client can acknowledge past them. |
position | ResumeToken | Where the stream continues after this event: pass it on reconnect to resume here, and to AcknowledgeSubscription once the event is held. |
AcknowledgeSubscription
State that the client holds, durably, every event of its subscription before position: the registration may stop protecting that history, and a stream resumed without a token starts at position. Only this call releases history; acknowledging an earlier position than a previous one changes nothing. Refused with INVALID_ARGUMENT for a position past what the serving node has applied (nothing that far was sent) or for another shard, and like CancelSubscription for an ended or unknown incarnation.
Request: AcknowledgeSubscriptionRequest
| Field | Type | Description |
|---|---|---|
consumer_id | string | The consumer acknowledging. |
incarnation | uint64 | Its incarnation, from the Subscribe response metadata. |
position | ResumeToken | The position of the last event the client no longer needs the source for: it holds the events durably, or it has durably taken their effect into the state it keeps (for example by reading the current state at a snapshot that covers this position). Everything up to it is acknowledged: the server may release that source for this consumer, a resume continues after it, and a BOUNDED consumer's progress lag is measured from it. |
Response: AcknowledgeSubscriptionResponse
Empty response (no fields).
CancelSubscription
End a consumer's registration: it stops holding the log, and its incarnation is refused from then on. Refused with FAILED_PRECONDITION and reason CONSUMER_TERMINATED when the incarnation has already ended, and NOT_FOUND for an id never registered.
Request: CancelSubscriptionRequest
| Field | Type | Description |
|---|---|---|
consumer_id | string | The consumer to cancel. |
incarnation | uint64 | The incarnation to cancel, from the Subscribe response metadata. A different incarnation of the same id is left alone. |
Response: CancelSubscriptionResponse
Empty response (no fields).
Types
AcknowledgeSubscriptionRequest
| Field | Type | Description |
|---|---|---|
consumer_id | string | The consumer acknowledging. |
incarnation | uint64 | Its incarnation, from the Subscribe response metadata. |
position | ResumeToken | The position of the last event the client no longer needs the source for: it holds the events durably, or it has durably taken their effect into the state it keeps (for example by reading the current state at a snapshot that covers this position). Everything up to it is acknowledged: the server may release that source for this consumer, a resume continues after it, and a BOUNDED consumer's progress lag is measured from it. |
AcknowledgeSubscriptionResponse
No fields.
BoundedRetention
Finite bounds after which the registration ends. Both limits are required and must be above zero.
| Field | Type | Description |
|---|---|---|
max_progress_lag_ms | uint64 | Most age, in milliseconds, of the oldest log entry the consumer has not yet been sent. |
max_retained_bytes | uint64 | Most bytes of log the consumer's position may require the shard to keep. |
liveness_timeout_ms | uint64 | Longest time, in milliseconds, without the consumer being connected and heartbeating. Absent: the registration does not end for being absent, only for the two limits above. |
CancelSubscriptionRequest
| Field | Type | Description |
|---|---|---|
consumer_id | string | The consumer to cancel. |
incarnation | uint64 | The incarnation to cancel, from the Subscribe response metadata. A different incarnation of the same id is left alone. |
CancelSubscriptionResponse
No fields.
CdcFilters
Filters applied server-side before streaming events to the client.
| Field | Type | Description |
|---|---|---|
edge_types | string[] | If non-empty, only deliver events that touch at least one adjacency or edge-property key whose edge type matches one of these strings. Example: ["FOLLOWS", "LIKES"] Leave empty to receive all edge types. |
is_migration | bool | When set: only deliver events whose is_migration flag equals this value. When absent (HasField returns false): deliver both migration and non-migration events. |
ChangeEvent
A single applied Raft log entry delivered to the CDC consumer.
| Field | Type | Description |
|---|---|---|
ts | uint64 | HLC commit timestamp of the entry, in wall-clock microseconds since the Unix epoch with logical increments inside one microsecond: the latest commit timestamp of the transactions it carries. 0 for an entry that carries none (membership, leader blank). |
term | uint64 | Term of the leader that wrote this entry. |
log_index | uint64 | Raft log index. Monotonically increasing per shard. |
shard_id | uint32 | Shard that produced this entry. |
is_migration | bool | True if this entry was written during a shard migration (e.g. bulk data transfer). Most CDC consumers should skip migration entries. |
ops | ChangeOp[] | The mutations carried by this log entry. Empty for an entry that carries none, and for a progress event: entries up to position held nothing the stream's filters keep, so the client can acknowledge past them. |
position | ResumeToken | Where the stream continues after this event: pass it on reconnect to resume here, and to AcknowledgeSubscription once the event is held. |
ChangeOp
A single storage mutation within a ChangeEvent.
| Field | Type | Description |
|---|---|---|
type | ChangeOpType | — |
partition | uint32 | Storage partition (see coordinode_storage::engine::partition::Partition). |
key | bytes | Storage key. Format depends on partition type. For REMOVE_RANGE this carries the range's inclusive start bound. |
value | bytes | Value bytes. Empty for DELETE ops; carries the merge operand for MERGE. For REMOVE_RANGE this carries the range's exclusive end bound (the half-open range deleted is [key, value)). |
ConsumerRetention
How long a consumer's registration protects the history it still needs. One of the two must be chosen; there is no default.
| Field | Type | Description |
|---|---|---|
strict | StrictRetention | Protection lasts until the consumer is cancelled. Missing heartbeats do not end it. |
bounded | BoundedRetention | Protection ends once one of the declared bounds is crossed. |
ResumeToken
Position in a shard's Raft log used to resume after disconnect.
The token names one log index, segment_id + entry_offset: the next entry to deliver. It stays valid when the server rewrites its log files.
| Field | Type | Description |
|---|---|---|
shard_id | uint32 | Shard this stream targets. Always 0 in CE single-shard mode. |
segment_id | uint64 | Log index of the first entry of the segment that held the last delivered entry. 0 together with entry_offset 0 means "the oldest retained entry". |
entry_offset | uint64 | Entries of that segment already delivered, counted from segment_id. |
StrictRetention
Protection until cancellation.
No fields.
SubscribeRequest
| Field | Type | Description |
|---|---|---|
resume_token | ResumeToken | Token from the last received ChangeEvent.position. Omit (all-zero) to start from the oldest retained entry. |
filters | CdcFilters | Optional filter to reduce bandwidth. |
consumer_id | string | The consumer this stream reads as: globally unique, chosen by the client. Required. |
incarnation | uint64 | 0 registers the consumer anew with retention, starting at resume_token (or the oldest retained entry without one); refused with ALREADY_EXISTS while another incarnation of the id is live. Any other value resumes that live incarnation, and retention is ignored. |
retention | ConsumerRetention | Required when registering (incarnation 0). |
Enums
ChangeOpType
| Value | Number | Description |
|---|---|---|
CHANGE_OP_TYPE_UNSPECIFIED | 0 | — |
INSERT | 1 | — |
DELETE | 2 | — |
MERGE | 3 | — |
NOOP | 4 | — |
RAFT_ENTRY | 5 | — |
RAFT_TRUNCATION | 6 | — |
REMOVE_RANGE | 7 | Delete every key in the half-open range [key, value) of partition with a single range tombstone. Emitted for a dense contiguous run of deleted keys (run-length coalescing) or a whole-prefix DROP; the range covers only keys that are all being deleted, never a surviving key in a gap. |