Skip to content

Add mutate_id wave tag to Subscribe delivery frames #340

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tylerhawkes merged 1 commit into from
Jul 9, 2026
Merged

Add mutate_id wave tag to Subscribe delivery frames #340

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 36 additions & 18 deletions proto/mls/api/v1/mls.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -451,19 +451,25 @@ message SubscribeRequest {
// adopted via the capabilities advertised on Started.
message Mutate {
repeated Subscription adds = 1; // begin delivering these topics
repeated bytes removes = 2; // topics to stop delivering
repeated bytes removes = 2; // stop delivering; clears the topic's cursor floor so a re-add replays

// Catch this Mutate's adds up to the live edge — history, TopicsLive
// markers, and the wave's CatchupComplete — but do NOT register them
// for live delivery. The markers then mean "you have everything as of
// now". Combined with half-closing the request stream, this is the
// Catch this Mutate's adds up — history, TopicsLive markers, and the
// wave's CatchupComplete — but do NOT register them for live delivery.
// The markers then mean "you have everything as of the wave's start";
// later messages arrive on no lane of this stream. Combined with
// half-closing the request stream, this is the
// bounded catch-up ("sync") mode: the server finishes the wave and then
// closes the stream itself. Removals in the Mutate are unaffected.
bool history_only = 3;

// Client-chosen correlation id, echoed on this wave's CatchupComplete
// so completions are attributable when waves overlap. SHOULD be unique
// per stream; 0 = no correlation requested (still echoed as 0).
// Client-chosen correlation id: echoed on this wave's CatchupComplete,
// and stamped on every delivery frame of the wave's catch-up replay
// (Messages.mutate_id). MUST be nonzero when adds are present (0 is the
// live tag), and MUST NOT match the mutate_id of a wave still in flight
// on the stream (an in-flight collision would make two waves' frames and
// completions indistinguishable) — either violation fails the stream
// with INVALID_ARGUMENT. SHOULD be unique per stream so completed waves
// stay attributable too.
uint64 mutate_id = 4;

// A topic to subscribe, with the cursor to resume from.
Expand Down Expand Up @@ -492,15 +498,22 @@ message SubscribeResponse {
Started started = 2; // sent once, immediately on open, before any catch-up
Ping ping = 3; // idle liveness challenge; receiver MUST answer with Pong
Pong pong = 4; // answer to a client Ping
TopicsLive topics_live = 5; // these topics just crossed from catch-up to live
CatchupComplete catchup_complete = 6; // a Mutate's adds are fully delivered
TopicsLive topics_live = 5; // no more replay for these topics; live begins after CatchupComplete
CatchupComplete catchup_complete = 6; // acks a Mutate; wave completion if it started one
}

// A batch of new messages; group and welcome messages share the stream,
// depending on which subscriptions are active.
// depending on which subscriptions are active. A frame belongs to exactly
// one catch-up wave or to live — the server never mixes lanes, or two
// waves, in one frame — and each lane is delivered in ascending id order
// per message kind (live: across all live topics on the stream; a wave:
// across the wave's topics, one merged cursor-ordered pass).
message Messages {
repeated GroupMessage group_messages = 1;
repeated WelcomeMessage welcome_messages = 2;
// The catch-up wave that produced this frame: the Mutate's mutate_id
// for wave replay, 0 for live tail.
uint64 mutate_id = 3;
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
}

// The first frame on every stream.
Expand All @@ -515,20 +528,25 @@ message SubscribeResponse {
repeated Capability capabilities = 2;
}

// Sent once per Mutate that adds subscriptions (a catch-up "wave"), after
// the wave's last TopicsLive: everything the Mutate asked for is delivered.
// Sent once per Mutate: at wave completion (after the wave's last
// TopicsLive) for a Mutate that started a catch-up "wave", immediately for
// one that did not (nothing added — removes-only or empty — or every add
// a no-op). Also the catch-up
// seam: live frames (mutate_id 0) for the wave's topics begin only after
// this frame.
message CatchupComplete {
uint64 mutate_id = 1; // echoes the Mutate that started this wave (0 if none given)
uint64 mutate_id = 1; // echoes the Mutate; 0 only if a waveless Mutate carried 0
}

// Emitted when topics finish catch-up, AFTER the last history frame for
// them — including any live messages that queued up behind the catch-up,
// which were equally historical from the client's perspective — so every
// later frame for a listed topic is live tail. Informational only: delivery
// them — including messages that arrived mid-wave and were folded into it,
// which were equally historical from the client's perspective — so no
// further replay for a listed topic follows; its live (mutate_id 0) frames
// begin after the wave's CatchupComplete. Informational only: delivery
// correctness (no duplicates, no gaps) never depends on it. Re-adding a
// topic re-runs catch-up and re-emits it; receivers treat it idempotently.
message TopicsLive {
repeated bytes topics = 1; // kind-prefixed topics now tailing live
repeated bytes topics = 1; // kind-prefixed topics done replaying
}

// Optional per-stream protocol features (none defined yet; future
Expand Down
60 changes: 39 additions & 21 deletions proto/xmtpv4/message_api/message_api.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,19 +95,25 @@ message SubscribeRequest {
// whose kind the node does not serve fails the stream with INVALID_ARGUMENT.
message Mutate {
repeated Subscription adds = 1; // begin delivering these topics
repeated bytes removes = 2; // topics to stop delivering

// Catch this Mutate's adds up to the live edge — history, TopicsLive
// markers, and the wave's CatchupComplete — but do NOT register them for
// live delivery. The markers then mean "you have everything as of now".
// Combined with half-closing the request stream, this is the bounded
// catch-up ("sync") mode: the node finishes the wave then closes the
// stream itself. Removals in the Mutate are unaffected.
repeated bytes removes = 2; // stop delivering; clears the topic's cursor floor so a re-add replays

// Catch this Mutate's adds up — history, TopicsLive markers, and the
// wave's CatchupComplete — but do NOT register them for live delivery.
// The markers then mean "you have everything as of the wave's start";
// later envelopes arrive on no lane of this stream. Combined with
// half-closing the request stream, this is the bounded catch-up ("sync")
// mode: the node finishes the wave then closes the stream itself.
// Removals in the Mutate are unaffected.
bool history_only = 3;

// Client-chosen correlation id, echoed on this wave's CatchupComplete so
// completions are attributable when waves overlap. SHOULD be unique per
// stream; 0 = no correlation requested (still echoed as 0).
// Client-chosen correlation id: echoed on this wave's CatchupComplete,
// and stamped on every delivery frame of the wave's catch-up replay
// (Envelopes.mutate_id). MUST be nonzero when adds are present (0 is the
// live tag), and MUST NOT match the mutate_id of a wave still in flight
// on the stream (an in-flight collision would make two waves' frames and
// completions indistinguishable) — either violation fails the stream
// with INVALID_ARGUMENT. SHOULD be unique per stream so completed waves
// stay attributable too.
uint64 mutate_id = 4;

// A topic to subscribe, with the vector cursor to resume from.
Expand Down Expand Up @@ -135,14 +141,21 @@ message SubscribeResponse {
Started started = 2; // sent once, immediately on open, before any catch-up
Ping ping = 3; // idle liveness challenge; receiver MUST answer with Pong
Pong pong = 4; // answer to a client Ping
TopicsLive topics_live = 5; // these topics just crossed from catch-up to live
CatchupComplete catchup_complete = 6; // a Mutate's adds are fully delivered
TopicsLive topics_live = 5; // no more replay for these topics; live begins after CatchupComplete
CatchupComplete catchup_complete = 6; // acks a Mutate; wave completion if it started one
}

// A batch of envelopes across the active subscriptions; the client demuxes
// by each envelope's target topic.
// by each envelope's target topic. A frame belongs to exactly one catch-up
// wave or to live — the node never mixes lanes, or two waves, in one frame
// — and each lane delivers every originator's envelopes in ascending
// originator_sequence_id (live: across all live topics on the stream; a
// wave: across the wave's topics).
message Envelopes {
repeated xmtp.xmtpv4.envelopes.OriginatorEnvelope envelopes = 1;
// The catch-up wave that produced this frame: the Mutate's mutate_id
// for wave replay, 0 for live tail.
uint64 mutate_id = 2;
}

// The first frame on every stream.
Expand All @@ -157,20 +170,25 @@ message SubscribeResponse {
repeated Capability capabilities = 2;
}

// Sent once per Mutate that adds subscriptions (a catch-up "wave"), after
// the wave's last TopicsLive: everything the Mutate asked for is delivered.
// Sent once per Mutate: at wave completion (after the wave's last
// TopicsLive) for a Mutate that started a catch-up "wave", immediately for
// one that did not (nothing added — removes-only or empty — or every add
// a no-op). Also the catch-up
// seam: live frames (mutate_id 0) for the wave's topics begin only after
// this frame.
message CatchupComplete {
uint64 mutate_id = 1; // echoes the Mutate that started this wave (0 if none given)
uint64 mutate_id = 1; // echoes the Mutate; 0 only if a waveless Mutate carried 0
}

// Emitted when topics finish catch-up, AFTER the last history frame for
// them — including any live envelopes that queued behind the catch-up,
// which were equally historical from the client's perspective — so every
// later frame for a listed topic is live tail. Informational only: delivery
// them — including envelopes that arrived mid-wave and were folded into it,
// which were equally historical from the client's perspective — so no
// further replay for a listed topic follows; its live (mutate_id 0) frames
// begin after the wave's CatchupComplete. Informational only: delivery
// correctness (no duplicates, no gaps) never depends on it. Re-adding a
// topic re-runs catch-up and re-emits it; receivers treat it idempotently.
message TopicsLive {
repeated bytes topics = 1; // kind-prefixed topics now tailing live
repeated bytes topics = 1; // kind-prefixed topics done replaying
}

// Optional per-stream protocol features (none defined yet; future revisions
Expand Down
Loading