From da6dcf11ebf2c74e0d7c233152b82b480b22f761 Mon Sep 17 00:00:00 2001 From: Tyler Hawkes Date: Wed, 8 Jul 2026 13:43:38 -0600 Subject: [PATCH] Add mutate_id wave tag to Subscribe delivery frames Messages.mutate_id (mls v3) and Envelopes.mutate_id (xmtpv4) carry the catch-up wave that produced each frame (0 = live tail). Mutate.mutate_id is required nonzero when adds are present; comments document the order guarantees, the catch-up seam, and CatchupComplete acking every Mutate. Co-Authored-By: Claude Fable 5 --- proto/mls/api/v1/mls.proto | 54 ++++++++++++------- proto/xmtpv4/message_api/message_api.proto | 60 ++++++++++++++-------- 2 files changed, 75 insertions(+), 39 deletions(-) diff --git a/proto/mls/api/v1/mls.proto b/proto/mls/api/v1/mls.proto index 87845cf6..8f8b2229 100644 --- a/proto/mls/api/v1/mls.proto +++ b/proto/mls/api/v1/mls.proto @@ -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. @@ -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; } // The first frame on every stream. @@ -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 diff --git a/proto/xmtpv4/message_api/message_api.proto b/proto/xmtpv4/message_api/message_api.proto index 0a67dffd..11cd4370 100644 --- a/proto/xmtpv4/message_api/message_api.proto +++ b/proto/xmtpv4/message_api/message_api.proto @@ -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. @@ -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. @@ -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