diff --git a/pkg/api/message/subscribe.go b/pkg/api/message/subscribe.go index 1780f8d24..3e22bb119 100644 --- a/pkg/api/message/subscribe.go +++ b/pkg/api/message/subscribe.go @@ -7,18 +7,24 @@ import ( "io" "maps" "math" + "slices" + "sort" "time" "connectrpc.com/connect" + "github.com/cenkalti/backoff/v4" "go.uber.org/zap" "google.golang.org/protobuf/proto" + "github.com/xmtp/xmtpd/pkg/constants" "github.com/xmtp/xmtpd/pkg/db" + "github.com/xmtp/xmtpd/pkg/db/queries" "github.com/xmtp/xmtpd/pkg/envelopes" envelopesProto "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/envelopes" "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/message_api" "github.com/xmtp/xmtpd/pkg/topic" "github.com/xmtp/xmtpd/pkg/utils" + "github.com/xmtp/xmtpd/pkg/utils/retryerrors" ) const ( @@ -33,6 +39,22 @@ const ( // connections (which would just hit rate limits). At ~hundreds of bytes/topic this is a // large-but-bounded budget; exceeding it fails the Mutate with ResourceExhausted. maxActiveSubscribeTopics = 1_000_000 + // maxMutateAdds caps the raw adds a single Mutate may carry, bounding that wave's merged + // catch-up scan: the scan's floor arrays hold one entry per (topic, originator) pair and are + // resent with EVERY page query, so per-page query params scale with the Mutate's add count. + // Counted pre-dedup so the check is stateless. A client with a larger set splits it across + // Mutates, whose waves run concurrently (XIP-83 server requirement 8); an over-cap Mutate is + // rejected with ResourceExhausted, never silently truncated. + maxMutateAdds = 100_000 + // maxMutateCursorEntries caps the total cursor entries a single Mutate may carry across its + // adds. maxMutateAdds bounds the topic count, but each add's cursor is a per-originator + // vector, so without this cap one Mutate (bounded only by the transport message limit) could + // name millions of (topic, originator) pairs — every pair rides the wave's floor arrays, + // resent with EVERY page query, and every named originator joins the wave's one-shot ceilings + // query. Counted pre-dedup so the check is stateless; an over-cap Mutate is rejected with + // ResourceExhausted, never silently truncated. Generous for real clients: it fits a full + // maxMutateAdds wave whose every topic names a 10-originator vector. + maxMutateCursorEntries = 1_000_000 // maxInflightSubscribeWaves caps the concurrent catch-up waves one stream may have running. Each // wave is a detached fetcher goroutine plus paginated DB queries, and maxActiveSubscribeTopics // does NOT bound it: a remove+re-add (reset) leaves the old wave running while the active-topic @@ -322,6 +344,10 @@ type subscribeSession struct { sendTimer *time.Timer // maxFrameBytes overrides maxSubscribeFrameBytes when > 0 (tests only). maxFrameBytes int + // maxAdds overrides maxMutateAdds when > 0 (tests only). + maxAdds int + // maxCursorEntries overrides maxMutateCursorEntries when > 0 (tests only). + maxCursorEntries int outbound chan *message_api.SubscribeResponse // senderDone is closed exactly once, when the sender goroutine exits; sendErr is its terminal @@ -440,7 +466,13 @@ func (sess *subscribeSession) send(resp *message_api.SubscribeResponse) error { // sendEnvelopes delivers envelopes split into frames each under maxSubscribeFrameBytes, so a large // catch-up page or flushed pending buffer never goes out as one oversized (stream-aborting) frame. -func (sess *subscribeSession) sendEnvelopes(envs []*envelopesProto.OriginatorEnvelope) error { +// Every frame is stamped with the catch-up wave that produced it — a Mutate's mutate_id for wave +// replay, 0 for live tail (XIP-83 server requirement 3). A frame is exactly one or the other; +// callers never mix lanes in one call. +func (sess *subscribeSession) sendEnvelopes( + envs []*envelopesProto.OriginatorEnvelope, + mutateID uint64, +) error { if len(envs) == 0 { return nil } @@ -454,7 +486,7 @@ func (sess *subscribeSession) sendEnvelopes(envs []*envelopesProto.OriginatorEnv if len(frame) == 0 { return nil } - if err := sess.send(newSubscribeEnvelopes(frame)); err != nil { + if err := sess.send(newSubscribeEnvelopes(frame, mutateID)); err != nil { return err } frame = nil // do NOT reuse the backing array; the sent frame still references it @@ -463,11 +495,29 @@ func (sess *subscribeSession) sendEnvelopes(envs []*envelopesProto.OriginatorEnv } for _, env := range envs { size := proto.Size(env) - // An envelope larger than `limit` on its own is NOT dropped: limit is a soft batching + // An envelope larger than `limit` is not dropped just for that: limit is a soft batching // target (2 MiB), an order of magnitude under the transport's hard cap (GRPCPayloadLimit, - // 25 MiB). Such an envelope simply flushes the current frame and then goes out alone — and - // it always fits, because it was publishable under that same 25 MiB cap. Skipping it (as the - // old batchAndSendEnvelopes did) would silently lose a valid, deliverable message. + // 25 MiB) — it flushes the current frame and goes out alone. But publish admission does + // not guarantee the lone frame fits: the stored envelope carries the originator's wrapper + // on top of what the payer sent, plus the response framing, so it can exceed the hard cap + // by a sliver. Sending would abort the stream — and a reconnecting client's wave would hit + // the same row again, wedging it permanently — so such an envelope is skipped with a + // warning instead, matching batchAndSendEnvelopes on the legacy paths. Callers advance + // their dedup cursors before calling send, so a skip cannot stall a wave or re-deliver. + if size > limit { + framed := proto.Size(newSubscribeEnvelopes( + []*envelopesProto.OriginatorEnvelope{env}, mutateID, + )) + if framed > constants.GRPCPayloadLimit { + sess.logger.Warn( + "skipping oversized envelope", + zap.Int("framed_bytes", framed), + zap.Int("limit", constants.GRPCPayloadLimit), + zap.Uint64("mutate_id", mutateID), + ) + continue + } + } if len(frame) > 0 && frameBytes+size > limit { if err := flush(); err != nil { return err @@ -484,8 +534,7 @@ func (sess *subscribeSession) sendEnvelopes(envs []*envelopesProto.OriginatorEnv func (sess *subscribeSession) routeLive(batch []*envelopes.OriginatorEnvelope) error { var toSend []*envelopes.OriginatorEnvelope for _, env := range batch { - ts := sess.topics[string(env.TargetTopic().Bytes())] - if ts != nil && ts.phase == topicGated { + if ts, ok := sess.topics[string(env.TargetTopic().Bytes())]; ok && ts.phase == topicGated { if err := sess.bufferLive(ts, env); err != nil { return err } @@ -494,7 +543,10 @@ func (sess *subscribeSession) routeLive(batch []*envelopes.OriginatorEnvelope) e // Live, or not (or no longer) ours: advanceLive dedups the live topics and drops the rest. toSend = append(toSend, env) } - return sess.sendEnvelopes(sess.advanceLive(toSend)) + // Live tail: tagged 0. The worker dispatches each originator's envelopes in ascending + // sequence order and the writer sends in arrival order, so the live lane stays totally + // ordered per originator (XIP-83 server requirement 4). + return sess.sendEnvelopes(sess.advanceLive(toSend), 0) } // bufferLive holds a live envelope for a gated topic until its wave opens the gate, enforcing the @@ -514,13 +566,17 @@ func (sess *subscribeSession) bufferLive(ts *topicState, env *envelopes.Originat // advanceLive dedups envs against their topics' live cursors, advancing each cursor in place, and // returns the proto envelopes ready to send. An envelope for a topic that is not (or no longer) live // is dropped — the per-topic analogue of advanceTopicCursors, reading the cursor from topicState. +// The cursor-max dedup relies on the system invariant that each originator's envelopes become +// visible in sequence order (one sequential writer per originator); a row committing out of order +// after a reader passed it would be undeliverable stream-wide — the same pre-existing assumption +// the live pollers make when advancing lastSeen from raw rows. func (sess *subscribeSession) advanceLive( envs []*envelopes.OriginatorEnvelope, ) []*envelopesProto.OriginatorEnvelope { result := make([]*envelopesProto.OriginatorEnvelope, 0, len(envs)) for _, env := range envs { - ts := sess.topics[string(env.TargetTopic().Bytes())] - if ts == nil { + ts, ok := sess.topics[string(env.TargetTopic().Bytes())] + if !ok { sess.logger.Warn( "received envelope for unsubscribed topic", zap.Binary("topic", env.TargetTopic().Bytes()), @@ -549,15 +605,20 @@ func (sess *subscribeSession) handleCatchUp(b catchUpBatch) (bool, error) { fmt.Errorf("catch-up failed: %w", b.err), ) } - w := sess.waves[b.wave] - if w == nil { - return false, nil // wave already torn down (e.g. all its topics were removed) + w, ok := sess.waves[b.wave] + if !ok { + // Defensive only: a wave is deleted solely when its own done marker is processed, so + // this cannot fire today. Removes do NOT tear down an in-flight wave — an orphaned + // scan just drains to its ceilings (its pages dropped by envsOwnedByWave) and + // completes normally. + return false, nil } - // Deliver this page's history. A history_only wave dedups against its own throwaway cursors; a - // live wave first drops pages for topics it no longer owns (removed, or reset under a newer - // wave) — so it cannot advance a reset topic's live cursor and skip history the newer wave owes - // — then dedups the rest against each topic's live cursor. + // Deliver this page's history, stamped with the wave's mutate_id. A history_only wave dedups + // against its own throwaway cursors; a live wave first drops pages for topics it no longer + // owns (removed, or reset under a newer wave) — so it cannot advance a reset topic's live + // cursor and skip history the newer wave owes — then dedups the rest against each topic's + // live cursor. var toSend []*envelopesProto.OriginatorEnvelope if w.historyOnly { toSend = advanceTopicCursors(w.cursors, b.envs, sess.logger) @@ -565,7 +626,7 @@ func (sess *subscribeSession) handleCatchUp(b catchUpBatch) (bool, error) { toSend = sess.advanceLive(sess.envsOwnedByWave(b.envs, b.wave)) } if len(toSend) > 0 { - if err := sess.sendEnvelopes(toSend); err != nil { + if err := sess.sendEnvelopes(toSend, w.mutateID); err != nil { return false, err } } @@ -574,23 +635,45 @@ func (sess *subscribeSession) handleCatchUp(b catchUpBatch) (bool, error) { return false, nil } - // Wave complete: open the gate for each live topic this wave still owns (flushing its buffered - // live, deduped against the now-advanced cursor) and collect the topics to announce; then - // CatchupComplete. flushAndGoLive is a no-op for a topic removed or reset under a newer wave, so - // a stale wave never opens the newer wave's gate or flushes its buffer out of order. + // Wave complete: fold in the live envelopes buffered while its topics were gated — merged + // into per-originator sequence order and stamped with the wave's mutate_id, since the wave + // owns them (their sequence ids sit above the scan's pinned ceilings). The fold appends after + // every scan page, so end-to-end the wave guarantees only ascending sequence ids per + // originator, not the scan's global (originator, sequence) tuple order — then announce + // the surviving topics in one TopicsLive and emit the wave's CatchupComplete. Only then are + // the gates open (topicLive), so a live (mutate_id 0) frame for a wave's topic is never + // delivered before its CatchupComplete (XIP-83 server requirement 4: the seam). A topic + // removed or reset under a newer wave is skipped, so a stale wave never opens the newer + // wave's gate or flushes its buffer. wire := make([][]byte, 0, len(w.topics)) + var folded []*envelopes.OriginatorEnvelope for _, t := range w.topics { if w.historyOnly { wire = append(wire, t.wire) continue } - announced, err := sess.flushAndGoLive(t.cursorKey, b.wave) - if err != nil { - return false, err + ts, ok := sess.topics[t.cursorKey] + if !ok || ts.phase != topicGated || ts.wave != b.wave { + continue } - if announced { - wire = append(wire, t.wire) + for _, e := range ts.pending { + sess.pendingBytes -= proto.Size(e.Proto()) } + folded = append(folded, ts.pending...) + ts.pending = nil + ts.phase = topicLive + wire = append(wire, t.wire) + } + // Each topic's buffer is in per-originator dispatch order, but the wave's replay must stay + // totally ordered per originator ACROSS its topics: merge before framing. + sort.SliceStable(folded, func(i, j int) bool { + if folded[i].OriginatorNodeID() != folded[j].OriginatorNodeID() { + return folded[i].OriginatorNodeID() < folded[j].OriginatorNodeID() + } + return folded[i].OriginatorSequenceID() < folded[j].OriginatorSequenceID() + }) + if err := sess.sendEnvelopes(sess.advanceLive(folded), w.mutateID); err != nil { + return false, err } if len(wire) > 0 { if err := sess.send(newSubscribeTopicsLive(wire)); err != nil { @@ -604,35 +687,12 @@ func (sess *subscribeSession) handleCatchUp(b catchUpBatch) (bool, error) { return sess.halfClosed && len(sess.waves) == 0, nil } -// flushAndGoLive completes a gated topic owned by `wave`: it flushes the live envelopes buffered -// during catch-up (deduped against the now-advanced live cursor) and transitions it to live, -// returning true once announced. It is a no-op returning false if the topic is gone or now owned by -// a newer wave (a reset), so a stale wave never opens the newer wave's gate or replays its buffer. -func (sess *subscribeSession) flushAndGoLive(cursorKey string, wave int) (bool, error) { - ts := sess.topics[cursorKey] - if ts == nil || ts.phase != topicGated || ts.wave != wave { - return false, nil - } - if len(ts.pending) > 0 { - for _, e := range ts.pending { - sess.pendingBytes -= proto.Size(e.Proto()) - } - buf := ts.pending - ts.pending = nil - if err := sess.sendEnvelopes(sess.advanceLive(buf)); err != nil { - return false, err - } - } - ts.phase = topicLive - return true, nil -} - // handleRequest dispatches one client frame. func (sess *subscribeSession) handleRequest(req *message_api.SubscribeRequest) error { v1 := req.GetV1() if v1 == nil { // Unrecognized version arm: fail rather than silently ignore, so a forward-version - // client is not left waiting on a response (XIP-83 req 8). + // client is not left waiting on a response (XIP-83 req 10, version pinning). return connect.NewError( connect.CodeInvalidArgument, errors.New("unrecognized SubscribeRequest version"), @@ -664,6 +724,67 @@ func (sess *subscribeSession) handleMutate(m *message_api.SubscribeRequest_V1_Mu // ---- Validate (no state mutation): any failure here returns before a single change. ---- + // A wave's replay frames are stamped with its mutate_id, and 0 is the live tag, so a Mutate + // with adds cannot ride on 0 (XIP-83 server requirement 3). + if len(m.GetAdds()) > 0 && m.GetMutateId() == 0 { + return connect.NewError( + connect.CodeInvalidArgument, + errors.New("a Mutate with adds requires a nonzero mutate_id"), + ) + } + + // A mutate_id may not collide with a wave still in flight: the two waves' replay frames and + // CatchupComplete acks would be indistinguishable to the client (XIP-83 server requirement 3). + // Enforced for ANY Mutate — even a removes-only reuse would emit an immediate CatchupComplete + // ambiguous with the in-flight wave's. Reuse AFTER a wave's CatchupComplete stays legal. Waves + // exist only for Mutates with adds, so every in-flight mutateID is nonzero and 0 never collides. + if m.GetMutateId() != 0 { + for _, w := range sess.waves { + if w.mutateID == m.GetMutateId() { + return connect.NewError( + connect.CodeInvalidArgument, + fmt.Errorf("mutate_id %d is already in flight on this stream", m.GetMutateId()), + ) + } + } + } + + // Bound the raw adds (pre-dedup, so the check is stateless) so one Mutate's merged catch-up + // scan cannot carry unbounded per-page query params (see maxMutateAdds). The client splits a + // larger set across Mutates, whose waves run concurrently. + addLimit := maxMutateAdds + if sess.maxAdds > 0 { + addLimit = sess.maxAdds + } + if len(m.GetAdds()) > addLimit { + return connect.NewError( + connect.CodeResourceExhausted, + fmt.Errorf("adds per Mutate limit %d exceeded; split adds across multiple Mutates", + addLimit), + ) + } + + // Bound the total cursor entries across the adds the same way (pre-dedup): the add cap alone + // does not bound the wave's floor pairs or ceiling originators, because a single add's cursor + // may name arbitrarily many originators (see maxMutateCursorEntries). + entryLimit := maxMutateCursorEntries + if sess.maxCursorEntries > 0 { + entryLimit = sess.maxCursorEntries + } + cursorEntries := 0 + for _, a := range m.GetAdds() { + cursorEntries += len(a.GetLastSeen().GetNodeIdToSequenceId()) + } + if cursorEntries > entryLimit { + return connect.NewError( + connect.CodeResourceExhausted, + fmt.Errorf( + "cursor entries per Mutate limit %d exceeded; split adds across multiple Mutates", + entryLimit, + ), + ) + } + // Parse removes up front so a malformed remove fails the whole Mutate, and so the add cap and // history_only checks below can account for topics this Mutate will drop. removes := make([]*topic.Topic, 0, len(m.GetRemoves())) @@ -793,14 +914,11 @@ func (sess *subscribeSession) handleMutate(m *message_api.SubscribeRequest_V1_Mu if len(order) == 0 { // Removes-only (or empty) Mutate: no catch-up, but confirm it applied so a client that - // subscribed to nothing still learns the mutate took effect. - if err := sess.send(newSubscribeCatchupComplete(m.GetMutateId())); err != nil { - return err - } - if sess.halfClosed && len(sess.waves) == 0 { - return sess.flush() - } - return nil + // subscribed to nothing still learns the mutate took effect. No half-close handling here: + // no Mutate is ever processed after halfClosed is set (the main loop nils the request + // channel; drainPendingRequests stops at closure), and the drain-finished check lives + // where waves actually complete (handleCatchUp's done flag). + return sess.send(newSubscribeCatchupComplete(m.GetMutateId())) } wave := &subscribeWave{mutateID: m.GetMutateId(), historyOnly: historyOnly} @@ -811,11 +929,9 @@ func (sess *subscribeSession) handleMutate(m *message_api.SubscribeRequest_V1_Mu // originator set) off the writer goroutine. The persisted live cursor stays sparse (provided // only, grown as originators are actually seen) to bound memory at the 1M ceiling. providedCursors := make(db.TopicCursors, len(order)) - cursorKeys := make([]string, 0, len(order)) for _, k := range order { a := byKey[k] wave.topics = append(wave.topics, a.t) - cursorKeys = append(cursorKeys, k) providedCursors[k] = cloneVectorClock(a.provided) if historyOnly { @@ -832,7 +948,6 @@ func (sess *subscribeSession) handleMutate(m *message_api.SubscribeRequest_V1_Mu sess.ctx, sess.nextWave, providedCursors, - cursorKeys, sess.catchUpCh, sess.logger, ) @@ -877,8 +992,8 @@ func (sess *subscribeSession) removeTopic(parsed *topic.Topic) { // touch any in-flight wave: a wave only ever acts on topics it still owns (flushAndGoLive / // envsOwnedByWave both re-check ownership), so a removed topic's pages and completion are ignored. func (sess *subscribeSession) removeTopicState(cursorKey string) { - ts := sess.topics[cursorKey] - if ts == nil { + ts, ok := sess.topics[cursorKey] + if !ok { return } for _, e := range ts.pending { @@ -896,8 +1011,8 @@ func (sess *subscribeSession) envsOwnedByWave( wave int, ) []*envelopes.OriginatorEnvelope { owned := func(env *envelopes.OriginatorEnvelope) bool { - ts := sess.topics[string(env.TargetTopic().Bytes())] - return ts != nil && ts.phase == topicGated && ts.wave == wave + ts, ok := sess.topics[string(env.TargetTopic().Bytes())] + return ok && ts.phase == topicGated && ts.wave == wave } // Fast path: every envelope belongs to a topic this wave still owns (the common case — no // concurrent remove/reset), so the page passes through without reallocation. @@ -920,17 +1035,22 @@ func (sess *subscribeSession) envsOwnedByWave( return out } -// runSubscribeCatchUp paginates history for a wave's topics (off the writer goroutine) and hands -// raw pages back over catchUpCh, ending with a done marker. It resolves the originator set and fills -// providedCursors into its own query cursors here — that originator lookup is a DB round-trip on a -// cache miss, so it MUST stay off the writer goroutine (else a slow DB would stall liveness and -// live delivery and could false-reap a healthy stream). The writer owns the sparse live cursors. -// Every channel send is guarded by ctx so the fetcher cannot leak if the writer has torn down. +// runSubscribeCatchUp replays history for a wave's topics (off the writer goroutine) and hands +// raw pages back over catchUpCh, ending with a done marker. The replay is ONE merged keyset scan +// in (originator, sequence) order across ALL of the wave's topics — not per-topic bursts — pinned +// to the per-originator ceiling captured at wave start, so the wave's replay is delivered in +// total cursor order per originator and the scan terminates under sustained publishing (XIP-83 +// server requirement 4); everything newer reaches the client through the gated live path and is +// folded into the wave when it completes. It resolves the originator set and fills +// providedCursors into the scan's floor cursors here — that originator lookup is a DB round-trip +// on a cache miss, so it MUST stay off the writer goroutine (else a slow DB would stall liveness +// and live delivery and could false-reap a healthy stream). The writer owns the sparse live +// cursors. Every channel send is guarded by ctx so the fetcher cannot leak if the writer has torn +// down. func (s *Service) runSubscribeCatchUp( ctx context.Context, wave int, providedCursors db.TopicCursors, - cursorKeys []string, catchUpCh chan<- catchUpBatch, logger *zap.Logger, ) { @@ -954,51 +1074,172 @@ func (s *Service) runSubscribeCatchUp( emit(catchUpBatch{wave: wave, err: fmt.Errorf("could not get originator list: %w", err)}) return } - // queryCursors are FILLED (every originator from the provided/zero start) so catch-up covers all - // originators; the fetcher owns and advances them for pagination. - queryCursors := make(db.TopicCursors, len(providedCursors)) + // Pin the wave's replay ceiling: the newest sequence id per originator at wave start. The + // ceiling set is the UNION of the cached list and every originator a provided cursor names, + // so a cursor-named originator the TTL-stale cache has not seen still gets a ceiling row + // instead of being silently dropped by the scan's inner ceiling join (the legacy per-topic + // catch-up replayed it — its unbounded LATERAL scan needed no ceiling). One with no rows in + // gateway_envelopes_meta COALESCEs to ceiling 0 — an empty replay range, harmless. + ceilings, err := s.fetchWaveCeilingsWithRetry( + ctx, + ceilingOriginators(knownOriginators, providedCursors), + ) + if err != nil { + emit(catchUpBatch{wave: wave, err: err}) + return + } + // Floor cursors are FILLED (every originator from the provided/zero start) so catch-up covers + // the full originator set; flattened once into the query params and reused every page. Each + // topic's floors come from its own cursor plus the cached list, so the residual replay gap is + // per topic: an originator that NEITHER the cache NOR that topic's cursor names (a brand-new + // originator this client has never seen on that topic) — cache-bounded, like live originator + // registration, matching the old per-topic catch-up (which filled its cursor floors from the + // same cached list). + floors := make(db.TopicCursors, len(providedCursors)) for k, provided := range providedCursors { filled := cloneVectorClock(provided) db.FillMissingOriginators(filled, knownOriginators) - queryCursors[k] = filled + floors[k] = filled } + params := queries.SelectGatewayEnvelopesWaveScanParams{RowLimit: topicPageLimit} + db.SetWaveScanCursors(¶ms, floors) + db.SetWaveScanCeilings(¶ms, ceilings) - for _, chunkKeys := range utils.ChunkSlice(cursorKeys, maxTopicsPerChunk) { - rowsPerEntry := db.CalculateRowsPerEntry(len(chunkKeys), topicPageLimit) - for { - if ctx.Err() != nil { - return - } - subCursors := make(db.TopicCursors, len(chunkKeys)) - for _, k := range chunkKeys { - subCursors[k] = queryCursors[k] - } - rows, err := s.fetchTopicEnvelopesWithRetry( - ctx, - subCursors, - topicPageLimit, - rowsPerEntry, - ) - if err != nil { - emit(catchUpBatch{wave: wave, err: err}) - return - } - envs := unmarshalEnvelopes(rows, logger) - // Advance the fetcher's own (filled) cursors from the RAW rows so pagination always - // progresses even if some rows fail to unmarshal (otherwise a single bad row in a full - // page re-fetches forever); the writer re-dedups the emitted envs against the live cursor. - advanceCursorsFromRows(queryCursors, rows) - if !emit(catchUpBatch{wave: wave, envs: envs}) { + for { + if ctx.Err() != nil { + return + } + rows, err := s.fetchWaveScanPageWithRetry(ctx, params) + if err != nil { + emit(catchUpBatch{wave: wave, err: err}) + return + } + if len(rows) > 0 { + // The scan position advances from the RAW rows so pagination always progresses + // even if some rows fail to unmarshal (otherwise a single bad row in a full page + // re-fetches forever); the writer re-dedups against the live cursor. The advance + // is strictly past the last raw row (the query resumes on a `>` row-value + // comparison); a `>=` resume would only re-fetch the boundary row each page — + // client-invisible, since the writer's cursor dedup absorbs the duplicate — at + // the cost of one wasted row per page. + last := rows[len(rows)-1] + params.ScanNodeID = last.OriginatorNodeID + params.ScanSequenceID = last.OriginatorSequenceID + if !emit(catchUpBatch{wave: wave, envs: unmarshalEnvelopes(rows, logger)}) { return } - if int32(len(rows)) < rowsPerEntry { - break - } + } + if int32(len(rows)) < topicPageLimit { + break } } emit(catchUpBatch{wave: wave, done: true}) } +// ceilingOriginators returns the union of the cached originator list and every originator named +// in a provided cursor, sorted for a deterministic query parameter order. Every originator with +// a floor entry in the wave's scan must appear here: the scan's ceiling join is INNER, so one +// without a ceiling row would be silently excluded from the replay. +func ceilingOriginators(known []uint32, provided db.TopicCursors) []uint32 { + set := make(map[uint32]struct{}, len(known)) + for _, id := range known { + set[id] = struct{}{} + } + for _, vc := range provided { + for id := range vc { + set[id] = struct{}{} + } + } + out := make([]uint32, 0, len(set)) + for id := range set { + out = append(out, id) + } + slices.Sort(out) + return out +} + +// fetchWaveCeilingsWithRetry pins a wave's replay ceiling — the newest sequence id per originator +// at wave start, as a vector — with the same backoff as the scan itself. The MAX(seq) snapshot is +// a sound replay boundary because of the system invariant that each originator's envelopes become +// visible in sequence order (one sequential writer per originator); a row committing below the +// snapshot after it was taken would be undeliverable stream-wide — the same pre-existing +// assumption the live pollers make when advancing lastSeen from raw rows. +// +// The pin is the first SUCCESSFUL read: a retried fetch pins the wave's boundary at retry time, +// which is indistinguishable from the Mutate having been processed later — no replay frame has +// been sent yet, and once the scan starts the ceiling never moves. (A failed attempt yields no +// snapshot to preserve, and aborting instead would turn a transient DB blip into a stream +// failure for every topic on the stream.) +func (s *Service) fetchWaveCeilingsWithRetry( + ctx context.Context, + originators []uint32, +) (db.VectorClock, error) { + nodeIDs := make([]int32, 0, len(originators)) + for _, o := range originators { + if o > math.MaxInt32 { + continue + } + nodeIDs = append(nodeIDs, int32(o)) + } + boCtx := backoff.WithContext( + utils.NewBackoff(50*time.Millisecond, 300*time.Millisecond, 2*time.Second), ctx, + ) + var ceilings db.VectorClock + operation := func() error { + rows, err := s.store.ReadQuery().SelectOriginatorCeilings(ctx, nodeIDs) + if err != nil { + if !retryerrors.IsRetryableSQLError(err) { + return backoff.Permanent(err) + } + return err + } + // Rebuilt per attempt so a retry after a partial fill cannot mix rows from two snapshots. + ceilings = make(db.VectorClock, len(nodeIDs)) + for _, r := range rows { + ceilings[uint32(r.OriginatorNodeID)] = uint64(r.MaxSequenceID) + } + return nil + } + if err := backoff.Retry(operation, boCtx); err != nil { + return nil, connect.NewError( + connect.CodeInternal, + fmt.Errorf("could not select originator ceilings: %w", err), + ) + } + return ceilings, nil +} + +// fetchWaveScanPageWithRetry fetches one page of a wave's merged replay scan with exponential +// backoff. params carries the wave's pre-flattened floors and ceilings; the caller advances the +// scan position (ScanNodeID / ScanSequenceID) between pages. +func (s *Service) fetchWaveScanPageWithRetry( + ctx context.Context, + params queries.SelectGatewayEnvelopesWaveScanParams, +) ([]queries.GatewayEnvelopesView, error) { + boCtx := backoff.WithContext( + utils.NewBackoff(50*time.Millisecond, 300*time.Millisecond, 2*time.Second), ctx, + ) + var result []queries.GatewayEnvelopesView + operation := func() error { + rows, err := s.store.ReadQuery().SelectGatewayEnvelopesWaveScan(ctx, params) + if err == nil { + result = db.TransformRowsWaveScan(rows) + return nil + } + if !retryerrors.IsRetryableSQLError(err) { + return backoff.Permanent(err) + } + return err + } + if err := backoff.Retry(operation, boCtx); err != nil { + return nil, connect.NewError( + connect.CodeInternal, + fmt.Errorf("could not select envelopes: %w", err), + ) + } + return result, nil +} + func cloneVectorClock(vc db.VectorClock) db.VectorClock { out := make(db.VectorClock, len(vc)) maps.Copy(out, vc) @@ -1063,10 +1304,14 @@ func newSubscribeStarted(keepaliveIntervalMs uint32) *message_api.SubscribeRespo func newSubscribeEnvelopes( envs []*envelopesProto.OriginatorEnvelope, + mutateID uint64, ) *message_api.SubscribeResponse { return wrapSubscribeV1(&message_api.SubscribeResponse_V1{ Response: &message_api.SubscribeResponse_V1_Envelopes_{ - Envelopes: &message_api.SubscribeResponse_V1_Envelopes{Envelopes: envs}, + Envelopes: &message_api.SubscribeResponse_V1_Envelopes{ + Envelopes: envs, + MutateId: mutateID, + }, }, }) } diff --git a/pkg/api/message/subscribe_internal_test.go b/pkg/api/message/subscribe_internal_test.go index 422893fc6..9ca5919bb 100644 --- a/pkg/api/message/subscribe_internal_test.go +++ b/pkg/api/message/subscribe_internal_test.go @@ -12,11 +12,12 @@ import ( "github.com/stretchr/testify/require" "go.uber.org/zap" + "github.com/xmtp/xmtpd/pkg/constants" "github.com/xmtp/xmtpd/pkg/db" - "github.com/xmtp/xmtpd/pkg/db/queries" "github.com/xmtp/xmtpd/pkg/envelopes" envelopesProto "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/envelopes" "github.com/xmtp/xmtpd/pkg/proto/xmtpv4/message_api" + envelopeTestUtils "github.com/xmtp/xmtpd/pkg/testutils/envelopes" "github.com/xmtp/xmtpd/pkg/topic" ) @@ -278,9 +279,11 @@ func TestSubscribeSessionRejectsTooManyInflightWaves(t *testing.T) { } sess.nextWave = maxInflightSubscribeWaves + // Mutate ids past the saturated waves' 0..cap-1 range, so the in-flight mutate_id collision + // check cannot preempt the cap check this test targets. tp := topic.NewTopic(topic.TopicKindGroupMessagesV1, []byte("over-cap")) err := sess.handleMutate(&message_api.SubscribeRequest_V1_Mutate{ - MutateId: 1, + MutateId: maxInflightSubscribeWaves + 1, Adds: []*message_api.SubscribeRequest_V1_Mutate_Subscription{{Topic: tp.Bytes()}}, }) require.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err), @@ -288,52 +291,172 @@ func TestSubscribeSessionRejectsTooManyInflightWaves(t *testing.T) { // A removes-only Mutate creates no wave, so it is not blocked by the cap. err = sess.handleMutate(&message_api.SubscribeRequest_V1_Mutate{ - MutateId: 2, + MutateId: maxInflightSubscribeWaves + 2, Removes: [][]byte{tp.Bytes()}, }) require.NoError(t, err, "a removes-only Mutate must not be blocked by the in-flight cap") } -// TestAdvanceCursorsFromRowsAdvancesPastUnmarshalFailures covers the catch-up spin fix: pagination -// cursors must advance from the RAW rows even when an envelope's bytes don't unmarshal, otherwise a -// bad row in a full page re-fetches forever and the wave never completes. -func TestAdvanceCursorsFromRowsAdvancesPastUnmarshalFailures(t *testing.T) { - tp := topic.NewTopic(topic.TopicKindGroupMessagesV1, []byte("rows-topic")) - key := string(tp.Bytes()) - cursors := db.TopicCursors{key: make(db.VectorClock)} - // Envelope bytes are intentionally garbage (would be dropped by unmarshalEnvelopes). - rows := []queries.GatewayEnvelopesView{ - { - Topic: tp.Bytes(), - OriginatorNodeID: 100, - OriginatorSequenceID: 5, - OriginatorEnvelope: []byte("garbage"), - }, - { - Topic: tp.Bytes(), - OriginatorNodeID: 100, - OriginatorSequenceID: 6, - OriginatorEnvelope: []byte("garbage"), +// TestSubscribeSessionMutateAddsCapRejected covers the adds-per-Mutate cap (review finding): a +// wave's merged catch-up scan flattens one floor entry per (topic, originator) pair into EVERY +// page query, so the raw adds one Mutate may carry are bounded. An over-cap Mutate must be +// rejected with ResourceExhausted BEFORE any state changes — atomically, so not even the removes +// riding the same Mutate apply — and an at-cap Mutate must succeed (wave created, topics gated). +func TestSubscribeSessionMutateAddsCapRejected(t *testing.T) { + ctx := context.Background() + sess := &subscribeSession{ + svc: &Service{ctx: ctx, originatorList: stubOriginatorLister{}}, + logger: zap.NewNop(), + ctx: ctx, + keepAlive: time.Second, + outbound: make(chan *message_api.SubscribeResponse, 16), + catchUpCh: make(chan catchUpBatch, subscribeCatchUpQueueDepth), + topics: make(map[string]*topicState), + waves: make(map[int]*subscribeWave), + maxAdds: 3, + // Minimal sub so gateTopic/removeTopic's worker (un)register is a no-op, not a nil deref. + sub: &mutableSubscription{ + worker: &subscribeWorker{}, + l: &listener{ + topics: make(map[string]struct{}), + originators: make(map[uint32]struct{}), + }, }, - { - Topic: tp.Bytes(), - OriginatorNodeID: 200, - OriginatorSequenceID: 3, - OriginatorEnvelope: []byte("garbage"), + } + + // A topic already live on the stream; the over-cap Mutate below also tries to remove it. + pre := wbTopic("cap-pre") + sess.topics[pre.cursorKey] = &topicState{ + subscribeTopic: pre, + phase: topicLive, + cursor: make(db.VectorClock), + } + + adds := func(names ...string) []*message_api.SubscribeRequest_V1_Mutate_Subscription { + out := make([]*message_api.SubscribeRequest_V1_Mutate_Subscription, 0, len(names)) + for _, n := range names { + out = append(out, &message_api.SubscribeRequest_V1_Mutate_Subscription{ + Topic: wbTopic(n).wire, + }) + } + return out + } + + // 4 adds > cap of 3: rejected — and atomically, so the remove riding the same Mutate must + // not have applied either. + err := sess.handleMutate(&message_api.SubscribeRequest_V1_Mutate{ + MutateId: 1, + Adds: adds("cap-a", "cap-b", "cap-c", "cap-d"), + Removes: [][]byte{pre.wire}, + }) + require.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err), + "a Mutate over the adds cap must be rejected with ResourceExhausted") + require.Len(t, sess.topics, 1, "no topic from the rejected Mutate may be registered") + require.Contains(t, sess.topics, pre.cursorKey, + "the rejected Mutate's removes must not have applied (atomicity)") + require.Empty(t, sess.waves, "the rejected Mutate must not have created a wave") + require.Zero(t, sess.nextWave) + require.Empty(t, drainResponses(sess.outbound), "a rejected Mutate must emit no frames") + + // Exactly at the cap: accepted — wave created, its topics gated. + require.NoError(t, sess.handleMutate(&message_api.SubscribeRequest_V1_Mutate{ + MutateId: 2, + Adds: adds("cap-a", "cap-b", "cap-c"), + })) + require.Len(t, sess.waves, 1, "an at-cap Mutate must create its wave") + require.Len(t, sess.topics, 4, "the at-cap Mutate's topics must be registered") + for _, n := range []string{"cap-a", "cap-b", "cap-c"} { + ts := sess.topics[wbTopic(n).cursorKey] + require.NotNil(t, ts) + require.Equal(t, topicGated, ts.phase) + } + // The wave's fetcher started; the stub lister fails it immediately (no DB in a white-box + // test), which also synchronizes the detached goroutine's exit before the test returns. + b := <-sess.catchUpCh + require.Error(t, b.err) +} + +// TestSubscribeSessionMutateCursorEntriesCapRejected covers the adds cap's companion bound: the +// add COUNT cap alone does not bound a wave's (topic, originator) floor pairs, because a single +// add's cursor is a per-originator vector that may name arbitrarily many originators. An over-cap +// Mutate must be rejected with ResourceExhausted before any state changes — atomically, like the +// adds cap — and an at-cap Mutate must succeed. +func TestSubscribeSessionMutateCursorEntriesCapRejected(t *testing.T) { + ctx := context.Background() + sess := &subscribeSession{ + svc: &Service{ctx: ctx, originatorList: stubOriginatorLister{}}, + logger: zap.NewNop(), + ctx: ctx, + keepAlive: time.Second, + outbound: make(chan *message_api.SubscribeResponse, 16), + catchUpCh: make(chan catchUpBatch, subscribeCatchUpQueueDepth), + topics: make(map[string]*topicState), + waves: make(map[int]*subscribeWave), + maxCursorEntries: 3, + // Minimal sub so gateTopic/removeTopic's worker (un)register is a no-op, not a nil deref. + sub: &mutableSubscription{ + worker: &subscribeWorker{}, + l: &listener{ + topics: make(map[string]struct{}), + originators: make(map[uint32]struct{}), + }, }, } - advanceCursorsFromRows(cursors, rows) - require.Equal( - t, - uint64(6), - cursors[key][100], - "cursor must advance to the max raw seq for an originator", - ) - require.Equal(t, uint64(3), cursors[key][200]) + + // A topic already live on the stream; the over-cap Mutate below also tries to remove it. + pre := wbTopic("ce-pre") + sess.topics[pre.cursorKey] = &topicState{ + subscribeTopic: pre, + phase: topicLive, + cursor: make(db.VectorClock), + } + + // ONE add whose cursor names n originators: the add count stays far under maxMutateAdds, so + // only the cursor-entry cap can reject it. + addWithEntries := func(name string, n int) []*message_api.SubscribeRequest_V1_Mutate_Subscription { + c := make(map[uint32]uint64, n) + for i := range n { + c[uint32(100+i)] = uint64(i + 1) + } + return []*message_api.SubscribeRequest_V1_Mutate_Subscription{ + {Topic: wbTopic(name).wire, LastSeen: &envelopesProto.Cursor{NodeIdToSequenceId: c}}, + } + } + + // 4 entries > cap of 3: rejected — and atomically, so the remove riding the same Mutate must + // not have applied either. + err := sess.handleMutate(&message_api.SubscribeRequest_V1_Mutate{ + MutateId: 1, + Adds: addWithEntries("ce-a", 4), + Removes: [][]byte{pre.wire}, + }) + require.Equal(t, connect.CodeResourceExhausted, connect.CodeOf(err), + "a Mutate over the cursor-entry cap must be rejected with ResourceExhausted") + require.Len(t, sess.topics, 1, "no topic from the rejected Mutate may be registered") + require.Contains(t, sess.topics, pre.cursorKey, + "the rejected Mutate's removes must not have applied (atomicity)") + require.Empty(t, sess.waves, "the rejected Mutate must not have created a wave") + require.Zero(t, sess.nextWave) + require.Empty(t, drainResponses(sess.outbound), "a rejected Mutate must emit no frames") + + // Exactly at the cap: accepted — wave created, its topic gated. + require.NoError(t, sess.handleMutate(&message_api.SubscribeRequest_V1_Mutate{ + MutateId: 2, + Adds: addWithEntries("ce-b", 3), + })) + require.Len(t, sess.waves, 1, "an at-cap Mutate must create its wave") + ts := sess.topics[wbTopic("ce-b").cursorKey] + require.NotNil(t, ts) + require.Equal(t, topicGated, ts.phase) + // The wave's fetcher started; the stub lister fails it immediately (no DB in a white-box + // test), which also synchronizes the detached goroutine's exit before the test returns. + b := <-sess.catchUpCh + require.Error(t, b.err) } // TestSubscribeSessionSendEnvelopesSplitsFrames covers the frame-splitting fix: a batch larger than -// the frame limit must go out as multiple frames, never one oversized (stream-aborting) frame. +// the frame limit must go out as multiple frames, never one oversized (stream-aborting) frame — and +// every split frame must carry the caller's wave tag (XIP-83 server requirement 3). func TestSubscribeSessionSendEnvelopesSplitsFrames(t *testing.T) { ctx := context.Background() sess := &subscribeSession{ @@ -346,7 +469,7 @@ func TestSubscribeSessionSendEnvelopesSplitsFrames(t *testing.T) { } require.NoError(t, sess.sendEnvelopes([]*envelopesProto.OriginatorEnvelope{ mkEnv(50), mkEnv(50), mkEnv(50), - })) + }, 5)) var counts []int total := 0 @@ -354,12 +477,397 @@ func TestSubscribeSessionSendEnvelopesSplitsFrames(t *testing.T) { if env := f.GetV1().GetEnvelopes(); env != nil { counts = append(counts, len(env.GetEnvelopes())) total += len(env.GetEnvelopes()) + require.Equal(t, uint64(5), env.GetMutateId(), "every split frame carries the wave tag") } } require.Len(t, counts, 2, "batch must be split into 2 frames") require.Equal(t, 3, total, "every envelope must be delivered exactly once") } +// TestSubscribeSessionSendEnvelopesSkipsOverCapEnvelope covers frame splitting's hard-cap corner: +// publish admission bounds what the payer sent, not the stored envelope (originator wrapper) plus +// response framing, so a lone envelope can exceed the transport's send cap — and sending it would +// abort the stream, with every reconnecting wave hitting the same row again. It must be skipped +// with a warning (like the legacy batchAndSendEnvelopes) while its neighbors are still delivered +// in order under the caller's wave tag. +func TestSubscribeSessionSendEnvelopesSkipsOverCapEnvelope(t *testing.T) { + ctx := context.Background() + sess := &subscribeSession{ + svc: &Service{ctx: ctx}, logger: zap.NewNop(), ctx: ctx, keepAlive: time.Second, + outbound: make(chan *message_api.SubscribeResponse, 16), + } + mkEnv := func(payload int) *envelopesProto.OriginatorEnvelope { + return &envelopesProto.OriginatorEnvelope{UnsignedOriginatorEnvelope: make([]byte, payload)} + } + require.NoError(t, sess.sendEnvelopes([]*envelopesProto.OriginatorEnvelope{ + mkEnv(50), mkEnv(constants.GRPCPayloadLimit + 1024), mkEnv(60), + }, 9)) + + var sizes []int + for _, f := range drainResponses(sess.outbound) { + env := f.GetV1().GetEnvelopes() + require.NotNil(t, env) + require.Equal(t, uint64(9), env.GetMutateId(), "surviving frames must keep the wave tag") + for _, e := range env.GetEnvelopes() { + sizes = append(sizes, len(e.GetUnsignedOriginatorEnvelope())) + } + } + require.Equal(t, []int{50, 60}, sizes, + "the over-cap envelope must be skipped; its neighbors delivered in order") +} + +// TestSubscribeSessionFoldTagsAndOrders is the deterministic pin on the wave-completion fold: the +// live envelopes buffered while the wave's topics were gated go out stamped with the wave's +// mutate_id, merged into per-originator sequence order across topics, then TopicsLive, then +// CatchupComplete — and never on the live (tag-0) lane. +func TestSubscribeSessionFoldTagsAndOrders(t *testing.T) { + t1, t2 := wbTopic("fold-t1"), wbTopic("fold-t2") + sess := newWaveTestSession(0, 42, t1, t2) + + // One scan page: cursors advance to (100 -> 1) on t1 and (100 -> 2) on t2. + done, err := sess.handleCatchUp(catchUpBatch{wave: 0, envs: []*envelopes.OriginatorEnvelope{ + wbEnv(t, t1, 100, 1), + wbEnv(t, t2, 100, 2), + }}) + require.NoError(t, err) + require.False(t, done) + page := drainResponses(sess.outbound) + require.Equal(t, [][2]uint64{{100, 1}, {100, 2}}, wbEnvelopeKeys(t, page, 42)) + require.Empty(t, wbEnvelopeKeys(t, page, 0), "no scan page may ride the live tag") + + // Live envelopes for the gated topics: buffered, nothing sent. + require.NoError(t, sess.routeLive([]*envelopes.OriginatorEnvelope{ + wbEnv(t, t1, 100, 4), + wbEnv(t, t2, 100, 3), + wbEnv(t, t2, 200, 1), + })) + require.Empty(t, drainResponses(sess.outbound), "gated live envelopes must be withheld") + require.Positive(t, sess.pendingBytes) + + done, err = sess.handleCatchUp(catchUpBatch{wave: 0, done: true}) + require.NoError(t, err) + require.False(t, done) + + frames := drainResponses(sess.outbound) + require.Equal(t, [][2]uint64{{100, 3}, {100, 4}, {200, 1}}, wbEnvelopeKeys(t, frames, 42), + "fold must merge the pending buffers into per-originator sequence order, tagged 42") + require.Empty(t, wbEnvelopeKeys(t, frames, 0), "no fold envelope may ride the live tag") + + lastEnv, liveIdx, ccIdx := -1, -1, -1 + for i, f := range frames { + switch { + case f.GetV1().GetEnvelopes() != nil: + lastEnv = i + case f.GetV1().GetTopicsLive() != nil: + liveIdx = i + case f.GetV1().GetCatchupComplete() != nil: + ccIdx = i + } + } + require.True(t, lastEnv < liveIdx && liveIdx < ccIdx, + "fold frames must precede TopicsLive, which precedes CatchupComplete (%d, %d, %d)", + lastEnv, liveIdx, ccIdx) + require.Equal(t, [][]byte{t1.wire, t2.wire}, topicsLiveFrames(frames)) + require.Equal(t, []uint64{42}, catchupCompleteIDs(frames)) + + require.Zero(t, sess.pendingBytes, "the fold must refund every buffered byte") + require.Equal(t, topicLive, sess.topics[t1.cursorKey].phase) + require.Equal(t, topicLive, sess.topics[t2.cursorKey].phase) +} + +// TestSubscribeSessionFoldDedupsScanDeliveredPending pins exactly-once when the same envelope +// travels both lanes (a lagging worker dispatches a row the scan already delivered): the fold's +// cursor dedup must drop it, delivering only the genuinely-new pending envelope. +func TestSubscribeSessionFoldDedupsScanDeliveredPending(t *testing.T) { + tp := wbTopic("dual-lane") + sess := newWaveTestSession(0, 7, tp) + + done, err := sess.handleCatchUp(catchUpBatch{wave: 0, envs: []*envelopes.OriginatorEnvelope{ + wbEnv(t, tp, 100, 1), + wbEnv(t, tp, 100, 2), + }}) + require.NoError(t, err) + require.False(t, done) + drainResponses(sess.outbound) + + // The worker lagged: it dispatches (100,2) — already delivered by the scan — plus (100,3). + require.NoError(t, sess.routeLive([]*envelopes.OriginatorEnvelope{ + wbEnv(t, tp, 100, 2), + wbEnv(t, tp, 100, 3), + })) + + done, err = sess.handleCatchUp(catchUpBatch{wave: 0, done: true}) + require.NoError(t, err) + require.False(t, done) + + frames := drainResponses(sess.outbound) + require.Equal(t, [][2]uint64{{100, 3}}, wbEnvelopeKeys(t, frames, 7), + "the scan-delivered (100,2) must not be resent by the fold") + require.Equal(t, uint64(3), sess.topics[tp.cursorKey].cursor[100]) +} + +// TestSubscribeSessionNewOriginatorMidWaveFoldsTagged: an originator that first publishes +// mid-wave has no ceiling row and no cursor entry; its gated envelopes must flow through the +// fold (tagged) and seed the live cursor, so its subsequent live tail is neither duplicated +// nor dropped. +func TestSubscribeSessionNewOriginatorMidWaveFoldsTagged(t *testing.T) { + tp := wbTopic("new-orig") + sess := newWaveTestSession(0, 9, tp) + + done, err := sess.handleCatchUp(catchUpBatch{wave: 0, envs: []*envelopes.OriginatorEnvelope{ + wbEnv(t, tp, 100, 1), + }}) + require.NoError(t, err) + require.False(t, done) + drainResponses(sess.outbound) + + // Originator 200 first publishes mid-wave: the wave's ceilings never saw it. + require.NoError(t, sess.routeLive([]*envelopes.OriginatorEnvelope{ + wbEnv(t, tp, 200, 1), + wbEnv(t, tp, 200, 2), + })) + + done, err = sess.handleCatchUp(catchUpBatch{wave: 0, done: true}) + require.NoError(t, err) + require.False(t, done) + frames := drainResponses(sess.outbound) + require.Equal(t, [][2]uint64{{200, 1}, {200, 2}}, wbEnvelopeKeys(t, frames, 9), + "a mid-wave originator's gated envelopes fold in ascending order, tagged") + require.Equal(t, uint64(2), sess.topics[tp.cursorKey].cursor[200]) + + // The topic is live now: the originator's tail arrives on the live (tag-0) lane. + require.NoError(t, sess.routeLive([]*envelopes.OriginatorEnvelope{ + wbEnv(t, tp, 200, 3), + })) + live := drainResponses(sess.outbound) + require.Equal(t, [][2]uint64{{200, 3}}, wbEnvelopeKeys(t, live, 0)) + require.Empty(t, wbEnvelopeKeys(t, live, 9)) +} + +// TestSubscribeSessionRemoveMidWaveDisposesPending pins the pending-bytes budget across a +// mid-wave remove: dropping the topic refunds its buffered bytes, the buffered envelopes are +// never delivered, and the orphaned wave still acks its own CatchupComplete without announcing +// or flushing anything. +func TestSubscribeSessionRemoveMidWaveDisposesPending(t *testing.T) { + tp := wbTopic("remove-mid-wave") + sess := newWaveTestSession(0, 5, tp) + // Minimal sub so removeTopic's worker unregister is a no-op, not a nil deref. + sess.sub = &mutableSubscription{ + worker: &subscribeWorker{}, + l: &listener{ + topics: make(map[string]struct{}), + originators: make(map[uint32]struct{}), + }, + } + + require.NoError(t, sess.routeLive([]*envelopes.OriginatorEnvelope{ + wbEnv(t, tp, 100, 1), + wbEnv(t, tp, 100, 2), + })) + require.Positive(t, sess.pendingBytes) + + // A removes-only Mutate is acked with its own CatchupComplete; the pending budget is refunded. + require.NoError(t, sess.handleMutate(&message_api.SubscribeRequest_V1_Mutate{ + MutateId: 6, + Removes: [][]byte{tp.wire}, + })) + require.Zero(t, sess.pendingBytes, "removing a gated topic must refund its pending bytes") + require.Equal(t, []uint64{6}, catchupCompleteIDs(drainResponses(sess.outbound))) + + done, err := sess.handleCatchUp(catchUpBatch{wave: 0, done: true}) + require.NoError(t, err) + require.False(t, done) + frames := drainResponses(sess.outbound) + for _, f := range frames { + require.Nil(t, f.GetV1().GetEnvelopes(), + "the removed topic's buffered envelopes must never be delivered") + } + require.Empty(t, topicsLiveFrames(frames), "an orphaned wave announces nothing") + require.Equal(t, []uint64{5}, catchupCompleteIDs(frames), + "the orphaned wave still acks its own CatchupComplete") +} + +// TestSubscribeSessionDBErrorMidWaveFailsStreamNoCatchupComplete: a fetch error mid-wave must +// fail the stream (the client reconnects from its cursors) and never emit the wave's +// CatchupComplete over a history gap. +func TestSubscribeSessionDBErrorMidWaveFailsStreamNoCatchupComplete(t *testing.T) { + tp := wbTopic("db-err") + sess := newWaveTestSession(0, 13, tp) + + done, err := sess.handleCatchUp(catchUpBatch{wave: 0, err: errors.New("db down")}) + require.False(t, done) + require.Equal(t, connect.CodeUnavailable, connect.CodeOf(err)) + require.Empty(t, catchupCompleteIDs(drainResponses(sess.outbound)), + "a failed wave must never emit CatchupComplete") +} + +// TestSubscribeSessionSendErrorMidFoldNoCatchupComplete: if the sender dies while the fold is +// being delivered, the error must propagate and the wave's CatchupComplete must not be emitted — +// the client must never believe it is synced past an undelivered fold. +func TestSubscribeSessionSendErrorMidFoldNoCatchupComplete(t *testing.T) { + tp := wbTopic("send-err") + sess := newWaveTestSession(0, 11, tp) + // Unbuffered outbound with no reader plus a dead sender: send() can only observe senderDone. + sess.outbound = make(chan *message_api.SubscribeResponse) + sess.senderDone = make(chan struct{}) + wantErr := errors.New("stream send failed") + sess.sendErr = wantErr + close(sess.senderDone) + + require.NoError(t, sess.routeLive([]*envelopes.OriginatorEnvelope{ + wbEnv(t, tp, 100, 1), + })) + + done, err := sess.handleCatchUp(catchUpBatch{wave: 0, done: true}) + require.False(t, done) + require.ErrorIs(t, err, wantErr, "the sender's error must surface from the fold") + require.Empty(t, catchupCompleteIDs(drainResponses(sess.outbound)), + "CatchupComplete must not follow an undelivered fold") +} + +// TestSubscribeSessionInflightMutateIdCollisionRejected covers the in-flight mutate_id collision +// rule (XIP-83 server requirement 3): a Mutate reusing the mutate_id of a wave still in flight is +// rejected — the two waves' replay frames and CatchupComplete acks would be indistinguishable — +// atomically, BEFORE any state changes. Even a removes-only reuse is rejected (its immediate +// CatchupComplete would be ambiguous with the in-flight wave's). Reuse after the wave's +// CatchupComplete stays legal. +func TestSubscribeSessionInflightMutateIdCollisionRejected(t *testing.T) { + pre := wbTopic("collide-pre") + sess := newWaveTestSession(0, 7, pre) // wave 0 in flight with mutateID 7, gating `pre` + sess.svc = &Service{ctx: sess.ctx, originatorList: stubOriginatorLister{}} + sess.catchUpCh = make(chan catchUpBatch, subscribeCatchUpQueueDepth) + // Minimal sub so gateTopic/removeTopic's worker (un)register is a no-op, not a nil deref. + sess.sub = &mutableSubscription{ + worker: &subscribeWorker{}, + l: &listener{ + topics: make(map[string]struct{}), + originators: make(map[uint32]struct{}), + }, + } + + // Adds riding the in-flight id 7: rejected, and no state may have changed. + fresh := wbTopic("collide-new") + err := sess.handleMutate(&message_api.SubscribeRequest_V1_Mutate{ + MutateId: 7, + Adds: []*message_api.SubscribeRequest_V1_Mutate_Subscription{{Topic: fresh.wire}}, + }) + require.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err), + "a Mutate reusing an in-flight mutate_id must be rejected") + require.Len(t, sess.topics, 1, "the rejected Mutate must register no topic") + require.NotContains(t, sess.topics, fresh.cursorKey) + require.Len(t, sess.waves, 1, "the rejected Mutate must not create a wave") + require.Contains(t, sess.waves, 0) + require.Equal(t, 1, sess.nextWave) + require.Empty(t, drainResponses(sess.outbound), "a rejected Mutate must emit no frames") + + // A removes-only reuse is rejected the same way, and its remove must not have applied. + err = sess.handleMutate(&message_api.SubscribeRequest_V1_Mutate{ + MutateId: 7, + Removes: [][]byte{pre.wire}, + }) + require.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err), + "a removes-only Mutate reusing an in-flight mutate_id must be rejected") + require.Contains(t, sess.topics, pre.cursorKey, + "the rejected Mutate's removes must not have applied") + require.Empty(t, drainResponses(sess.outbound)) + + // Wave 7 completes (its CatchupComplete goes out): id 7 is reusable again. + done, err := sess.handleCatchUp(catchUpBatch{wave: 0, done: true}) + require.NoError(t, err) + require.False(t, done) + require.Equal(t, []uint64{7}, catchupCompleteIDs(drainResponses(sess.outbound))) + + require.NoError(t, sess.handleMutate(&message_api.SubscribeRequest_V1_Mutate{ + MutateId: 7, + Adds: []*message_api.SubscribeRequest_V1_Mutate_Subscription{{Topic: fresh.wire}}, + }), "reusing a mutate_id after its wave's CatchupComplete must be accepted") + require.Contains(t, sess.waves, 1, "the reused id's wave must be created") + // The wave's fetcher started; the stub lister fails it immediately (no DB in a white-box + // test), which also synchronizes the detached goroutine's exit before the test returns. + b := <-sess.catchUpCh + require.Error(t, b.err) +} + +// newWaveTestSession builds a writer-owned session with one in-flight live wave owning the given +// gated topics — the state gateTopic/handleMutate would have left, minus the fetcher goroutine +// (tests feed catchUpBatches directly). +func newWaveTestSession(wave int, mutateID uint64, topics ...subscribeTopic) *subscribeSession { + ctx := context.Background() + sess := &subscribeSession{ + svc: &Service{ctx: ctx}, + logger: zap.NewNop(), + ctx: ctx, + keepAlive: time.Second, + outbound: make(chan *message_api.SubscribeResponse, 16), + topics: make(map[string]*topicState), + waves: make(map[int]*subscribeWave), + } + sess.waves[wave] = &subscribeWave{mutateID: mutateID, topics: topics} + for _, st := range topics { + sess.topics[st.cursorKey] = &topicState{ + subscribeTopic: st, + phase: topicGated, + wave: wave, + cursor: make(db.VectorClock), + } + } + sess.nextWave = wave + 1 + return sess +} + +// stubOriginatorLister gives handleMutate's detached fetcher goroutine a terminal originator +// lookup (an immediate error) so it exits without ever touching a DB. +type stubOriginatorLister struct{} + +func (stubOriginatorLister) GetOriginatorNodeIDs(context.Context) ([]uint32, error) { + return nil, errors.New("no originator list in white-box tests") +} + +func wbTopic(name string) subscribeTopic { + tp := topic.NewTopic(topic.TopicKindGroupMessagesV1, []byte(name)) + return subscribeTopic{wire: tp.Bytes(), cursorKey: string(tp.Bytes()), listenKey: tp.String()} +} + +func wbEnv( + t *testing.T, + st subscribeTopic, + nodeID uint32, + seqID uint64, +) *envelopes.OriginatorEnvelope { + t.Helper() + env, err := envelopes.NewOriginatorEnvelope( + envelopeTestUtils.CreateOriginatorEnvelopeWithTopic(t, nodeID, seqID, st.wire), + ) + require.NoError(t, err) + return env +} + +// wbEnvelopeKeys returns the (originator, sequence) keys of envelopes carried by Envelopes frames +// stamped with the given wave tag, in frame order. +func wbEnvelopeKeys( + t *testing.T, + frames []*message_api.SubscribeResponse, + tag uint64, +) [][2]uint64 { + t.Helper() + var keys [][2]uint64 + for _, f := range frames { + env := f.GetV1().GetEnvelopes() + if env == nil || env.GetMutateId() != tag { + continue + } + for _, e := range env.GetEnvelopes() { + parsed, err := envelopes.NewOriginatorEnvelope(e) + require.NoError(t, err) + keys = append( + keys, + [2]uint64{uint64(parsed.OriginatorNodeID()), parsed.OriginatorSequenceID()}, + ) + } + } + return keys +} + func drainResponses(ch chan *message_api.SubscribeResponse) []*message_api.SubscribeResponse { var out []*message_api.SubscribeResponse for { diff --git a/pkg/api/message/subscribe_test.go b/pkg/api/message/subscribe_test.go index 08f0ce191..b718d9019 100644 --- a/pkg/api/message/subscribe_test.go +++ b/pkg/api/message/subscribe_test.go @@ -1667,6 +1667,17 @@ func TestSubscribe_HalfCloseHistoryOnlyDrains(t *testing.T) { require.True(t, hasEnvKey(keys, 100, 1) && hasEnvKey(keys, 200, 1), "history must be delivered") require.True(t, hasTopicBytes(subTopicsLive(frames), topicA)) require.True(t, hasMutateID(subCatchupCompletes(frames), 7)) + + // Every data frame carries the wave's mutate_id: history_only pages sent with the live + // tag (0) would satisfy the delivery assertions above but violate XIP-83 requirement 3. + require.Empty(t, subEnvelopeKeysTagged(t, frames, 0), + "history_only pages must never ride the live tag") + for _, f := range frames { + if env := f.GetV1().GetEnvelopes(); env != nil { + require.Equal(t, uint64(7), env.GetMutateId(), + "every history_only data frame carries the wave's mutate_id") + } + } } // TestSubscribe_HistoryOnlyOnLiveRejected verifies a history_only add targeting a topic already @@ -1784,3 +1795,663 @@ func mustFrames(r *bidiReader) []*message_api.SubscribeResponse { frames, _ := r.snapshot() return frames } + +// ---- XIP-83 delivery tagging & ordering (server requirements 3 and 4) ---- + +// envRow builds one insertable gateway envelope row for the tagging/ordering tests. +func envRow( + t *testing.T, + payerID sql.NullInt32, + nodeID int32, + seqID int64, + topicBytes []byte, +) queries.InsertGatewayEnvelopeV3Params { + t.Helper() + return queries.InsertGatewayEnvelopeV3Params{ + OriginatorNodeID: nodeID, + OriginatorSequenceID: seqID, + Topic: topicBytes, + PayerID: payerID, + OriginatorEnvelope: testutils.Marshal( + t, + envelopeTestUtils.CreateOriginatorEnvelopeWithTopic( + t, + uint32(nodeID), + uint64(seqID), + topicBytes, + ), + ), + } +} + +// subEnvelopeKeysTagged returns the (originator, sequence) keys of envelopes carried by +// Envelopes frames stamped with the given wave tag, in receive order. +func subEnvelopeKeysTagged( + t *testing.T, + frames []*message_api.SubscribeResponse, + tag uint64, +) [][2]uint64 { + t.Helper() + var keys [][2]uint64 + for _, f := range frames { + env := f.GetV1().GetEnvelopes() + if env == nil || env.GetMutateId() != tag { + continue + } + for _, e := range env.GetEnvelopes() { + u := envelopeTestUtils.UnmarshalUnsignedOriginatorEnvelope( + t, + e.GetUnsignedOriginatorEnvelope(), + ) + keys = append( + keys, + [2]uint64{uint64(u.GetOriginatorNodeId()), u.GetOriginatorSequenceId()}, + ) + } + } + return keys +} + +// requirePerOriginatorAscending asserts each originator's sequence ids strictly ascend in +// the given key order — the total-order shape both delivery lanes guarantee per originator. +func requirePerOriginatorAscending(t *testing.T, keys [][2]uint64, desc string) { + t.Helper() + last := make(map[uint64]uint64) + for _, k := range keys { + if prev, ok := last[k[0]]; ok { + require.Greater(t, k[1], prev, + "%s: originator %d sequences must strictly ascend", desc, k[0]) + } + last[k[0]] = k[1] + } +} + +// requireExactlyOnce asserts no (originator, sequence) key appears twice across the keys. +func requireExactlyOnce(t *testing.T, keys [][2]uint64, desc string) { + t.Helper() + seen := make(map[[2]uint64]struct{}, len(keys)) + for _, k := range keys { + _, dup := seen[k] + require.Falsef(t, dup, "%s: envelope %v delivered more than once", desc, k) + seen[k] = struct{}{} + } +} + +// TestSubscribe_ReplayTaggedWithWaveLiveTaggedZero pins delivery tagging under overlapping +// waves: every replay frame carries exactly the mutate_id of the wave that produced it, and +// live frames carry 0. +func TestSubscribe_ReplayTaggedWithWaveLiveTaggedZero(t *testing.T) { + suite := setupTest(t) + payerID := db.NullInt32(testutils.CreatePayer(t, suite.DB)) + + // History: topicA fed by originator 100, topicB by originator 200. + testutils.InsertGatewayEnvelopes(t, suite.DB, []queries.InsertGatewayEnvelopeV3Params{ + envRow(t, payerID, 100, 1, topicA), + envRow(t, payerID, 100, 2, topicA), + envRow(t, payerID, 100, 3, topicA), + envRow(t, payerID, 200, 1, topicB), + envRow(t, payerID, 200, 2, topicB), + }) + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + require.NoError(t, suite.MessageService.AwaitCursor(ctx, db.VectorClock{100: 3, 200: 2})) + + stream := suite.ClientQuery.Subscribe(t.Context()) + reader := newBidiReader(stream) + + // Two overlapping waves, back to back: their replays race on the same stream. + require.NoError(t, stream.Send(subMutate( + 7, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{addSub(topicA, nil)}, + nil, + ))) + require.NoError(t, stream.Send(subMutate( + 8, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{addSub(topicB, nil)}, + nil, + ))) + require.Eventually(t, func() bool { + cc := subCatchupCompletes(mustFrames(reader)) + return hasMutateID(cc, 7) && hasMutateID(cc, 8) + }, 10*time.Second, 20*time.Millisecond, "both waves must complete") + + frames := mustFrames(reader) + wave7 := subEnvelopeKeysTagged(t, frames, 7) + wave8 := subEnvelopeKeysTagged(t, frames, 8) + require.Len(t, wave7, 3, "wave 7 delivers exactly topicA's history, tagged 7") + require.Len(t, wave8, 2, "wave 8 delivers exactly topicB's history, tagged 8") + for _, k := range wave7 { + require.Equal(t, uint64(100), k[0], "wave 7 must carry only topicA's originator") + } + for _, k := range wave8 { + require.Equal(t, uint64(200), k[0], "wave 8 must carry only topicB's originator") + } + require.Empty(t, subEnvelopeKeysTagged(t, frames, 0), "no replay frame may carry the live tag") + + // Live tail on both topics is tagged 0. + testutils.InsertGatewayEnvelopes(t, suite.DB, []queries.InsertGatewayEnvelopeV3Params{ + envRow(t, payerID, 100, 4, topicA), + envRow(t, payerID, 200, 3, topicB), + }) + require.Eventually(t, func() bool { + live := subEnvelopeKeysTagged(t, mustFrames(reader), 0) + return hasEnvKey(live, 100, 4) && hasEnvKey(live, 200, 3) + }, 10*time.Second, 20*time.Millisecond, "live tail must be tagged 0") + + requireExactlyOnce(t, subEnvelopeKeys(t, mustFrames(reader)), "all lanes") +} + +// TestSubscribe_WaveReplayPerOriginatorOrderAcrossTopics pins wave order: one wave covering +// two topics whose envelopes interleave per originator must replay each originator's +// envelopes in ascending sequence order across BOTH topics — one merged cursor-ordered +// pass, not one topic's burst then the other's. +func TestSubscribe_WaveReplayPerOriginatorOrderAcrossTopics(t *testing.T) { + suite := setupTest(t) + payerID := db.NullInt32(testutils.CreatePayer(t, suite.DB)) + + // Originator 100 alternates between the two topics; originator 200 interleaves too. + testutils.InsertGatewayEnvelopes(t, suite.DB, []queries.InsertGatewayEnvelopeV3Params{ + envRow(t, payerID, 100, 1, topicA), + envRow(t, payerID, 100, 2, topicB), + envRow(t, payerID, 100, 3, topicA), + envRow(t, payerID, 100, 4, topicB), + envRow(t, payerID, 100, 5, topicA), + envRow(t, payerID, 100, 6, topicB), + envRow(t, payerID, 200, 1, topicB), + envRow(t, payerID, 200, 2, topicA), + }) + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + require.NoError(t, suite.MessageService.AwaitCursor(ctx, db.VectorClock{100: 6, 200: 2})) + + stream := suite.ClientQuery.Subscribe(t.Context()) + reader := newBidiReader(stream) + + require.NoError(t, stream.Send(subMutate( + 3, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{ + addSub(topicA, nil), + addSub(topicB, nil), + }, + nil, + ))) + require.Eventually(t, func() bool { + return hasMutateID(subCatchupCompletes(mustFrames(reader)), 3) + }, 10*time.Second, 20*time.Millisecond) + + replay := subEnvelopeKeysTagged(t, mustFrames(reader), 3) + require.Len(t, replay, 8, "the wave delivers both topics' history, tagged 3") + requirePerOriginatorAscending(t, replay, "wave replay across interleaved topics") + requireExactlyOnce(t, replay, "wave replay") +} + +// TestSubscribe_LivePerOriginatorOrderAcrossTopics pins live order: live (mutate_id 0) +// envelopes across all live topics arrive in ascending sequence order per originator. +func TestSubscribe_LivePerOriginatorOrderAcrossTopics(t *testing.T) { + suite := setupTest(t) + payerID := db.NullInt32(testutils.CreatePayer(t, suite.DB)) + + stream := suite.ClientQuery.Subscribe(t.Context()) + reader := newBidiReader(stream) + + require.NoError(t, stream.Send(subMutate( + 1, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{ + addSub(topicA, nil), + addSub(topicB, nil), + }, + nil, + ))) + require.Eventually(t, func() bool { + return hasMutateID(subCatchupCompletes(mustFrames(reader)), 1) + }, 10*time.Second, 20*time.Millisecond) + + // Each originator alternates between the two live topics. + testutils.InsertGatewayEnvelopes(t, suite.DB, []queries.InsertGatewayEnvelopeV3Params{ + envRow(t, payerID, 100, 1, topicA), + envRow(t, payerID, 100, 2, topicB), + envRow(t, payerID, 100, 3, topicA), + envRow(t, payerID, 200, 1, topicB), + envRow(t, payerID, 200, 2, topicA), + }) + require.Eventually(t, func() bool { + live := subEnvelopeKeysTagged(t, mustFrames(reader), 0) + return hasEnvKey(live, 100, 3) && hasEnvKey(live, 200, 2) + }, 10*time.Second, 20*time.Millisecond) + + live := subEnvelopeKeysTagged(t, mustFrames(reader), 0) + require.Len(t, live, 5) + requirePerOriginatorAscending(t, live, "live lane across topics") + requireExactlyOnce(t, live, "live lane") +} + +// TestSubscribe_SeamLiveWaitsForCatchupComplete pins the seam: while a wave replays a +// topic, the topic speaks live (mutate_id 0) only after the wave's CatchupComplete. +// Envelopes published mid-wave arrive exactly once — either folded into the wave (tagged) +// or live after its CatchupComplete — and never as a live frame before it. A pre-live +// sentinel topic publishes throughout: the gate is per-topic, so live delivery for other +// subscriptions keeps flowing while the wave's topic is gated. +func TestSubscribe_SeamLiveWaitsForCatchupComplete(t *testing.T) { + suite := setupTest(t) + payerID := db.NullInt32(testutils.CreatePayer(t, suite.DB)) + sentinelTopic := topic.NewTopic(topic.TopicKindGroupMessagesV1, []byte("seam-sentinel")). + Bytes() + + testutils.InsertGatewayEnvelopes(t, suite.DB, []queries.InsertGatewayEnvelopeV3Params{ + envRow(t, payerID, 100, 1, topicA), + envRow(t, payerID, 100, 2, topicA), + envRow(t, payerID, 100, 3, topicA), + }) + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + require.NoError(t, suite.MessageService.AwaitCursor(ctx, db.VectorClock{100: 3})) + + stream := suite.ClientQuery.Subscribe(t.Context()) + reader := newBidiReader(stream) + + // Sentinel first: subscribed, live, and demonstrably delivering before the wave starts. + require.NoError(t, stream.Send(subMutate( + 1, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{addSub(sentinelTopic, nil)}, + nil, + ))) + require.Eventually(t, func() bool { + return hasMutateID(subCatchupCompletes(mustFrames(reader)), 1) + }, 10*time.Second, 20*time.Millisecond) + testutils.InsertGatewayEnvelopes(t, suite.DB, []queries.InsertGatewayEnvelopeV3Params{ + envRow(t, payerID, 200, 1, sentinelTopic), + }) + require.Eventually(t, func() bool { + return hasEnvKey(subEnvelopeKeysTagged(t, mustFrames(reader), 0), 200, 1) + }, 10*time.Second, 20*time.Millisecond, "sentinel must be live before the wave starts") + + // Start the wave and immediately publish into it, racing the replay — and keep the + // sentinel publishing during the wave. + require.NoError(t, stream.Send(subMutate( + 9, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{addSub(topicA, nil)}, + nil, + ))) + testutils.InsertGatewayEnvelopes(t, suite.DB, []queries.InsertGatewayEnvelopeV3Params{ + envRow(t, payerID, 100, 4, topicA), + envRow(t, payerID, 100, 5, topicA), + envRow(t, payerID, 200, 2, sentinelTopic), + }) + + require.Eventually(t, func() bool { + frames := mustFrames(reader) + keys := subEnvelopeKeys(t, frames) + return hasMutateID(subCatchupCompletes(frames), 9) && + hasEnvKey(keys, 100, 4) && hasEnvKey(keys, 100, 5) + }, 10*time.Second, 20*time.Millisecond, "wave complete + racers delivered") + + // After the wave: the sentinel's live lane must still be flowing. + testutils.InsertGatewayEnvelopes(t, suite.DB, []queries.InsertGatewayEnvelopeV3Params{ + envRow(t, payerID, 200, 3, sentinelTopic), + }) + require.Eventually(t, func() bool { + live := subEnvelopeKeysTagged(t, mustFrames(reader), 0) + return hasEnvKey(live, 200, 2) && hasEnvKey(live, 200, 3) + }, 10*time.Second, 20*time.Millisecond, + "sentinel tag-0 delivery must keep flowing across the wave") + + frames, err := reader.snapshot() + require.NoError(t, err) + requireExactlyOnce(t, subEnvelopeKeys(t, frames), "wave + live") + requirePerOriginatorAscending(t, subEnvelopeKeysTagged(t, frames, 9), "wave 9 replay") + requirePerOriginatorAscending(t, subEnvelopeKeysTagged(t, frames, 0), "live lane") + + // The sentinel's envelopes (originator 200) travel the live lane only — never the wave's. + for _, k := range subEnvelopeKeysTagged(t, frames, 9) { + require.Equal(t, uint64(100), k[0], "wave 9 must not capture the sentinel's envelopes") + } + + // The seam, scoped to the wave's topic (originator 100 publishes only to topicA here): no + // live frame for it may precede CatchupComplete(9), and every wave-tagged frame must + // precede it. Sentinel (originator 200) tag-0 frames may land on either side. + ccIdx := -1 + for i, f := range frames { + if cc := f.GetV1().GetCatchupComplete(); cc != nil && cc.GetMutateId() == 9 { + ccIdx = i + } + } + require.GreaterOrEqual(t, ccIdx, 0) + for i, f := range frames { + env := f.GetV1().GetEnvelopes() + if env == nil || len(env.GetEnvelopes()) == 0 { + continue + } + switch env.GetMutateId() { + case 0: + for _, e := range env.GetEnvelopes() { + u := envelopeTestUtils.UnmarshalUnsignedOriginatorEnvelope( + t, + e.GetUnsignedOriginatorEnvelope(), + ) + if u.GetOriginatorNodeId() == 100 { + require.Greater(t, i, ccIdx, + "a live frame for the wave's topic must follow its CatchupComplete") + } + } + case 9: + require.Less(t, i, ccIdx, + "a wave replay frame must precede its CatchupComplete") + default: + t.Fatalf("frame %d carries unexpected tag %d", i, env.GetMutateId()) + } + } +} + +// TestSubscribe_ResetMidWaveReplaysUnderNewTag covers remove+re-add (reset) racing an +// envelope-bearing first wave: the reset topic is owned by the newer wave, whose replay is +// stamped with the new mutate_id; the stale wave's remaining pages are dropped, so nothing +// is delivered twice and only the owning wave announces the topic. +func TestSubscribe_ResetMidWaveReplaysUnderNewTag(t *testing.T) { + suite := setupTest(t) + payerID := db.NullInt32(testutils.CreatePayer(t, suite.DB)) + resetTopic := topic.NewTopic(topic.TopicKindGroupMessagesV1, []byte("reset-mid-wave")). + Bytes() + + // More than one wave-scan page (topicPageLimit rows) so the reset can land between the + // stale wave's pages. + const total = 600 + rows := make([]queries.InsertGatewayEnvelopeV3Params, 0, total) + for i := int64(1); i <= total; i++ { + rows = append(rows, envRow(t, payerID, 100, i, resetTopic)) + } + testutils.InsertGatewayEnvelopes(t, suite.DB, rows) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + require.NoError(t, suite.MessageService.AwaitCursor(ctx, db.VectorClock{100: total})) + + stream := suite.ClientQuery.Subscribe(t.Context()) + reader := newBidiReader(stream) + + // Wave 1 starts the replay; the reset (remove + re-add from an empty cursor) is sent + // immediately, racing wave 1's pages. + require.NoError(t, stream.Send(subMutate( + 1, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{addSub(resetTopic, nil)}, + nil, + ))) + require.NoError(t, stream.Send(subMutate( + 2, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{addSub(resetTopic, nil)}, + [][]byte{resetTopic}, + ))) + + require.Eventually(t, func() bool { + cc := subCatchupCompletes(mustFrames(reader)) + return hasMutateID(cc, 1) && hasMutateID(cc, 2) + }, 20*time.Second, 20*time.Millisecond, "both waves must complete") + + frames, err := reader.snapshot() + require.NoError(t, err) + wave1 := subEnvelopeKeysTagged(t, frames, 1) + wave2 := subEnvelopeKeysTagged(t, frames, 2) + requireExactlyOnce(t, wave1, "stale wave replay") + requireExactlyOnce(t, wave2, "reset wave replay") + requirePerOriginatorAscending(t, wave1, "stale wave replay") + requirePerOriginatorAscending(t, wave2, "reset wave replay") + + // Between them the two waves cover every seeded envelope exactly once: the reset wave owns + // everything the stale wave had not yet delivered when the reset applied. (Do not assert + // how the split falls — the reset races the stale wave's pages.) + seen := make(map[[2]uint64]struct{}, total) + for _, k := range wave1 { + seen[k] = struct{}{} + } + for _, k := range wave2 { + _, both := seen[k] + require.Falsef(t, both, "envelope %v delivered under both tag 1 and tag 2", k) + seen[k] = struct{}{} + } + require.Len(t, seen, total, "the two waves together must cover every envelope exactly once") + for i := int64(1); i <= total; i++ { + require.Contains(t, seen, [2]uint64{100, uint64(i)}) + } + require.True(t, hasEnvKey(wave2, 100, total), + "the reset wave must deliver at least the tail of the history") + + // Only the owning (reset) wave announces the topic, and nothing rides the live tag: the + // history predates the subscription, so the live lane has nothing to say before CC(2). + announced := 0 + for _, tl := range subTopicsLive(frames) { + if bytes.Equal(tl, resetTopic) { + announced++ + } + } + require.Equal(t, 1, announced, "exactly one TopicsLive may announce the reset topic") + require.Empty(t, subEnvelopeKeysTagged(t, frames, 0), "no envelope may ride the live tag") +} + +// TestSubscribe_WaveScanPaginatesPastPageLimit drives the wave's merged keyset scan across a +// page boundary that lands mid-originator: the resume must be strictly-after the last row (a +// >= row-value comparison would re-deliver the boundary row) without skipping the next +// originator's rows. +func TestSubscribe_WaveScanPaginatesPastPageLimit(t *testing.T) { + suite := setupTest(t) + payerID := db.NullInt32(testutils.CreatePayer(t, suite.DB)) + pageT1 := topic.NewTopic(topic.TopicKindGroupMessagesV1, []byte("wave-page-t1")).Bytes() + pageT2 := topic.NewTopic(topic.TopicKindGroupMessagesV1, []byte("wave-page-t2")).Bytes() + + // topicPageLimit (500) + 1 rows for originator 100, interleaved across both topics, put + // the page boundary inside originator 100's run; originator 200's rows follow on page 2. + const heavy = 501 + const light = 5 + rows := make([]queries.InsertGatewayEnvelopeV3Params, 0, heavy+light) + for i := int64(1); i <= heavy; i++ { + tp := pageT1 + if i%2 == 0 { + tp = pageT2 + } + rows = append(rows, envRow(t, payerID, 100, i, tp)) + } + for i := int64(1); i <= light; i++ { + rows = append(rows, envRow(t, payerID, 200, i, pageT1)) + } + testutils.InsertGatewayEnvelopes(t, suite.DB, rows) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + require.NoError( + t, + suite.MessageService.AwaitCursor(ctx, db.VectorClock{100: heavy, 200: light}), + ) + + stream := suite.ClientQuery.Subscribe(t.Context()) + reader := newBidiReader(stream) + + require.NoError(t, stream.Send(subMutate( + 5, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{ + addSub(pageT1, nil), + addSub(pageT2, nil), + }, + nil, + ))) + require.Eventually(t, func() bool { + return hasMutateID(subCatchupCompletes(mustFrames(reader)), 5) + }, 20*time.Second, 20*time.Millisecond) + + frames, err := reader.snapshot() + require.NoError(t, err) + replay := subEnvelopeKeysTagged(t, frames, 5) + require.Len(t, replay, heavy+light, "every seeded envelope arrives tagged with the wave") + requireExactlyOnce(t, replay, "wave replay across pages") + requirePerOriginatorAscending(t, replay, "wave replay across pages") + + waveFrames := 0 + for _, f := range frames { + if env := f.GetV1().GetEnvelopes(); env != nil && env.GetMutateId() == 5 { + waveFrames++ + } + } + require.GreaterOrEqual(t, waveFrames, 2, "the replay must span multiple scan pages") +} + +// TestSubscribe_CursorNamedUnknownOriginatorReplayed covers a client cursor naming an +// originator the TTL-cached originator list has not seen (its rows exist in +// gateway_envelopes_meta, but the cached list is stale): the wave must still pin a ceiling +// for it and replay its rows, rather than silently dropping them from the catch-up (the +// legacy per-topic path replayed them — see TestSubscribeTopics_AcceptsUnknownOriginatorInCursor). +func TestSubscribe_CursorNamedUnknownOriginatorReplayed(t *testing.T) { + suite := setupTest(t) + payerID := db.NullInt32(testutils.CreatePayer(t, suite.DB)) + + // topicA history from originator 100 (known everywhere) and originator 300 (not in the + // registry, so the worker never polls it — nothing can arrive via the live lane). + testutils.InsertGatewayEnvelopes(t, suite.DB, []queries.InsertGatewayEnvelopeV3Params{ + envRow(t, payerID, 100, 1, topicA), + envRow(t, payerID, 300, 1, topicA), + envRow(t, payerID, 300, 2, topicA), + envRow(t, payerID, 300, 3, topicA), + }) + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + require.NoError(t, suite.MessageService.AwaitCursor(ctx, db.VectorClock{100: 1})) + + // Fabricate the stale-cache state: drop 300 from gateway_envelopes_latest (the cached + // originator list's source) while its rows stay in gateway_envelopes_meta. This is what a + // TTL-stale CachedOriginatorList sees when an originator's first rows land after the + // cache was filled — deterministic here instead of racing the 100ms test TTL. + _, err := suite.DB.ExecContext(t.Context(), + "DELETE FROM gateway_envelopes_latest WHERE originator_node_id = 300") + require.NoError(t, err) + + stream := suite.ClientQuery.Subscribe(t.Context()) + reader := newBidiReader(stream) + + // The cursor names originator 300 below its newest row: the wave owes (300,2),(300,3) + // even though the originator list has never heard of 300. + require.NoError(t, stream.Send(subMutate( + 1, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{ + addSub(topicA, map[uint32]uint64{300: 1}), + }, + nil, + ))) + + require.Eventually(t, func() bool { + frames := mustFrames(reader) + replay := subEnvelopeKeysTagged(t, frames, 1) + return hasEnvKey(replay, 100, 1) && + hasEnvKey(replay, 300, 2) && hasEnvKey(replay, 300, 3) && + hasMutateID(subCatchupCompletes(frames), 1) + }, 10*time.Second, 20*time.Millisecond, + "the wave must replay the cursor-named originator 300, then CatchupComplete(1)") + + frames, err := reader.snapshot() + require.NoError(t, err) + replay := subEnvelopeKeysTagged(t, frames, 1) + require.False(t, hasEnvKey(replay, 300, 1), "the cursor floor (300,1) must not be re-delivered") + requireExactlyOnce(t, subEnvelopeKeys(t, frames), "unknown-originator replay") + requirePerOriginatorAscending(t, replay, "unknown-originator replay") +} + +// TestSubscribe_DuplicateAddsFirstCursorWins pins handleMutate's add dedup: when one Mutate +// carries two adds for the same topic, the first add's cursor is the floor (the duplicate is +// dropped), so replay starts strictly after it — and the topic is announced once. +func TestSubscribe_DuplicateAddsFirstCursorWins(t *testing.T) { + suite := setupTest(t) + payerID := db.NullInt32(testutils.CreatePayer(t, suite.DB)) + dupTopic := topic.NewTopic(topic.TopicKindGroupMessagesV1, []byte("dup-adds")).Bytes() + + testutils.InsertGatewayEnvelopes(t, suite.DB, []queries.InsertGatewayEnvelopeV3Params{ + envRow(t, payerID, 100, 1, dupTopic), + envRow(t, payerID, 100, 2, dupTopic), + envRow(t, payerID, 100, 3, dupTopic), + envRow(t, payerID, 100, 4, dupTopic), + }) + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + require.NoError(t, suite.MessageService.AwaitCursor(ctx, db.VectorClock{100: 4})) + + stream := suite.ClientQuery.Subscribe(t.Context()) + reader := newBidiReader(stream) + + // First add carries cursor {100: 2}; the duplicate second add carries an empty cursor. + require.NoError(t, stream.Send(subMutate( + 4, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{ + addSub(dupTopic, map[uint32]uint64{100: 2}), + addSub(dupTopic, map[uint32]uint64{}), + }, + nil, + ))) + require.Eventually(t, func() bool { + return hasMutateID(subCatchupCompletes(mustFrames(reader)), 4) + }, 10*time.Second, 20*time.Millisecond) + + frames, err := reader.snapshot() + require.NoError(t, err) + require.Equal(t, [][2]uint64{{100, 3}, {100, 4}}, subEnvelopeKeysTagged(t, frames, 4), + "replay must start after the FIRST add's cursor, exactly once") + require.Empty(t, subEnvelopeKeysTagged(t, frames, 0)) + + announced := 0 + for _, tl := range subTopicsLive(frames) { + if bytes.Equal(tl, dupTopic) { + announced++ + } + } + require.Equal(t, 1, announced, "the deduped topic is announced exactly once") +} + +// TestSubscribe_AddsRequireNonzeroMutateId pins the request-side rule the tag depends on: +// a Mutate with adds and mutate_id 0 (the live tag) fails the stream with InvalidArgument. +func TestSubscribe_AddsRequireNonzeroMutateId(t *testing.T) { + suite := setupTest(t) + + stream := suite.ClientQuery.Subscribe(t.Context()) + reader := newBidiReader(stream) + + require.NoError(t, stream.Send(subMutate( + 0, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{addSub(topicA, nil)}, + nil, + ))) + + require.Eventually(t, func() bool { + _, err := reader.snapshot() + return err != nil + }, 10*time.Second, 20*time.Millisecond, "adds with mutate_id 0 must fail the stream") + + _, err := reader.snapshot() + require.Equal(t, connect.CodeInvalidArgument, connect.CodeOf(err)) +} + +// TestSubscribe_EmptyMutateAcked pins the ack rule for the degenerate Mutate shape: a Mutate +// with no adds and no removes is still confirmed with exactly one CatchupComplete echoing its +// mutate_id, and the stream stays healthy afterwards (a subsequent subscribe works end to end). +func TestSubscribe_EmptyMutateAcked(t *testing.T) { + suite := setupTest(t) + + stream := suite.ClientQuery.Subscribe(t.Context()) + reader := newBidiReader(stream) + + // Empty Mutate: no adds, no removes. + require.NoError(t, stream.Send(subMutate(3, false, nil, nil))) + require.Eventually(t, func() bool { + return hasMutateID(subCatchupCompletes(mustFrames(reader)), 3) + }, 10*time.Second, 20*time.Millisecond, "an empty Mutate must be acked promptly") + + // The stream is still usable: a subsequent subscribe completes its own wave. + require.NoError(t, stream.Send(subMutate( + 4, false, + []*message_api.SubscribeRequest_V1_Mutate_Subscription{addSub(topicA, nil)}, + nil, + ))) + require.Eventually(t, func() bool { + return hasMutateID(subCatchupCompletes(mustFrames(reader)), 4) + }, 10*time.Second, 20*time.Millisecond, "the stream must stay healthy after an empty Mutate") + + frames, err := reader.snapshot() + require.NoError(t, err) + acks := 0 + for _, id := range subCatchupCompletes(frames) { + if id == 3 { + acks++ + } + } + require.Equal(t, 1, acks, "exactly one CatchupComplete must echo the empty Mutate's id") +} diff --git a/pkg/api/message/subscribe_topics.go b/pkg/api/message/subscribe_topics.go index e78aec233..85bec21c2 100644 --- a/pkg/api/message/subscribe_topics.go +++ b/pkg/api/message/subscribe_topics.go @@ -357,25 +357,6 @@ func advanceTopicCursors( return result } -// advanceCursorsFromRows advances per-topic pagination cursors from the RAW query rows (not just the -// rows that successfully unmarshal). A catch-up paginator that advanced only from unmarshaled -// envelopes would never move past a row whose envelope bytes fail to parse: since pagination -// terminates on raw row count, a single bad row in an otherwise-full page would re-fetch that page -// forever and the wave would never complete. Advancing from raw rows guarantees forward progress. -func advanceCursorsFromRows(cursors db.TopicCursors, rows []queries.GatewayEnvelopesView) { - for i := range rows { - vc, ok := cursors[string(rows[i].Topic)] - if !ok { - continue - } - origID := uint32(rows[i].OriginatorNodeID) - seqID := uint64(rows[i].OriginatorSequenceID) - if cur, seen := vc[origID]; !seen || cur < seqID { - vc[origID] = seqID - } - } -} - func newSubscriptionStatusMessage( status message_api.SubscribeTopicsResponse_SubscriptionStatus, ) *message_api.SubscribeTopicsResponse { diff --git a/pkg/db/queries/db.go b/pkg/db/queries/db.go index 1774554aa..6a89dd18d 100644 --- a/pkg/db/queries/db.go +++ b/pkg/db/queries/db.go @@ -216,12 +216,18 @@ func Prepare(ctx context.Context, db DBTX) (*Queries, error) { if q.selectGatewayEnvelopesUnfilteredStmt, err = db.PrepareContext(ctx, selectGatewayEnvelopesUnfiltered); err != nil { return nil, fmt.Errorf("error preparing query SelectGatewayEnvelopesUnfiltered: %w", err) } + if q.selectGatewayEnvelopesWaveScanStmt, err = db.PrepareContext(ctx, selectGatewayEnvelopesWaveScan); err != nil { + return nil, fmt.Errorf("error preparing query SelectGatewayEnvelopesWaveScan: %w", err) + } if q.selectNewestFromTopicsStmt, err = db.PrepareContext(ctx, selectNewestFromTopics); err != nil { return nil, fmt.Errorf("error preparing query SelectNewestFromTopics: %w", err) } if q.selectNodeInfoStmt, err = db.PrepareContext(ctx, selectNodeInfo); err != nil { return nil, fmt.Errorf("error preparing query SelectNodeInfo: %w", err) } + if q.selectOriginatorCeilingsStmt, err = db.PrepareContext(ctx, selectOriginatorCeilings); err != nil { + return nil, fmt.Errorf("error preparing query SelectOriginatorCeilings: %w", err) + } if q.selectOriginatorNodeIDsStmt, err = db.PrepareContext(ctx, selectOriginatorNodeIDs); err != nil { return nil, fmt.Errorf("error preparing query SelectOriginatorNodeIDs: %w", err) } @@ -583,6 +589,11 @@ func (q *Queries) Close() error { err = fmt.Errorf("error closing selectGatewayEnvelopesUnfilteredStmt: %w", cerr) } } + if q.selectGatewayEnvelopesWaveScanStmt != nil { + if cerr := q.selectGatewayEnvelopesWaveScanStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing selectGatewayEnvelopesWaveScanStmt: %w", cerr) + } + } if q.selectNewestFromTopicsStmt != nil { if cerr := q.selectNewestFromTopicsStmt.Close(); cerr != nil { err = fmt.Errorf("error closing selectNewestFromTopicsStmt: %w", cerr) @@ -593,6 +604,11 @@ func (q *Queries) Close() error { err = fmt.Errorf("error closing selectNodeInfoStmt: %w", cerr) } } + if q.selectOriginatorCeilingsStmt != nil { + if cerr := q.selectOriginatorCeilingsStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing selectOriginatorCeilingsStmt: %w", cerr) + } + } if q.selectOriginatorNodeIDsStmt != nil { if cerr := q.selectOriginatorNodeIDsStmt.Close(); cerr != nil { err = fmt.Errorf("error closing selectOriginatorNodeIDsStmt: %w", cerr) @@ -756,8 +772,10 @@ type Queries struct { selectGatewayEnvelopesBySingleOriginatorStmt *sql.Stmt selectGatewayEnvelopesByTopicsStmt *sql.Stmt selectGatewayEnvelopesUnfilteredStmt *sql.Stmt + selectGatewayEnvelopesWaveScanStmt *sql.Stmt selectNewestFromTopicsStmt *sql.Stmt selectNodeInfoStmt *sql.Stmt + selectOriginatorCeilingsStmt *sql.Stmt selectOriginatorNodeIDsStmt *sql.Stmt selectStagedOriginatorEnvelopesStmt *sql.Stmt selectVectorClockStmt *sql.Stmt @@ -840,8 +858,10 @@ func (q *Queries) WithTx(tx *sql.Tx) *Queries { selectGatewayEnvelopesBySingleOriginatorStmt: q.selectGatewayEnvelopesBySingleOriginatorStmt, selectGatewayEnvelopesByTopicsStmt: q.selectGatewayEnvelopesByTopicsStmt, selectGatewayEnvelopesUnfilteredStmt: q.selectGatewayEnvelopesUnfilteredStmt, + selectGatewayEnvelopesWaveScanStmt: q.selectGatewayEnvelopesWaveScanStmt, selectNewestFromTopicsStmt: q.selectNewestFromTopicsStmt, selectNodeInfoStmt: q.selectNodeInfoStmt, + selectOriginatorCeilingsStmt: q.selectOriginatorCeilingsStmt, selectOriginatorNodeIDsStmt: q.selectOriginatorNodeIDsStmt, selectStagedOriginatorEnvelopesStmt: q.selectStagedOriginatorEnvelopesStmt, selectVectorClockStmt: q.selectVectorClockStmt, diff --git a/pkg/db/queries/envelopes_v2.sql.go b/pkg/db/queries/envelopes_v2.sql.go index b4ff7cfbd..0b3d1883f 100644 --- a/pkg/db/queries/envelopes_v2.sql.go +++ b/pkg/db/queries/envelopes_v2.sql.go @@ -646,6 +646,106 @@ func (q *Queries) SelectGatewayEnvelopesUnfiltered(ctx context.Context, arg Sele return items, nil } +const selectGatewayEnvelopesWaveScan = `-- name: SelectGatewayEnvelopesWaveScan :many +WITH cursor_entries AS ( + SELECT t.topic, n.node_id, s.seq_id + FROM unnest($4::BYTEA[]) WITH ORDINALITY AS t(topic, ord) + JOIN unnest($5::INT[]) WITH ORDINALITY AS n(node_id, ord) USING (ord) + JOIN unnest($6::BIGINT[]) WITH ORDINALITY AS s(seq_id, ord) USING (ord) +), +ceilings AS ( + SELECT x.node_id, y.seq_id + FROM unnest($7::INT[]) WITH ORDINALITY AS x(node_id, ord) + JOIN unnest($8::BIGINT[]) WITH ORDINALITY AS y(seq_id, ord) USING (ord) +) +SELECT m.originator_node_id, + m.originator_sequence_id, + m.gateway_time, + m.topic, + b.originator_envelope +FROM gateway_envelopes_meta AS m +JOIN cursor_entries AS ce + ON m.topic = ce.topic AND m.originator_node_id = ce.node_id +JOIN ceilings AS cl + ON cl.node_id = m.originator_node_id +JOIN gateway_envelopes_blob AS b + ON b.originator_node_id = m.originator_node_id + AND b.originator_sequence_id = m.originator_sequence_id +WHERE m.originator_sequence_id > ce.seq_id + AND m.originator_sequence_id <= cl.seq_id + AND (m.originator_node_id, m.originator_sequence_id) > ($1::INT, $2::BIGINT) +ORDER BY m.originator_node_id, m.originator_sequence_id +LIMIT $3::INT +` + +type SelectGatewayEnvelopesWaveScanParams struct { + ScanNodeID int32 + ScanSequenceID int64 + RowLimit int32 + CursorTopics [][]byte + CursorNodeIds []int32 + CursorSequenceIds []int64 + CeilingNodeIds []int32 + CeilingSequenceIds []int64 +} + +type SelectGatewayEnvelopesWaveScanRow struct { + OriginatorNodeID int32 + OriginatorSequenceID int64 + GatewayTime time.Time + Topic []byte + OriginatorEnvelope []byte +} + +// One page of a Subscribe catch-up wave's replay: the wave's per-(topic, originator) +// cursor floors merged into a single (originator, sequence) keyset scan, bounded per +// originator by the ceiling pinned at wave start — so the wave's replay is delivered +// in total cursor order per originator across ALL of its topics, not per-topic +// bursts, and the scan terminates under sustained publishing (XIP-83 server +// requirement 4). The page starts strictly after (scan_node_id, scan_sequence_id) +// (the previous page's last row); a page shorter than row_limit means the wave is +// fully replayed up to its ceilings. +// +// Cursor keys MUST be unique per (topic, originator): unnest preserves duplicates +// and the join would return a repeated pair's rows more than once. +func (q *Queries) SelectGatewayEnvelopesWaveScan(ctx context.Context, arg SelectGatewayEnvelopesWaveScanParams) ([]SelectGatewayEnvelopesWaveScanRow, error) { + rows, err := q.query(ctx, q.selectGatewayEnvelopesWaveScanStmt, selectGatewayEnvelopesWaveScan, + arg.ScanNodeID, + arg.ScanSequenceID, + arg.RowLimit, + pq.Array(arg.CursorTopics), + pq.Array(arg.CursorNodeIds), + pq.Array(arg.CursorSequenceIds), + pq.Array(arg.CeilingNodeIds), + pq.Array(arg.CeilingSequenceIds), + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []SelectGatewayEnvelopesWaveScanRow + for rows.Next() { + var i SelectGatewayEnvelopesWaveScanRow + if err := rows.Scan( + &i.OriginatorNodeID, + &i.OriginatorSequenceID, + &i.GatewayTime, + &i.Topic, + &i.OriginatorEnvelope, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const selectNewestFromTopics = `-- name: SelectNewestFromTopics :many WITH latest AS (SELECT DISTINCT ON (m.topic) m.originator_node_id, m.originator_sequence_id, @@ -703,3 +803,42 @@ func (q *Queries) SelectNewestFromTopics(ctx context.Context, topics [][]byte) ( } return items, nil } + +const selectOriginatorCeilings = `-- name: SelectOriginatorCeilings :many +SELECT o.node_id::INT AS originator_node_id, + COALESCE((SELECT max(m.originator_sequence_id) + FROM gateway_envelopes_meta m + WHERE m.originator_node_id = o.node_id), 0)::BIGINT AS max_sequence_id +FROM unnest($1::INT[]) AS o(node_id) +` + +type SelectOriginatorCeilingsRow struct { + OriginatorNodeID int32 + MaxSequenceID int64 +} + +// Newest sequence id per originator: the per-originator replay ceiling a Subscribe +// catch-up wave pins at wave start (XIP-83 server requirement 4). One backward probe +// of the (originator_node_id, originator_sequence_id) primary key per originator. +func (q *Queries) SelectOriginatorCeilings(ctx context.Context, nodeIds []int32) ([]SelectOriginatorCeilingsRow, error) { + rows, err := q.query(ctx, q.selectOriginatorCeilingsStmt, selectOriginatorCeilings, pq.Array(nodeIds)) + if err != nil { + return nil, err + } + defer rows.Close() + var items []SelectOriginatorCeilingsRow + for rows.Next() { + var i SelectOriginatorCeilingsRow + if err := rows.Scan(&i.OriginatorNodeID, &i.MaxSequenceID); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/pkg/db/sqlc/envelopes_v2.sql b/pkg/db/sqlc/envelopes_v2.sql index 2617e2645..fbb18892c 100644 --- a/pkg/db/sqlc/envelopes_v2.sql +++ b/pkg/db/sqlc/envelopes_v2.sql @@ -261,6 +261,58 @@ CROSS JOIN LATERAL ( ) AS bl ORDER BY bl.originator_node_id, bl.originator_sequence_id; +-- name: SelectOriginatorCeilings :many +-- Newest sequence id per originator: the per-originator replay ceiling a Subscribe +-- catch-up wave pins at wave start (XIP-83 server requirement 4). One backward probe +-- of the (originator_node_id, originator_sequence_id) primary key per originator. +SELECT o.node_id::INT AS originator_node_id, + COALESCE((SELECT max(m.originator_sequence_id) + FROM gateway_envelopes_meta m + WHERE m.originator_node_id = o.node_id), 0)::BIGINT AS max_sequence_id +FROM unnest(@node_ids::INT[]) AS o(node_id); + +-- name: SelectGatewayEnvelopesWaveScan :many +-- One page of a Subscribe catch-up wave's replay: the wave's per-(topic, originator) +-- cursor floors merged into a single (originator, sequence) keyset scan, bounded per +-- originator by the ceiling pinned at wave start — so the wave's replay is delivered +-- in total cursor order per originator across ALL of its topics, not per-topic +-- bursts, and the scan terminates under sustained publishing (XIP-83 server +-- requirement 4). The page starts strictly after (scan_node_id, scan_sequence_id) +-- (the previous page's last row); a page shorter than row_limit means the wave is +-- fully replayed up to its ceilings. +-- +-- Cursor keys MUST be unique per (topic, originator): unnest preserves duplicates +-- and the join would return a repeated pair's rows more than once. +WITH cursor_entries AS ( + SELECT t.topic, n.node_id, s.seq_id + FROM unnest(@cursor_topics::BYTEA[]) WITH ORDINALITY AS t(topic, ord) + JOIN unnest(@cursor_node_ids::INT[]) WITH ORDINALITY AS n(node_id, ord) USING (ord) + JOIN unnest(@cursor_sequence_ids::BIGINT[]) WITH ORDINALITY AS s(seq_id, ord) USING (ord) +), +ceilings AS ( + SELECT x.node_id, y.seq_id + FROM unnest(@ceiling_node_ids::INT[]) WITH ORDINALITY AS x(node_id, ord) + JOIN unnest(@ceiling_sequence_ids::BIGINT[]) WITH ORDINALITY AS y(seq_id, ord) USING (ord) +) +SELECT m.originator_node_id, + m.originator_sequence_id, + m.gateway_time, + m.topic, + b.originator_envelope +FROM gateway_envelopes_meta AS m +JOIN cursor_entries AS ce + ON m.topic = ce.topic AND m.originator_node_id = ce.node_id +JOIN ceilings AS cl + ON cl.node_id = m.originator_node_id +JOIN gateway_envelopes_blob AS b + ON b.originator_node_id = m.originator_node_id + AND b.originator_sequence_id = m.originator_sequence_id +WHERE m.originator_sequence_id > ce.seq_id + AND m.originator_sequence_id <= cl.seq_id + AND (m.originator_node_id, m.originator_sequence_id) > (@scan_node_id::INT, @scan_sequence_id::BIGINT) +ORDER BY m.originator_node_id, m.originator_sequence_id +LIMIT @row_limit::INT; + -- name: InsertGatewayEnvelopeBatchV2 :one -- Pre-rename batch insert. Calls the v2 stored function which still -- references the legacy gateway_envelope_blobs table. No production code diff --git a/pkg/db/types.go b/pkg/db/types.go index 0ebd84b09..b7b1b8a83 100644 --- a/pkg/db/types.go +++ b/pkg/db/types.go @@ -163,6 +163,63 @@ func TransformRowsByPerTopicCursors( return result } +// SetWaveScanCursors flattens TopicCursors into the parallel floor arrays required by +// SelectGatewayEnvelopesWaveScan. Each (topic, nodeID, seqID) triple produces one entry +// in the three arrays. +func SetWaveScanCursors( + q *queries.SelectGatewayEnvelopesWaveScanParams, + tc TopicCursors, +) { + total := 0 + for _, vc := range tc { + total += len(vc) + } + + q.CursorTopics = make([][]byte, 0, total) + q.CursorNodeIds = make([]int32, 0, total) + q.CursorSequenceIds = make([]int64, 0, total) + + for topicKey, vc := range tc { + topicBytes := []byte(topicKey) + for nodeID, seqID := range vc { + if nodeID > math.MaxInt32 || seqID > uint64(math.MaxInt64) { + continue + } + q.CursorTopics = append(q.CursorTopics, topicBytes) + q.CursorNodeIds = append(q.CursorNodeIds, int32(nodeID)) + q.CursorSequenceIds = append(q.CursorSequenceIds, int64(seqID)) + } + } +} + +// SetWaveScanCeilings flattens the per-originator ceiling vector into the parallel +// arrays required by SelectGatewayEnvelopesWaveScan. +func SetWaveScanCeilings( + q *queries.SelectGatewayEnvelopesWaveScanParams, + ceilings VectorClock, +) { + q.CeilingNodeIds = make([]int32, 0, len(ceilings)) + q.CeilingSequenceIds = make([]int64, 0, len(ceilings)) + for nodeID, seqID := range ceilings { + if nodeID > math.MaxInt32 || seqID > uint64(math.MaxInt64) { + continue + } + q.CeilingNodeIds = append(q.CeilingNodeIds, int32(nodeID)) + q.CeilingSequenceIds = append(q.CeilingSequenceIds, int64(seqID)) + } +} + +// TransformRowsWaveScan converts wave-scan rows to the common GatewayEnvelopesView type. +func TransformRowsWaveScan( + rows []queries.SelectGatewayEnvelopesWaveScanRow, +) []queries.GatewayEnvelopesView { + result := make([]queries.GatewayEnvelopesView, len(rows)) + for i, row := range rows { + result[i] = queries.GatewayEnvelopesView(row) + } + return result +} + // CalculateRowsPerEntry computes the per-(topic, originator) sub-limit // for the per-topic cursor query. Returns at least 10 to avoid starving // low-volume originators. diff --git a/pkg/proto/openapi/xmtpv4/message_api/message_api.swagger.json b/pkg/proto/openapi/xmtpv4/message_api/message_api.swagger.json index 70c10dc55..6643afcbf 100644 --- a/pkg/proto/openapi/xmtpv4/message_api/message_api.swagger.json +++ b/pkg/proto/openapi/xmtpv4/message_api/message_api.swagger.json @@ -17,6 +17,18 @@ ], "paths": {}, "definitions": { + "GetInboxIdsRequestRequest": { + "type": "object", + "properties": { + "identifier": { + "type": "string" + }, + "identifierKind": { + "$ref": "#/definitions/associationsIdentifierKind" + } + }, + "title": "A single request for a given address" + }, "SubscribeOriginatorsRequestOriginatorFilter": { "type": "object", "properties": { @@ -28,7 +40,7 @@ } }, "lastSeen": { - "$ref": "#/definitions/xmtpv4envelopesCursor" + "$ref": "#/definitions/envelopesCursor" } } }, @@ -40,7 +52,7 @@ "format": "byte" }, "lastSeen": { - "$ref": "#/definitions/xmtpv4envelopesCursor" + "$ref": "#/definitions/envelopesCursor" } } }, @@ -94,6 +106,19 @@ }, "title": "An alternative to a signature for blockchain payloads" }, + "envelopesCursor": { + "type": "object", + "properties": { + "nodeIdToSequenceId": { + "type": "object", + "additionalProperties": { + "type": "string", + "format": "uint64" + } + } + }, + "description": "The last seen entry per originator. Originators that have not been seen are omitted." + }, "envelopesOriginatorEnvelope": { "type": "object", "properties": { @@ -153,11 +178,39 @@ "title": "Node queries" }, "lastSeen": { - "$ref": "#/definitions/xmtpv4envelopesCursor" + "$ref": "#/definitions/envelopesCursor" } }, "title": "Query for envelopes, shared by query and subscribe endpoints\nEither topics or originator_node_ids may be set, but not both" }, + "message_apiGetInboxIdsResponse": { + "type": "object", + "properties": { + "responses": { + "type": "array", + "items": { + "type": "object", + "$ref": "#/definitions/message_apiGetInboxIdsResponseResponse" + } + } + }, + "title": "Response with the XIDs for the requested addresses" + }, + "message_apiGetInboxIdsResponseResponse": { + "type": "object", + "properties": { + "identifier": { + "type": "string" + }, + "inboxId": { + "type": "string" + }, + "identifierKind": { + "$ref": "#/definitions/associationsIdentifierKind" + } + }, + "title": "A single response for a given address" + }, "message_apiGetNewestEnvelopeResponse": { "type": "object", "properties": { @@ -290,59 +343,6 @@ } } } - }, - "xmtpv4envelopesCursor": { - "type": "object", - "properties": { - "nodeIdToSequenceId": { - "type": "object", - "additionalProperties": { - "type": "string", - "format": "uint64" - } - } - }, - "description": "The last seen entry per originator. Originators that have not been seen are omitted." - }, - "xmtpv4message_apiGetInboxIdsRequestRequest": { - "type": "object", - "properties": { - "identifier": { - "type": "string" - }, - "identifierKind": { - "$ref": "#/definitions/associationsIdentifierKind" - } - }, - "title": "A single request for a given address" - }, - "xmtpv4message_apiGetInboxIdsResponse": { - "type": "object", - "properties": { - "responses": { - "type": "array", - "items": { - "type": "object", - "$ref": "#/definitions/xmtpv4message_apiGetInboxIdsResponseResponse" - } - } - }, - "title": "Response with the XIDs for the requested addresses" - }, - "xmtpv4message_apiGetInboxIdsResponseResponse": { - "type": "object", - "properties": { - "identifier": { - "type": "string" - }, - "inboxId": { - "type": "string" - }, - "identifierKind": { - "$ref": "#/definitions/associationsIdentifierKind" - } - }, - "title": "A single response for a given address" } } } diff --git a/pkg/proto/xmtpv4/message_api/message_api.pb.go b/pkg/proto/xmtpv4/message_api/message_api.pb.go index 15fcf8e5d..957c29a9f 100644 --- a/pkg/proto/xmtpv4/message_api/message_api.pb.go +++ b/pkg/proto/xmtpv4/message_api/message_api.pb.go @@ -1395,17 +1395,23 @@ func (*SubscribeRequest_V1_Pong) isSubscribeRequest_V1_Request() {} type SubscribeRequest_V1_Mutate struct { state protoimpl.MessageState `protogen:"open.v1"` Adds []*SubscribeRequest_V1_Mutate_Subscription `protobuf:"bytes,1,rep,name=adds,proto3" json:"adds,omitempty"` // begin delivering these topics - Removes [][]byte `protobuf:"bytes,2,rep,name=removes,proto3" json:"removes,omitempty"` // 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. + Removes [][]byte `protobuf:"bytes,2,rep,name=removes,proto3" json:"removes,omitempty"` // 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. HistoryOnly bool `protobuf:"varint,3,opt,name=history_only,json=historyOnly,proto3" json:"history_only,omitempty"` - // 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. MutateId uint64 `protobuf:"varint,4,opt,name=mutate_id,json=mutateId,proto3" json:"mutate_id,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -1653,11 +1659,11 @@ type SubscribeResponse_V1_Pong struct { } type SubscribeResponse_V1_TopicsLive_ struct { - TopicsLive *SubscribeResponse_V1_TopicsLive `protobuf:"bytes,5,opt,name=topics_live,json=topicsLive,proto3,oneof"` // these topics just crossed from catch-up to live + TopicsLive *SubscribeResponse_V1_TopicsLive `protobuf:"bytes,5,opt,name=topics_live,json=topicsLive,proto3,oneof"` // no more replay for these topics; live begins after CatchupComplete } type SubscribeResponse_V1_CatchupComplete_ struct { - CatchupComplete *SubscribeResponse_V1_CatchupComplete `protobuf:"bytes,6,opt,name=catchup_complete,json=catchupComplete,proto3,oneof"` // a Mutate's adds are fully delivered + CatchupComplete *SubscribeResponse_V1_CatchupComplete `protobuf:"bytes,6,opt,name=catchup_complete,json=catchupComplete,proto3,oneof"` // acks a Mutate; wave completion if it started one } func (*SubscribeResponse_V1_Envelopes_) isSubscribeResponse_V1_Response() {} @@ -1673,10 +1679,17 @@ func (*SubscribeResponse_V1_TopicsLive_) isSubscribeResponse_V1_Response() {} func (*SubscribeResponse_V1_CatchupComplete_) isSubscribeResponse_V1_Response() {} // 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). type SubscribeResponse_V1_Envelopes struct { - state protoimpl.MessageState `protogen:"open.v1"` - Envelopes []*envelopes.OriginatorEnvelope `protobuf:"bytes,1,rep,name=envelopes,proto3" json:"envelopes,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + Envelopes []*envelopes.OriginatorEnvelope `protobuf:"bytes,1,rep,name=envelopes,proto3" json:"envelopes,omitempty"` + // The catch-up wave that produced this frame: the Mutate's mutate_id + // for wave replay, 0 for live tail. + MutateId uint64 `protobuf:"varint,2,opt,name=mutate_id,json=mutateId,proto3" json:"mutate_id,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1718,6 +1731,13 @@ func (x *SubscribeResponse_V1_Envelopes) GetEnvelopes() []*envelopes.OriginatorE return nil } +func (x *SubscribeResponse_V1_Envelopes) GetMutateId() uint64 { + if x != nil { + return x.MutateId + } + return 0 +} + // The first frame on every stream. type SubscribeResponse_V1_Started struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -1777,11 +1797,15 @@ func (x *SubscribeResponse_V1_Started) GetCapabilities() []SubscribeResponse_V1_ return nil } -// 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. type SubscribeResponse_V1_CatchupComplete struct { state protoimpl.MessageState `protogen:"open.v1"` - MutateId uint64 `protobuf:"varint,1,opt,name=mutate_id,json=mutateId,proto3" json:"mutate_id,omitempty"` // echoes the Mutate that started this wave (0 if none given) + MutateId uint64 `protobuf:"varint,1,opt,name=mutate_id,json=mutateId,proto3" json:"mutate_id,omitempty"` // echoes the Mutate; 0 only if a waveless Mutate carried 0 unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1824,14 +1848,15 @@ func (x *SubscribeResponse_V1_CatchupComplete) GetMutateId() uint64 { } // 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. type SubscribeResponse_V1_TopicsLive struct { state protoimpl.MessageState `protogen:"open.v1"` - Topics [][]byte `protobuf:"bytes,1,rep,name=topics,proto3" json:"topics,omitempty"` // kind-prefixed topics now tailing live + Topics [][]byte `protobuf:"bytes,1,rep,name=topics,proto3" json:"topics,omitempty"` // kind-prefixed topics done replaying unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -2174,9 +2199,9 @@ const file_xmtpv4_message_api_message_api_proto_rawDesc = "" + "\x05topic\x18\x01 \x01(\fR\x05topic\x12:\n" + "\tlast_seen\x18\x02 \x01(\v2\x1d.xmtp.xmtpv4.envelopes.CursorR\blastSeenB\t\n" + "\arequestB\t\n" + - "\aversion\"\xc5\a\n" + + "\aversion\"\xe2\a\n" + "\x11SubscribeResponse\x12?\n" + - "\x02v1\x18\x01 \x01(\v2-.xmtp.xmtpv4.message_api.SubscribeResponse.V1H\x00R\x02v1\x1a\xe3\x06\n" + + "\x02v1\x18\x01 \x01(\v2-.xmtp.xmtpv4.message_api.SubscribeResponse.V1H\x00R\x02v1\x1a\x80\a\n" + "\x02V1\x12W\n" + "\tenvelopes\x18\x01 \x01(\v27.xmtp.xmtpv4.message_api.SubscribeResponse.V1.EnvelopesH\x00R\tenvelopes\x12Q\n" + "\astarted\x18\x02 \x01(\v25.xmtp.xmtpv4.message_api.SubscribeResponse.V1.StartedH\x00R\astarted\x123\n" + @@ -2184,9 +2209,10 @@ const file_xmtpv4_message_api_message_api_proto_rawDesc = "" + "\x04pong\x18\x04 \x01(\v2\x1d.xmtp.xmtpv4.message_api.PongH\x00R\x04pong\x12[\n" + "\vtopics_live\x18\x05 \x01(\v28.xmtp.xmtpv4.message_api.SubscribeResponse.V1.TopicsLiveH\x00R\n" + "topicsLive\x12j\n" + - "\x10catchup_complete\x18\x06 \x01(\v2=.xmtp.xmtpv4.message_api.SubscribeResponse.V1.CatchupCompleteH\x00R\x0fcatchupComplete\x1aT\n" + + "\x10catchup_complete\x18\x06 \x01(\v2=.xmtp.xmtpv4.message_api.SubscribeResponse.V1.CatchupCompleteH\x00R\x0fcatchupComplete\x1aq\n" + "\tEnvelopes\x12G\n" + - "\tenvelopes\x18\x01 \x03(\v2).xmtp.xmtpv4.envelopes.OriginatorEnvelopeR\tenvelopes\x1a\x9b\x01\n" + + "\tenvelopes\x18\x01 \x03(\v2).xmtp.xmtpv4.envelopes.OriginatorEnvelopeR\tenvelopes\x12\x1b\n" + + "\tmutate_id\x18\x02 \x01(\x04R\bmutateId\x1a\x9b\x01\n" + "\aStarted\x122\n" + "\x15keepalive_interval_ms\x18\x01 \x01(\rR\x13keepaliveIntervalMs\x12\\\n" + "\fcapabilities\x18\x02 \x03(\x0e28.xmtp.xmtpv4.message_api.SubscribeResponse.V1.CapabilityR\fcapabilities\x1a.\n" +