From 6a42e518cdfaf24990429a5c42176a059a2093bd Mon Sep 17 00:00:00 2001 From: Tyler Hawkes Date: Wed, 8 Jul 2026 13:43:42 -0600 Subject: [PATCH] =?UTF-8?q?feat(mls/api):=20XIP-83=20Subscribe=20=E2=80=94?= =?UTF-8?q?=20tag=20deliveries=20with=20wave=20mutate=5Fid=20and=20enforce?= =?UTF-8?q?=20ordering=20guarantees?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace chunked per-topic catch-up with a per-wave ceiling-pinned merged id-order scan per message kind; gate live delivery for a wave's topics and fold gated arrivals into the wave at completion (tagged, sorted, deduped); stamp every Messages frame with its wave's mutate_id (0 = live); require nonzero mutate_id on adds; ack every Mutate with exactly one CatchupComplete. Welcome wave scans advance from raw rows so unknown message types cannot silently truncate a wave. Co-Authored-By: Claude Fable 5 --- pkg/mls/api/v1/service.go | 7 +- pkg/mls/api/v1/subscribe.go | 586 +++---- pkg/mls/api/v1/subscribe_test.go | 1403 ++++++++++++++++- pkg/mls/store/readStore.go | 110 +- pkg/mls/store/store.go | 32 +- pkg/mls/store/store_test.go | 62 + pkg/proto/mls/api/v1/mls.pb.go | 74 +- pkg/proto/openapi/mls/api/v1/mls.swagger.json | 351 ++++- 8 files changed, 2146 insertions(+), 479 deletions(-) diff --git a/pkg/mls/api/v1/service.go b/pkg/mls/api/v1/service.go index 1ed6402c..5263ab56 100644 --- a/pkg/mls/api/v1/service.go +++ b/pkg/mls/api/v1/service.go @@ -55,11 +55,14 @@ type Service struct { cutoverChecker *migration.CutoverChecker // Subscribe (XIP-83) tunables; overridable in tests. Default to the package consts - // (subscribePingInterval / subscribePongDeadline / maxPendingBytes / maxFrameBytes). + // (subscribePingInterval / subscribePongDeadline / maxPendingBytes / maxFrameBytes / + // catchUpScanPageLimit / maxMutateAdds). pingInterval time.Duration pongDeadline time.Duration maxPendingBytes int maxFrameBytes int + scanPageLimit int32 + maxMutateAdds int } func NewService( @@ -83,6 +86,8 @@ func NewService( pongDeadline: subscribePongDeadline, maxPendingBytes: maxPendingBytes, maxFrameBytes: maxFrameBytes, + scanPageLimit: catchUpScanPageLimit, + maxMutateAdds: maxMutateAdds, } s.ctx, s.ctxCancel = context.WithCancel(context.Background()) if s.dbWorker, err = newDBWorker(s.ctx, log, s.readOnlyStore.Queries(), subDispatcher, DEFAULT_POLL_INTERVAL); err != nil { diff --git a/pkg/mls/api/v1/subscribe.go b/pkg/mls/api/v1/subscribe.go index 46a7151f..8e050e52 100644 --- a/pkg/mls/api/v1/subscribe.go +++ b/pkg/mls/api/v1/subscribe.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "sort" "sync" "time" @@ -21,7 +22,7 @@ import ( const ( // subscribePingInterval is how long a Subscribe stream may be idle before the - // node sends a liveness Ping (XIP-83 server requirement 4; RECOMMENDED <= 30s). + // node sends a liveness Ping (XIP-83 server requirement 6; RECOMMENDED <= 30s). subscribePingInterval = 30 * time.Second // subscribePongDeadline is how long the node waits for the client's Pong before // reaping the stream (XIP-83 RECOMMENDED <= the ping interval). @@ -32,16 +33,20 @@ const ( // writer is not parked by a slow stream.Send. Small: it only smooths Send latency. sendQueueDepth = 8 - // Batched catch-up tuning (XIP-83). chunk = groups per DB round-trip; - // perGroupLimit = rows per group per round; concurrency = chunks in flight. - // The composite (group_id, id) index makes each per-group seek cheap, so the - // binding constraint is payload bytes, not the planner. All tunable. - catchUpChunkSize = 256 - catchUpPerGroupLimit = 50 - catchUpConcurrency = 4 + // Wave-scan catch-up tuning (XIP-83 server requirement 4). A wave's replay is a + // single ascending-id scan merged across all of its topics — not per-topic + // bursts — paged by a global scan cursor and pinned to the ceiling (the newest + // id at wave start) so it terminates under sustained publishing. scanPageLimit + // is rows per DB round-trip. + catchUpScanPageLimit = 512 // catchUpChannelBuffer smooths handoff from the catch-up fetchers to the writer; // when full, fetchers block (backpressure) rather than racing ahead of the client. catchUpChannelBuffer = 64 + // maxMutateAdds bounds a single wave's merged catch-up scan: the scan's floor arrays + // carry one entry per topic and ride every page query, so the raw adds one Mutate may + // carry are capped (pre-dedup — a stateless check). A client with a larger set splits + // it across Mutates, whose waves run concurrently (XIP-83 server requirement 8). + maxMutateAdds = 100_000 // maxPendingBytes caps live messages buffered while topics catch up. Exceeding it // drops the stream (the client reconnects from its cursor) rather than risk OOM. @@ -83,29 +88,34 @@ func buildMLSTopic(kind byte, id []byte) string { return topic.BuildMLSV1GroupTopic(id) } -// catchUpBatch is one unit of fetched catch-up history handed from a fetcher goroutine to -// the single writer. opened lists internal topics that finished catch-up in this batch — -// the writer flushes their buffered live messages, opens their gate, and announces them in -// a TopicsLive frame; openedWire carries the matching kind-prefixed wire topics the -// announcement echoes back to the client. wave identifies the Mutate whose adds this batch -// serves; the writer emits that wave's CatchupComplete once all its topics have opened. A -// non-nil err means the fetch failed and the writer should tear the stream down (the -// client reconnects from its cursor) rather than emit a misleading CatchupComplete or a -// history gap. +// catchUpBatch is one unit of fetched catch-up history handed from a wave's fetcher +// goroutine to the single writer, in scan order. wave identifies the Mutate whose adds +// the batch serves; the writer stamps its frames with that wave's mutate_id (XIP-83 +// server requirement 3). done marks the end of one kind's scan for the wave (the group +// and welcome scans run independently); when the wave's last scan is done the writer +// flushes the wave's gated live buffer, announces TopicsLive, emits CatchupComplete, +// and opens the gates. A non-nil err means the fetch failed and the writer should tear +// the stream down (the client reconnects from its cursor) rather than emit a misleading +// CatchupComplete or a history gap. type catchUpBatch struct { groupMsgs []*mlsv1.GroupMessage welcomeMsgs []*mlsv1.WelcomeMessage - opened []string - openedWire [][]byte wave int + done bool err error } -// waveState tracks one Mutate's in-flight catch-up: how many of its topics have -// yet to open, and the client's correlation id to echo on its CatchupComplete. +// waveState tracks one Mutate's in-flight catch-up. A wave completes when its last +// scan drains (scansLeft reaches 0) — or early, when every topic it owned has been +// removed (owned reaches 0). topics/wire hold what its completion TopicsLive will +// announce, filtered against topicWave at completion so topics removed or reset +// mid-wave are not announced. type waveState struct { - remaining int mutateID uint64 + scansLeft int + owned int + topics []string + wire [][]byte } // Subscribe is the XIP-83 bidirectional subscription. One long-lived stream that the @@ -153,7 +163,6 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { // writer emits one CatchupComplete (echoing the Mutate's mutate_id) per wave once all // of its topics have opened. topicWave maps a not-yet-opened topic to its wave so a // remove mid-catch-up can free the wave slot it would otherwise wait on forever. - mutateSeen := false nextWave := 0 waves := make(map[int]waveState) topicWave := make(map[string]int) @@ -269,8 +278,11 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { } // buildGroupFrames dedups by group id (advancing the high-water mark, so no duplicates - // across catch-up/live) and packs the survivors into <=maxFrameBytes frames. - buildGroupFrames := func(msgs []*mlsv1.GroupMessage) []*mlsv1.SubscribeResponse { + // across catch-up/live) and packs the survivors into <=maxFrameBytes frames, each + // 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. + buildGroupFrames := func(msgs []*mlsv1.GroupMessage, mutateID uint64) []*mlsv1.SubscribeResponse { var frames []*mlsv1.SubscribeResponse var frame []*mlsv1.GroupMessage frameBytes := 0 @@ -280,7 +292,10 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { } frames = append(frames, wrap(&mlsv1.SubscribeResponse_V1{ Response: &mlsv1.SubscribeResponse_V1_Messages_{ - Messages: &mlsv1.SubscribeResponse_V1_Messages{GroupMessages: frame}, + Messages: &mlsv1.SubscribeResponse_V1_Messages{ + GroupMessages: frame, + MutateId: mutateID, + }, }, })) frame = nil @@ -304,7 +319,7 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { } // buildWelcomeFrames is the welcome-topic analogue of buildGroupFrames. - buildWelcomeFrames := func(msgs []*mlsv1.WelcomeMessage) []*mlsv1.SubscribeResponse { + buildWelcomeFrames := func(msgs []*mlsv1.WelcomeMessage, mutateID uint64) []*mlsv1.SubscribeResponse { var frames []*mlsv1.SubscribeResponse var frame []*mlsv1.WelcomeMessage frameBytes := 0 @@ -314,7 +329,10 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { } frames = append(frames, wrap(&mlsv1.SubscribeResponse_V1{ Response: &mlsv1.SubscribeResponse_V1_Messages_{ - Messages: &mlsv1.SubscribeResponse_V1_Messages{WelcomeMessages: frame}, + Messages: &mlsv1.SubscribeResponse_V1_Messages{ + WelcomeMessages: frame, + MutateId: mutateID, + }, }, })) frame = nil @@ -338,32 +356,6 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { return frames } - // buildOpenGateFrames drains a topic's live messages buffered during catch-up (deduped) - // into frames and clears its gate, so subsequent live messages send directly. - buildOpenGateFrames := func(topicStr string) []*mlsv1.SubscribeResponse { - buffered := pending[topicStr] - delete(pending, topicStr) - delete(catchingUp, topicStr) - if topic.IsMLSV1Welcome(topicStr) { - welcomes := make([]*mlsv1.WelcomeMessage, 0, len(buffered)) - for _, env := range buffered { - pendingBytes -= len(env.Message) - if m, err := getWelcomeMessageFromEnvelope(env); err == nil { - welcomes = append(welcomes, m) - } - } - return buildWelcomeFrames(welcomes) - } - groups := make([]*mlsv1.GroupMessage, 0, len(buffered)) - for _, env := range buffered { - pendingBytes -= len(env.Message) - if m, err := getGroupMessageFromEnvelope(env); err == nil { - groups = append(groups, m) - } - } - return buildGroupFrames(groups) - } - startedFrame := func() []*mlsv1.SubscribeResponse { return []*mlsv1.SubscribeResponse{wrap(&mlsv1.SubscribeResponse_V1{ Response: &mlsv1.SubscribeResponse_V1_Started_{ @@ -410,10 +402,13 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { if wave, ok := topicWave[t]; ok { delete(topicWave, t) if w, ok := waves[wave]; ok { - w.remaining-- - if w.remaining > 0 { + w.owned-- + if w.owned > 0 { waves[wave] = w } else { + // Every topic the wave owned is gone; complete it now (there is + // nothing left to announce) rather than wait for its scans, whose + // remaining batches the writer drops once the wave is deleted. delete(waves, wave) if err := send(catchupCompleteFrame(w.mutateID)); err != nil { return err @@ -424,6 +419,76 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { return nil } + // completeWave finishes a wave whose scans have all drained: it flushes the live + // messages buffered while its topics were gated — merged back into ascending id + // order and stamped with the wave's mutate_id, since the wave folds them in (their + // ids sit above the scan's ceiling) — then announces the surviving topics in one + // TopicsLive and emits the wave's CatchupComplete. Only then do the gates open, 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). Topics removed or reset + // mid-wave were settled by dropTopic and are skipped here. + completeWave := func(wave int, w waveState) error { + var groups []*mlsv1.GroupMessage + var welcomes []*mlsv1.WelcomeMessage + marker := &mlsv1.SubscribeResponse_V1_TopicsLive{} + for i, t := range w.topics { + if wv, ok := topicWave[t]; !ok || wv != wave { + continue // removed or reset mid-wave; its remove already settled it + } + delete(topicWave, t) + marker.Topics = append(marker.Topics, w.wire[i]) + if !catchingUp[t] { + // history_only: never gated, nothing buffered — and no live registration + // follows (both live-over-history_only directions are rejected in the Mutate + // handler), so drop the floor here or one-shot reads leak it forever. + delete(highWaterMarks, t) + continue + } + delete(catchingUp, t) + for _, env := range pending[t] { + pendingBytes -= len(env.Message) + if topic.IsMLSV1Welcome(t) { + if m, err := getWelcomeMessageFromEnvelope(env); err == nil { + welcomes = append(welcomes, m) + } else { + log.Error("error parsing welcome message", zap.Error(err)) + } + } else { + if m, err := getGroupMessageFromEnvelope(env); err == nil { + groups = append(groups, m) + } else { + log.Error("error parsing group message", zap.Error(err)) + } + } + } + delete(pending, t) + } + // Each topic's buffer is in dispatch (= id) order, but the wave's replay must + // stay totally ordered across its topics: merge before framing. The frame + // builders' high-water dedup then drops anything the scan already delivered — + // sound only under the same id-visibility-order invariant the ceiling pin rests + // on (see catchUpGroups). + sort.Slice(groups, func(i, j int) bool { + return groups[i].GetV1().GetId() < groups[j].GetV1().GetId() + }) + sort.Slice(welcomes, func(i, j int) bool { + return welcomeID(welcomes[i]) < welcomeID(welcomes[j]) + }) + var liveMarker []*mlsv1.SubscribeResponse + if len(marker.Topics) > 0 { + liveMarker = []*mlsv1.SubscribeResponse{wrap(&mlsv1.SubscribeResponse_V1{ + Response: &mlsv1.SubscribeResponse_V1_TopicsLive_{TopicsLive: marker}, + })} + } + delete(waves, wave) + return send( + buildGroupFrames(groups, w.mutateID), + buildWelcomeFrames(welcomes, w.mutateID), + liveMarker, + catchupCompleteFrame(w.mutateID), + ) + } + // Started must be the first frame, before any catch-up, so proxied/buffered transports // keep the connection open (XIP-83 server requirement 1). Sent here on the sole // goroutine, before any producer is spawned. @@ -442,180 +507,107 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { } } - // catchUpGroups fetches catch-up history for the given groups — chunked across the DB - // with bounded concurrency — and forwards each page to the writer. It owns no shared + // catchUpGroups replays one wave's group topics as a single ascending-id scan + // merged across all of its groups (XIP-83 server requirement 4: a wave's replay is + // one cursor-ordered pass, not per-topic bursts). The scan is pinned to the newest + // id at wave start, so it terminates under sustained publishing; anything newer + // reaches the client through the gated live path and is folded into the wave when + // it completes. Pages are forwarded to the writer in scan order. It owns no shared // state and never sends to the stream; it just queries and hands results over. - catchUpGroups := func(adds []mlsstore.GroupCatchup, topics []string, wire [][]byte, wave int) { - processChunk := func(chunk []mlsstore.GroupCatchup, chunkTopics []string, chunkWire [][]byte) { - cursors := make([]uint64, len(chunk)) - for i := range chunk { - cursors[i] = chunk[i].IdCursor + catchUpGroups := func(adds []mlsstore.GroupCatchup, wave int) { + // The ceiling pin assumes v3's id-visibility-order invariant: ids become visible to + // readers in id order (the live poller in worker.go advances from raw rows on the same + // assumption — a row committing out of order behind a reader is undeliverable + // stream-wide, a pre-existing v3 property). + ceiling, err := s.readOnlyStore.Queries().GetLatestGroupMessageID(ctx) + if err != nil { + if !errors.Is(err, context.Canceled) { + log.Error("wave-scan ceiling (group)", zap.Error(err)) + forward(catchUpBatch{err: err}) } - active := make([]int, len(chunk)) - for i := range chunk { - active[i] = i + return + } + scanCursor := uint64(0) + for { + select { + case <-ctx.Done(): + return + case <-s.ctx.Done(): + return + default: } - for len(active) > 0 { - select { - case <-ctx.Done(): - return - case <-s.ctx.Done(): - return - default: - } - filters := make([]mlsstore.GroupCatchup, len(active)) - for j, idx := range active { - filters[j] = mlsstore.GroupCatchup{ - GroupID: chunk[idx].GroupID, - IdCursor: cursors[idx], - } - } - msgs, err := s.readOnlyStore.QueryGroupMessagesBatch( - ctx, - filters, - catchUpPerGroupLimit, - ) - if err != nil { - if !errors.Is(err, context.Canceled) { - log.Error("batch catch-up (group)", zap.Error(err)) - forward(catchUpBatch{err: err}) - } - return - } - counts := make(map[string]int) - lastID := make(map[string]uint64) - for _, m := range msgs { - gid := string(m.GetV1().GetGroupId()) - counts[gid]++ - lastID[gid] = m.GetV1().GetId() - } - var opened []string - var openedWire [][]byte - var next []int - for _, idx := range active { - gid := string(chunk[idx].GroupID) - if counts[gid] == catchUpPerGroupLimit { - cursors[idx] = lastID[gid] - next = append(next, idx) - } else { - opened = append(opened, chunkTopics[idx]) - openedWire = append(openedWire, chunkWire[idx]) - } + msgs, err := s.readOnlyStore.QueryGroupMessagesWaveScan( + ctx, + adds, + scanCursor, + uint64(ceiling), + s.scanPageLimit, + ) + if err != nil { + if !errors.Is(err, context.Canceled) { + log.Error("wave-scan catch-up (group)", zap.Error(err)) + forward(catchUpBatch{err: err}) } - forward(catchUpBatch{ - groupMsgs: msgs, - opened: opened, - openedWire: openedWire, - wave: wave, - }) - active = next + return } - } - - sem := make(chan struct{}, catchUpConcurrency) - var chunkWg sync.WaitGroup - for start := 0; start < len(adds); start += catchUpChunkSize { - end := start + catchUpChunkSize - if end > len(adds) { - end = len(adds) + if len(msgs) > 0 { + scanCursor = msgs[len(msgs)-1].GetV1().GetId() + forward(catchUpBatch{groupMsgs: msgs, wave: wave}) + } + if len(msgs) < int(s.scanPageLimit) { + forward(catchUpBatch{wave: wave, done: true}) + return } - chunk, chunkTopics, chunkWire := adds[start:end], topics[start:end], wire[start:end] - chunkWg.Add(1) - sem <- struct{}{} - go func(chunk []mlsstore.GroupCatchup, chunkTopics []string, chunkWire [][]byte) { - defer chunkWg.Done() - defer func() { <-sem }() - processChunk(chunk, chunkTopics, chunkWire) - }(chunk, chunkTopics, chunkWire) } - chunkWg.Wait() } // catchUpWelcomes is the welcome-topic analogue of catchUpGroups. - catchUpWelcomes := func(adds []mlsstore.WelcomeCatchup, topics []string, wire [][]byte, wave int) { - processChunk := func(chunk []mlsstore.WelcomeCatchup, chunkTopics []string, chunkWire [][]byte) { - cursors := make([]uint64, len(chunk)) - for i := range chunk { - cursors[i] = chunk[i].IdCursor + catchUpWelcomes := func(adds []mlsstore.WelcomeCatchup, wave int) { + ceiling, err := s.readOnlyStore.Queries().GetLatestWelcomeMessageID(ctx) + if err != nil { + if !errors.Is(err, context.Canceled) { + log.Error("wave-scan ceiling (welcome)", zap.Error(err)) + forward(catchUpBatch{err: err}) } - active := make([]int, len(chunk)) - for i := range chunk { - active[i] = i + return + } + scanCursor := uint64(0) + for { + select { + case <-ctx.Done(): + return + case <-s.ctx.Done(): + return + default: } - for len(active) > 0 { - select { - case <-ctx.Done(): - return - case <-s.ctx.Done(): - return - default: - } - filters := make([]mlsstore.WelcomeCatchup, len(active)) - for j, idx := range active { - filters[j] = mlsstore.WelcomeCatchup{ - InstallationKey: chunk[idx].InstallationKey, - IdCursor: cursors[idx], - } - } - msgs, err := s.readOnlyStore.QueryWelcomeMessagesBatch( - ctx, - filters, - catchUpPerGroupLimit, - ) - if err != nil { - if !errors.Is(err, context.Canceled) { - log.Error("batch catch-up (welcome)", zap.Error(err)) - forward(catchUpBatch{err: err}) - } - return - } - counts := make(map[string]int) - lastID := make(map[string]uint64) - for _, m := range msgs { - key := string(welcomeInstallationKey(m)) - counts[key]++ - lastID[key] = welcomeID(m) - } - var opened []string - var openedWire [][]byte - var next []int - for _, idx := range active { - key := string(chunk[idx].InstallationKey) - if counts[key] == catchUpPerGroupLimit { - cursors[idx] = lastID[key] - next = append(next, idx) - } else { - opened = append(opened, chunkTopics[idx]) - openedWire = append(openedWire, chunkWire[idx]) - } + msgs, lastRawID, rawCount, err := s.readOnlyStore.QueryWelcomeMessagesWaveScan( + ctx, + adds, + scanCursor, + uint64(ceiling), + s.scanPageLimit, + ) + if err != nil { + if !errors.Is(err, context.Canceled) { + log.Error("wave-scan catch-up (welcome)", zap.Error(err)) + forward(catchUpBatch{err: err}) } - forward(catchUpBatch{ - welcomeMsgs: msgs, - opened: opened, - openedWire: openedWire, - wave: wave, - }) - active = next + return } - } - - sem := make(chan struct{}, catchUpConcurrency) - var chunkWg sync.WaitGroup - for start := 0; start < len(adds); start += catchUpChunkSize { - end := start + catchUpChunkSize - if end > len(adds) { - end = len(adds) + // Advance from the RAW scan position, not the parsed slice: the store skips rows + // with an unknown message_type but they still consumed LIMIT slots, so paging by + // parsed rows would silently truncate the replay at the first skipped row. + if rawCount > 0 { + scanCursor = lastRawID + } + if len(msgs) > 0 { + forward(catchUpBatch{welcomeMsgs: msgs, wave: wave}) + } + if rawCount < int(s.scanPageLimit) { + forward(catchUpBatch{wave: wave, done: true}) + return } - chunk, chunkTopics, chunkWire := adds[start:end], topics[start:end], wire[start:end] - chunkWg.Add(1) - sem <- struct{}{} - go func(chunk []mlsstore.WelcomeCatchup, chunkTopics []string, chunkWire [][]byte) { - defer chunkWg.Done() - defer func() { <-sem }() - processChunk(chunk, chunkTopics, chunkWire) - }(chunk, chunkTopics, chunkWire) } - chunkWg.Wait() } // Read client frames in a dedicated goroutine (gRPC Recv blocks) and forward them to @@ -700,29 +692,59 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { v1 := req.GetV1() if v1 == nil { // Unrecognized request version arm: fail rather than silently ignore, so a - // forward-version client is not left waiting on a response (XIP-83 req 8). + // forward-version client is not left waiting on a response (XIP-83 req 10). return status.Errorf(codes.InvalidArgument, "unrecognized SubscribeRequest version") } switch { case v1.GetMutate() != nil: m := v1.GetMutate() - // Removes are applied before adds, so a topic appearing in both is reset: - // removed (clearing its cursor floor), then re-added with a fresh catch-up. - // dropTopic owns the gauge/metric for any topic that was actually live, so a - // duplicate or unknown remove is harmless rather than drifting the count. - for _, wire := range m.GetRemoves() { - kind, id, err := splitTopic(wire) - if err != nil { - return status.Errorf(codes.InvalidArgument, "remove: %v", err) - } - if err := dropTopic(buildMLSTopic(kind, id)); err != nil { - return err + // A wave's replay frames are stamped with its mutate_id, and 0 is the live + // tag, so a wave of adds cannot ride on 0 (XIP-83 server requirement 3). + // Validated before any state changes so the rejected frame is atomic. + if len(m.GetAdds()) > 0 && m.GetMutateId() == 0 { + return status.Errorf( + codes.InvalidArgument, + "a Mutate with adds requires a nonzero mutate_id", + ) + } + // Each add becomes a floor entry in the wave's merged catch-up scan, + // riding every page query (see maxMutateAdds). Checked on the raw adds — + // stateless, before any state changes — so the rejected frame is atomic. + if len(m.GetAdds()) > s.maxMutateAdds { + return status.Errorf( + codes.ResourceExhausted, + "adds-per-Mutate limit %d exceeded; split the adds across Mutates", + s.maxMutateAdds, + ) + } + // The frame tag and the CatchupComplete echo are the only keys correlating + // frames to mutations, so a mutate_id reused while its wave is still in + // flight would make the two waves' replay and completions indistinguishable + // (XIP-83 server requirement 3). This applies to ANY Mutate — even a + // removes-only one, whose immediate CatchupComplete would be ambiguous with + // the in-flight wave's — and is checked before the removes' side effects. + // In-flight ids are always nonzero (waves start only from adds-bearing + // Mutates, rejected above on 0), so a mutate_id of 0 never collides. Reuse + // after a wave's CatchupComplete stays legal. + if m.GetMutateId() != 0 { + for _, w := range waves { + if w.mutateID == m.GetMutateId() { + return status.Errorf( + codes.InvalidArgument, + "mutate_id %d is already in flight on this stream", + m.GetMutateId(), + ) + } } } - // history_only adds never touch the dispatcher: no live registration, // no gate, no pending buffer — a pure batched read with markers. historyOnly := m.GetHistoryOnly() + // Parse and kind-validate every add up front — pure parsing, no state + // decisions — so a malformed add fails the stream BEFORE any remove's side + // effects (dropTopic, including a freed wave's CatchupComplete) take hold. + // The state-dependent no-op/reset decisions stay below the removes: a + // same-Mutate remove+re-add must see the removed state. // Collapse duplicate topics within this Mutate's adds, lowest id_cursor // winning, so a repeated topic resolves deterministically and a lower cursor // still drives the replay path below. @@ -752,6 +774,20 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { addOrder = append(addOrder, t) } + // Removes are applied before adds, so a topic appearing in both is reset: + // removed (clearing its cursor floor), then re-added with a fresh catch-up. + // dropTopic owns the gauge/metric for any topic that was actually live, so a + // duplicate or unknown remove is harmless rather than drifting the count. + for _, wire := range m.GetRemoves() { + kind, id, err := splitTopic(wire) + if err != nil { + return status.Errorf(codes.InvalidArgument, "remove: %v", err) + } + if err := dropTopic(buildMLSTopic(kind, id)); err != nil { + return err + } + } + var groupAdds []mlsstore.GroupCatchup var groupTopics []string var groupWire [][]byte @@ -786,7 +822,7 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { // so honoring it would have to disturb the live subscription's floor — and // on the replay path (cursor below the floor) dropTopic would unsubscribe // the topic without re-registering it, silently severing a live tail. - // Reject rather than guess (XIP-83 req 8 stance: fail contradictory input). + // Reject rather than guess (XIP-83 req 11 stance: fail contradictory input). if historyOnly { return status.Errorf( codes.InvalidArgument, @@ -830,17 +866,30 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { } // Each mutate that catches up subscriptions starts a wave; its - // CatchupComplete (echoing mutate_id) is emitted once all of the wave's - // topics are live. A mutate whose adds were all already-live (no-ops) or that - // added nothing still gets an immediate CatchupComplete — both so the client's - // mutate_id is answered and so a client that subscribed nothing learns it is - // live. A pure remove-only mutate after the first yields neither. + // CatchupComplete (echoing mutate_id) is emitted once its scans drain and + // its gates open. Every other mutate — removes-only, empty, or adds that + // were all no-ops — is acknowledged with an immediate CatchupComplete, so + // every Mutate is answered by exactly one CatchupComplete echoing its + // mutate_id. adds := len(groupAdds) + len(welcomeAdds) switch { case adds > 0: wave := nextWave nextWave++ - waves[wave] = waveState{remaining: adds, mutateID: m.GetMutateId()} + scans := 0 + if len(groupAdds) > 0 { + scans++ + } + if len(welcomeAdds) > 0 { + scans++ + } + waves[wave] = waveState{ + mutateID: m.GetMutateId(), + scansLeft: scans, + owned: adds, + topics: append(append([]string{}, groupTopics...), welcomeTopics...), + wire: append(append([][]byte{}, groupWire...), welcomeWire...), + } for _, t := range groupTopics { topicWave[t] = wave } @@ -848,17 +897,16 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { topicWave[t] = wave } if len(groupAdds) > 0 { - go catchUpGroups(groupAdds, groupTopics, groupWire, wave) + go catchUpGroups(groupAdds, wave) } if len(welcomeAdds) > 0 { - go catchUpWelcomes(welcomeAdds, welcomeTopics, welcomeWire, wave) + go catchUpWelcomes(welcomeAdds, wave) } - case len(m.GetAdds()) > 0 || !mutateSeen: + default: if err := send(catchupCompleteFrame(m.GetMutateId())); err != nil { return err } } - mutateSeen = true case v1.GetPing() != nil: nonce := v1.GetPing().GetNonce() @@ -883,10 +931,13 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { // CATCHUP_COMPLETE. return status.Errorf(codes.Unavailable, "catch-up failed: %v", b.err) } - // A topic can be removed (or reset) while its catch-up page is in flight. wanted - // reports whether this batch's wave still owns the topic; history, the gate - // flush, TopicsLive and the wave count all skip topics no longer wanted, so a - // removed topic never gets stale history, a phantom marker, or a miscounted wave. + w, waveActive := waves[b.wave] + if !waveActive { + continue // the wave completed early (all topics removed); drop the straggler + } + // A topic can be removed (or reset) while its wave's scan page is in flight. + // wanted reports whether this batch's wave still owns the topic, so a removed + // topic never gets stale history. Dropping rows preserves the scan's order. wanted := func(t string) bool { wv, ok := topicWave[t] return ok && wv == b.wave @@ -899,7 +950,7 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { kept = append(kept, m) } } - history = buildGroupFrames(kept) + history = buildGroupFrames(kept, w.mutateID) } else if len(b.welcomeMsgs) > 0 { kept := b.welcomeMsgs[:0] for _, m := range b.welcomeMsgs { @@ -907,44 +958,20 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { kept = append(kept, m) } } - history = buildWelcomeFrames(kept) - } - var openFrames []*mlsv1.SubscribeResponse - var liveMarker []*mlsv1.SubscribeResponse - openedCount := 0 - if len(b.opened) > 0 { - marker := &mlsv1.SubscribeResponse_V1_TopicsLive{} - for i, t := range b.opened { - if wv, ok := topicWave[t]; !ok || wv != b.wave { - continue // removed or reset mid-catch-up; its remove already settled it - } - delete(topicWave, t) - openFrames = append(openFrames, buildOpenGateFrames(t)...) - marker.Topics = append(marker.Topics, b.openedWire[i]) - openedCount++ - } - if openedCount > 0 { - liveMarker = []*mlsv1.SubscribeResponse{wrap(&mlsv1.SubscribeResponse_V1{ - Response: &mlsv1.SubscribeResponse_V1_TopicsLive_{TopicsLive: marker}, - })} - } + history = buildWelcomeFrames(kept, w.mutateID) } - // Order is just program order here: history, then the flushed pending buffer - // (live messages that queued behind the catch-up — equally historical from the - // client's perspective), and only then the TopicsLive marker, so every frame a - // client sees after the marker really is live tail. - if err := send(history, openFrames, liveMarker); err != nil { + if err := send(history); err != nil { return err } - if w, ok := waves[b.wave]; ok { - w.remaining -= openedCount - if w.remaining > 0 { + if b.done { + w.scansLeft-- + if w.scansLeft > 0 { waves[b.wave] = w } else { - // The wave's last topic just went live; its CatchupComplete (echoing - // the Mutate's id) follows the wave's final TopicsLive in program order. - delete(waves, b.wave) - if err := send(catchupCompleteFrame(w.mutateID)); err != nil { + // The wave's last scan just drained: flush its gated live buffer (still + // the wave's replay), announce TopicsLive, emit its CatchupComplete, and + // open the gates — in that order, so the seam holds. + if err := completeWave(b.wave, w); err != nil { return err } if halfClosed && len(waves) == 0 { @@ -971,16 +998,19 @@ func (s *Service) Subscribe(stream mlsv1.MlsApi_SubscribeServer) error { } continue } + // Live tail: tagged 0. The dispatcher delivers each kind in ascending id + // order and the writer sends in arrival order, so the live lane stays + // totally ordered per kind (XIP-83 server requirement 4). var frames []*mlsv1.SubscribeResponse if topic.IsMLSV1Welcome(t) { if m, err := getWelcomeMessageFromEnvelope(env); err == nil { - frames = buildWelcomeFrames([]*mlsv1.WelcomeMessage{m}) + frames = buildWelcomeFrames([]*mlsv1.WelcomeMessage{m}, 0) } else { log.Error("error parsing welcome message", zap.Error(err)) } } else { if m, err := getGroupMessageFromEnvelope(env); err == nil { - frames = buildGroupFrames([]*mlsv1.GroupMessage{m}) + frames = buildGroupFrames([]*mlsv1.GroupMessage{m}, 0) } else { log.Error("error parsing group message", zap.Error(err)) } diff --git a/pkg/mls/api/v1/subscribe_test.go b/pkg/mls/api/v1/subscribe_test.go index 9e363614..e2c62a2b 100644 --- a/pkg/mls/api/v1/subscribe_test.go +++ b/pkg/mls/api/v1/subscribe_test.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "math" + "runtime" "sync" "testing" "time" @@ -333,7 +334,8 @@ func TestSubscribe_CatchUpThenLiveNoDuplicates(t *testing.T) { // Subscribe from the beginning, then immediately publish live messages so they race // the catch-up. stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ - Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupId), 0)}, + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupId), 0)}, + MutateId: 1, })) const live = 10 populateGroupMessages(t, ctx, svc, groupId, live, "live") @@ -388,6 +390,7 @@ func TestSubscribe_MutateRemoveStopsDelivery(t *testing.T) { addSub(groupTopic(groupId), 0), addSub(groupTopic(sentinelId), 0), }, + MutateId: 1, })) waitForResponses( t, @@ -565,6 +568,7 @@ func TestSubscribe_MultiplexesMultipleIdentities(t *testing.T) { addSub(welcomeTopic(instA), 0), addSub(welcomeTopic(instB), 0), }, + MutateId: 1, })) // Catch-up delivers each identity's history, correctly attributed. @@ -789,13 +793,17 @@ func TestSubscribe_HistoryOnlyDeliversNoLive(t *testing.T) { // Wave 1: a live sentinel subscription. Wave 2: groupA, history only. stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ - Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(sentinelId), 0)}, + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(sentinelId), 0), + }, + MutateId: 1, })) stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ addSub(groupTopic(groupA), 0), }, HistoryOnly: true, + MutateId: 2, })) markerAPred := func(r *mlsv1.SubscribeResponse) bool { @@ -859,7 +867,8 @@ func TestSubscribe_HistoryOnlyOnLiveTopicRejected(t *testing.T) { // Subscribe groupA live and wait until its catch-up completes — the cursor floor is now // above 0, so the history_only re-add below lands on the replay path. stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ - Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + MutateId: 1, })) waitForResponses( t, @@ -880,6 +889,7 @@ func TestSubscribe_HistoryOnlyOnLiveTopicRejected(t *testing.T) { addSub(groupTopic(groupA), 0), }, HistoryOnly: true, + MutateId: 2, })) select { @@ -1127,7 +1137,9 @@ func TestSubscribe_DuplicateAddsDeduped(t *testing.T) { } // TestSubscribe_ReplayAfterRemove verifies removing a topic clears its cursor floor, so a -// later re-add replays the history again (XIP-83 replay-after-remove). +// later re-add replays the history again (XIP-83 replay-after-remove) — and that each +// removes-only Mutate is acked with an immediate CatchupComplete echoing its mutate_id, +// including mutate_id 0. func TestSubscribe_ReplayAfterRemove(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -1156,9 +1168,11 @@ func TestSubscribe_ReplayAfterRemove(t *testing.T) { }, ) - // Remove, then re-add from cursor 0: the cleared floor lets the history replay. + // Remove, then re-add from cursor 0: the cleared floor lets the history replay. The + // removes-only Mutate spawns no wave, so it is acked immediately, echoing its id. stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ - Removes: [][]byte{groupTopic(groupA)}, + Removes: [][]byte{groupTopic(groupA)}, + MutateId: 5, })) stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, @@ -1169,10 +1183,24 @@ func TestSubscribe_ReplayAfterRemove(t *testing.T) { stream, 10*time.Second, "replayed catch-up (wave 2)", - func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 2 }, + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 3 }, ) require.Len(t, groupMsgsFrom(resps), 10, "remove + re-add must replay the full history") - require.Equal(t, []uint64{1, 2}, catchupCompletesFrom(resps)) + require.Equal(t, []uint64{1, 5, 2}, catchupCompletesFrom(resps)) + + // A removes-only Mutate with mutate_id 0 is valid (0 only rejects adds) and is acked + // with a CatchupComplete echoing 0. + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Removes: [][]byte{groupTopic(groupA)}, + })) + resps = waitForResponses( + t, + stream, + 5*time.Second, + "ack for the mutate_id-0 remove", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 4 }, + ) + require.Equal(t, []uint64{1, 5, 2, 0}, catchupCompletesFrom(resps)) stream.closeSend() require.NoError(t, <-errCh) @@ -1217,9 +1245,9 @@ func TestSubscribe_CursorAboveInt64ReturnsNoHistory(t *testing.T) { require.NoError(t, <-errCh) } -// TestSubscribe_InboundFramesResetIdleTimer verifies that inbound client frames which produce -// no response (here, no-op removes) still reset the idle timer, so the node does not ping or -// reap a stream the client is actively using. +// TestSubscribe_InboundFramesDoNotSuppressLivenessPing verifies that inbound client frames +// which produce no response (here, unsolicited Pongs) do not defer the liveness Ping, so a +// peer that sends but never reads is still probed and reaped. func TestSubscribe_InboundFramesDoNotSuppressLivenessPing(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -1236,19 +1264,19 @@ func TestSubscribe_InboundFramesDoNotSuppressLivenessPing(t *testing.T) { func(rs []*mlsv1.SubscribeResponse) bool { return hasStarted(rs) }) // The liveness Ping probes the client's RECEIVE path and is driven by SEND-side idleness - // only. A client streaming response-free frames (no-op removes) must NOT suppress it — - // otherwise a peer that sends but never reads could never be reaped. + // only. A client streaming response-free frames (unsolicited Pongs — every Mutate now + // yields a CatchupComplete) must NOT suppress it — otherwise a peer that sends but never + // reads could never be reaped. done := make(chan struct{}) defer close(done) go func() { - dummy := groupTopic([]byte(test.RandomString(32))) for { select { case <-done: return default: } - stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{Removes: [][]byte{dummy}})) + stream.send(subReqPong(999999)) // never a nonce the server is waiting on time.Sleep(40 * time.Millisecond) } }() @@ -1279,6 +1307,7 @@ func TestSubscribe_GroupAndWelcomeSameIdentifierNoCollision(t *testing.T) { addSub(groupTopic(id), 0), addSub(welcomeTopic(id), 0), }, + MutateId: 1, })) waitForResponses( t, @@ -1316,9 +1345,11 @@ func TestSubscribe_GroupAndWelcomeSameIdentifierNoCollision(t *testing.T) { require.NoError(t, <-errCh) } -// TestQueryGroupMessagesBatch_CursorAboveInt64ReturnsNothing exercises clampCursor directly: -// the Subscribe-level test is masked by the writer's high-water mark, so guard the store here. -func TestQueryGroupMessagesBatch_CursorAboveInt64ReturnsNothing(t *testing.T) { +// TestQueryGroupMessagesWaveScan_CursorAboveInt64ReturnsNothing exercises clampCursor +// directly: the Subscribe-level test is masked by the writer's high-water mark, so guard +// the store here. The ceiling above MaxInt64 clamps too (to "everything"), so the empty +// result is attributable to the cursor alone. +func TestQueryGroupMessagesWaveScan_CursorAboveInt64ReturnsNothing(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() svc, _, validationSvc, cleanup := newTestService(t, ctx) @@ -1328,9 +1359,11 @@ func TestQueryGroupMessagesBatch_CursorAboveInt64ReturnsNothing(t *testing.T) { mockValidateGroupMessages(validationSvc, groupA) populateGroupMessages(t, ctx, svc, groupA, 5, "hist") - msgs, err := svc.readOnlyStore.QueryGroupMessagesBatch( + msgs, err := svc.readOnlyStore.QueryGroupMessagesWaveScan( ctx, []mlsstore.GroupCatchup{{GroupID: groupA, IdCursor: math.MaxUint64}}, + 0, + math.MaxUint64, 50, ) require.NoError(t, err) @@ -1338,17 +1371,18 @@ func TestQueryGroupMessagesBatch_CursorAboveInt64ReturnsNothing(t *testing.T) { } // TestSubscribe_MultiPageCatchUp exercises catch-up pagination across more than -// catchUpPerGroupLimit messages: the active-loop continuation plus open-on-final-page wave -// accounting, with exactly one TopicsLive and one CatchupComplete. +// scanPageLimit messages: the scan-cursor continuation plus wave completion on the final +// short page, with exactly one TopicsLive and one CatchupComplete. func TestSubscribe_MultiPageCatchUp(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() svc, _, validationSvc, cleanup := newTestService(t, ctx) defer cleanup() + svc.scanPageLimit = 50 // small pages so a modest history spans several groupA := []byte(test.RandomString(32)) mockValidateGroupMessages(validationSvc, groupA) - const history = catchUpPerGroupLimit*2 + 17 + const history = 50*2 + 17 populateGroupMessages(t, ctx, svc, groupA, history, "hist") stream := newFakeSubscribeStream(ctx) @@ -1399,7 +1433,8 @@ func TestSubscribe_NonZeroStartingCursor(t *testing.T) { e1 := make(chan error, 1) go func() { e1 <- svc.Subscribe(s1) }() s1.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ - Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + MutateId: 1, })) r1 := waitForResponses(t, s1, 10*time.Second, "all 10", func(rs []*mlsv1.SubscribeResponse) bool { return len(groupMsgsFrom(rs)) >= 10 }) @@ -1417,7 +1452,10 @@ func TestSubscribe_NonZeroStartingCursor(t *testing.T) { e2 := make(chan error, 1) go func() { e2 <- svc.Subscribe(s2) }() s2.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ - Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), cursor)}, + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(groupA), cursor), + }, + MutateId: 2, })) r2 := waitForResponses(t, s2, 10*time.Second, "messages after the cursor", func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) @@ -1431,20 +1469,61 @@ func TestSubscribe_NonZeroStartingCursor(t *testing.T) { } // fakeReadStore wraps a ReadMlsStore to inject catch-up faults: groupErr makes -// QueryGroupMessagesBatch fail, and groupGate (if non-nil) blocks it until released, letting -// a test hold a topic mid-catch-up. All other methods delegate to the embedded store. +// QueryGroupMessagesWaveScan fail, and groupGate / welcomeGate (if non-nil) block the +// corresponding wave scan until released, letting a test hold a topic mid-catch-up. +// gateGroup / gateInstallation (if set) narrow the gate to scans covering that topic, so +// other waves on the same stream proceed. scanParked (if non-nil) is closed when a gated +// scan first parks — the wave's ceiling is pinned by then, so a test can publish +// deterministically above it. corruptWelcomeIDs (if set) simulates welcome rows the store +// skips as unparseable (an unknown message_type from a future writer): they are dropped +// from the parsed slice but keep their raw-scan slot (lastRawID / rawCount), matching the +// real store's paging contract. All other methods delegate to the embedded store +// (including Queries(), so the wave's ceiling lookup is unaffected). type fakeReadStore struct { mlsstore.ReadMlsStore - groupErr error - groupGate chan struct{} + groupErr error + groupGate chan struct{} + gateGroup []byte + welcomeGate chan struct{} + gateInstallation []byte + scanParked chan struct{} + parkedOnce sync.Once + corruptWelcomeIDs map[uint64]bool +} + +func filtersInclude(filters []mlsstore.GroupCatchup, groupID []byte) bool { + for _, flt := range filters { + if string(flt.GroupID) == string(groupID) { + return true + } + } + return false +} + +func welcomeFiltersInclude(filters []mlsstore.WelcomeCatchup, installationKey []byte) bool { + for _, flt := range filters { + if string(flt.InstallationKey) == string(installationKey) { + return true + } + } + return false +} + +func (f *fakeReadStore) parked() { + if f.scanParked != nil { + f.parkedOnce.Do(func() { close(f.scanParked) }) + } } -func (f *fakeReadStore) QueryGroupMessagesBatch( +func (f *fakeReadStore) QueryGroupMessagesWaveScan( ctx context.Context, filters []mlsstore.GroupCatchup, - perGroupLimit int32, + scanCursor uint64, + ceiling uint64, + limit int32, ) ([]*mlsv1.GroupMessage, error) { - if f.groupGate != nil { + if f.groupGate != nil && (f.gateGroup == nil || filtersInclude(filters, f.gateGroup)) { + f.parked() select { case <-f.groupGate: case <-ctx.Done(): @@ -1454,7 +1533,38 @@ func (f *fakeReadStore) QueryGroupMessagesBatch( if f.groupErr != nil { return nil, f.groupErr } - return f.ReadMlsStore.QueryGroupMessagesBatch(ctx, filters, perGroupLimit) + return f.ReadMlsStore.QueryGroupMessagesWaveScan(ctx, filters, scanCursor, ceiling, limit) +} + +func (f *fakeReadStore) QueryWelcomeMessagesWaveScan( + ctx context.Context, + filters []mlsstore.WelcomeCatchup, + scanCursor uint64, + ceiling uint64, + limit int32, +) ([]*mlsv1.WelcomeMessage, uint64, int, error) { + if f.welcomeGate != nil && + (f.gateInstallation == nil || welcomeFiltersInclude(filters, f.gateInstallation)) { + f.parked() + select { + case <-f.welcomeGate: + case <-ctx.Done(): + return nil, 0, 0, ctx.Err() + } + } + msgs, lastRawID, rawCount, err := f.ReadMlsStore.QueryWelcomeMessagesWaveScan( + ctx, filters, scanCursor, ceiling, limit, + ) + if err == nil && len(f.corruptWelcomeIDs) > 0 { + kept := msgs[:0] + for _, m := range msgs { + if !f.corruptWelcomeIDs[welcomeID(m)] { + kept = append(kept, m) + } + } + msgs = kept + } + return msgs, lastRawID, rawCount, err } // TestSubscribe_NonReadingClientIsReaped verifies a connected-but-non-reading client (a @@ -1520,6 +1630,7 @@ func TestSubscribe_HalfCloseFlushTimeoutFailsNotOK(t *testing.T) { addSub(groupTopic(groupA), 0), }, HistoryOnly: true, + MutateId: 1, })) stream.closeSend() @@ -1556,7 +1667,8 @@ func TestSubscribe_CatchUpFetchErrorTearsDownUnavailable(t *testing.T) { errCh := make(chan error, 1) go func() { errCh <- svc.Subscribe(stream) }() stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ - Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + MutateId: 1, })) select { @@ -1601,9 +1713,13 @@ func TestSubscribe_RemoveMidCatchUpFreesWaveSlot(t *testing.T) { ) resps := waitForResponses(t, stream, 5*time.Second, "CatchupComplete from the remove", - func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) - require.Equal(t, []uint64{7}, catchupCompletesFrom(resps), - "the removed topic's wave completes from the remove, echoing its mutate_id") + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 2 }) + require.Equal( + t, + []uint64{7, 0}, + catchupCompletesFrom(resps), + "the removed topic's wave completes from the remove (echoing its mutate_id), then the removes-only Mutate is acked", + ) require.Empty( t, groupMsgsFrom(resps), @@ -1634,7 +1750,8 @@ func TestSubscribe_PendingBufferCapAborts(t *testing.T) { errCh := make(chan error, 1) go func() { errCh <- svc.Subscribe(stream) }() stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ - Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + MutateId: 1, })) // Catch-up is gated, so the topic stays in catch-up; live messages pile into pending. @@ -1669,7 +1786,8 @@ func TestSubscribe_FramesSplitByMaxFrameBytes(t *testing.T) { errCh := make(chan error, 1) go func() { errCh <- svc.Subscribe(stream) }() stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ - Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + MutateId: 1, })) resps := waitForResponses(t, stream, 10*time.Second, "all history", func(rs []*mlsv1.SubscribeResponse) bool { return len(groupMsgsFrom(rs)) >= n }) @@ -1686,3 +1804,1212 @@ func TestSubscribe_FramesSplitByMaxFrameBytes(t *testing.T) { stream.closeSend() require.NoError(t, <-errCh) } + +// --- XIP-83 delivery tagging & ordering (server requirements 3 and 4) --- + +// groupMsgsTagged returns the group messages carried by Messages frames whose mutate_id +// equals tag, in receive order. +func groupMsgsTagged(resps []*mlsv1.SubscribeResponse, tag uint64) []*mlsv1.GroupMessage { + var out []*mlsv1.GroupMessage + for _, r := range resps { + if msgs := r.GetV1().GetMessages(); msgs != nil && msgs.GetMutateId() == tag { + out = append(out, msgs.GetGroupMessages()...) + } + } + return out +} + +// welcomeMsgsTagged is the welcome analogue of groupMsgsTagged. +func welcomeMsgsTagged(resps []*mlsv1.SubscribeResponse, tag uint64) []*mlsv1.WelcomeMessage { + var out []*mlsv1.WelcomeMessage + for _, r := range resps { + if msgs := r.GetV1().GetMessages(); msgs != nil && msgs.GetMutateId() == tag { + out = append(out, msgs.GetWelcomeMessages()...) + } + } + return out +} + +// requireAscendingGroupIds asserts the messages' ids strictly ascend in the given order — +// the total-order shape both delivery lanes guarantee per kind. +func requireAscendingGroupIds(t *testing.T, msgs []*mlsv1.GroupMessage, desc string) { + t.Helper() + for i := 1; i < len(msgs); i++ { + require.Greater( + t, + msgs[i].GetV1().GetId(), + msgs[i-1].GetV1().GetId(), + "%s: ids must strictly ascend (index %d)", + desc, + i, + ) + } +} + +// requireAscendingWelcomeIds is the welcome analogue of requireAscendingGroupIds. +func requireAscendingWelcomeIds(t *testing.T, msgs []*mlsv1.WelcomeMessage, desc string) { + t.Helper() + for i := 1; i < len(msgs); i++ { + require.Greater( + t, + welcomeID(msgs[i]), + welcomeID(msgs[i-1]), + "%s: welcome ids must strictly ascend (index %d)", + desc, + i, + ) + } +} + +// TestSubscribe_ReplayTaggedWithWaveLiveTaggedZero pins delivery tagging under overlapping +// waves: every replay frame carries exactly the mutate_id of the wave that produced it — +// even when a later wave completes first — and live frames carry 0. +func TestSubscribe_ReplayTaggedWithWaveLiveTaggedZero(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + groupA := []byte(test.RandomString(32)) + instB := []byte(test.RandomString(32)) + hpke := []byte(test.RandomString(32)) + mockValidateGroupMessages(validationSvc, groupA) + populateGroupMessages(t, ctx, svc, groupA, 5, "histA") + populateWelcomeMessages(t, ctx, svc, instB, hpke, 3, "welB") + + // Gate group scans so wave 7 stays in flight while wave 8 (welcomes) overtakes it. + gate := make(chan struct{}) + svc.readOnlyStore = &fakeReadStore{ReadMlsStore: svc.readOnlyStore, groupGate: gate} + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + MutateId: 7, + })) + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(welcomeTopic(instB), 0)}, + MutateId: 8, + })) + + // Wave 8 completes while wave 7 is still gated mid-fetch. + resps := waitForResponses(t, stream, 10*time.Second, "wave 8 catch-up", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + require.Equal(t, []uint64{8}, catchupCompletesFrom(resps), "wave 8 overtakes the gated wave 7") + require.Len(t, welcomeMsgsTagged(resps, 8), 3, "wave 8's replay is tagged 8") + require.Empty(t, groupMsgsTagged(resps, 7), "wave 7 is still gated: no frames yet") + + // Release wave 7: its replay arrives tagged 7, never mixed into another tag. + close(gate) + resps = waitForResponses(t, stream, 10*time.Second, "wave 7 catch-up", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 2 }) + require.Equal(t, []uint64{8, 7}, catchupCompletesFrom(resps)) + require.Len(t, groupMsgsTagged(resps, 7), 5, "wave 7's replay is tagged 7") + require.Empty(t, groupMsgsTagged(resps, 8), "no group frame may carry wave 8's tag") + require.Empty(t, groupMsgsTagged(resps, 0), "no replay frame may carry the live tag") + + // Live tail on both topics is tagged 0. + publishGroup(t, ctx, svc, validationSvc, groupA, "a-live") + populateWelcomeMessages(t, ctx, svc, instB, hpke, 1, "welB-live") + resps = waitForResponses(t, stream, 10*time.Second, "live tail", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsTagged(rs, 0)) >= 1 && len(welcomeMsgsTagged(rs, 0)) >= 1 + }) + require.Len(t, groupMsgsWithData(groupMsgsTagged(resps, 0), "a-live"), 1) + require.Len(t, welcomeMsgsTagged(resps, 0), 1) + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_InflightMutateIdCollisionRejected verifies that a Mutate reusing the +// mutate_id of a wave still in flight is rejected with InvalidArgument: the frame tag and +// the CatchupComplete echo are the only keys correlating frames to mutations, so an +// in-flight collision would make the two waves' replay and completions indistinguishable +// (XIP-83 server requirement 3). The gate holds wave 7 mid-fetch; both an adds-bearing +// Mutate and a removes-only Mutate (whose immediate CatchupComplete would be ambiguous +// with the in-flight wave's) must be rejected. +func TestSubscribe_InflightMutateIdCollisionRejected(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + groupA := []byte(test.RandomString(32)) + instB := []byte(test.RandomString(32)) + mockValidateGroupMessages(validationSvc, groupA) + populateGroupMessages(t, ctx, svc, groupA, 5, "hist") + gate := make(chan struct{}) + defer close(gate) + svc.readOnlyStore = &fakeReadStore{ReadMlsStore: svc.readOnlyStore, groupGate: gate} + + // Each rejection fails its stream, so each colliding Mutate gets a fresh stream with + // its own gated wave 7 in flight. + rejected := func(desc string, colliding *mlsv1.SubscribeRequest_V1_Mutate) { + t.Helper() + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + // Wave 7's catch-up blocks on the gate, staying in flight. + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(groupA), 0), + }, + MutateId: 7, + })) + stream.send(subReqMutate(colliding)) + + select { + case err := <-errCh: + require.Equal(t, codes.InvalidArgument, status.Code(err), desc) + case <-time.After(5 * time.Second): + t.Fatalf("%s: colliding Mutate was not rejected", desc) + } + } + + rejected("adds reusing an in-flight mutate_id must be rejected", + &mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(welcomeTopic(instB), 0), + }, + MutateId: 7, + }) + rejected("a removes-only Mutate reusing an in-flight mutate_id must be rejected", + &mlsv1.SubscribeRequest_V1_Mutate{ + Removes: [][]byte{welcomeTopic(instB)}, + MutateId: 7, + }) +} + +// TestSubscribe_MutateIdReuseAfterCompletionAccepted pins the boundary of the in-flight +// collision rule: once a wave's CatchupComplete has been emitted its mutate_id is free +// again, so a later adds-Mutate reusing it replays and completes normally. +func TestSubscribe_MutateIdReuseAfterCompletionAccepted(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + groupA := []byte(test.RandomString(32)) + instB := []byte(test.RandomString(32)) + hpke := []byte(test.RandomString(32)) + mockValidateGroupMessages(validationSvc, groupA) + populateGroupMessages(t, ctx, svc, groupA, 5, "histA") + populateWelcomeMessages(t, ctx, svc, instB, hpke, 3, "welB") + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + // Wave 7 runs to completion. + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + MutateId: 7, + })) + waitForResponses(t, stream, 10*time.Second, "wave 7 catch-up", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + + // The id is no longer in flight: a fresh wave reusing 7 replays and completes again. + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(welcomeTopic(instB), 0)}, + MutateId: 7, + })) + resps := waitForResponses(t, stream, 10*time.Second, "reused wave 7 catch-up", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 2 }) + require.Equal(t, []uint64{7, 7}, catchupCompletesFrom(resps), + "each wave completes with its own CatchupComplete echoing the reused id") + require.Len(t, groupMsgsTagged(resps, 7), 5, "the first wave's replay is tagged 7") + require.Len(t, welcomeMsgsTagged(resps, 7), 3, "the reused wave's replay is tagged 7") + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_WaveReplayMergedAcrossTopics pins wave order: one wave covering two groups +// whose histories interleave by id must replay the union in ascending id order — one merged +// cursor-ordered pass — not one topic's burst then the other's. +func TestSubscribe_WaveReplayMergedAcrossTopics(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + groupA := []byte(test.RandomString(32)) + groupB := []byte(test.RandomString(32)) + // Alternate publishes so the two groups' ids interleave globally. + const perGroup = 6 + for i := 0; i < perGroup; i++ { + publishGroup(t, ctx, svc, validationSvc, groupA, fmt.Sprintf("a-%d", i)) + publishGroup(t, ctx, svc, validationSvc, groupB, fmt.Sprintf("b-%d", i)) + } + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(groupA), 0), + addSub(groupTopic(groupB), 0), + }, + MutateId: 3, + })) + resps := waitForResponses(t, stream, 10*time.Second, "merged replay + CatchupComplete", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsFrom(rs)) >= 2*perGroup && len(catchupCompletesFrom(rs)) >= 1 + }) + + replay := groupMsgsTagged(resps, 3) + require.Len(t, replay, 2*perGroup, "the wave delivers both groups' history, tagged 3") + requireAscendingGroupIds(t, replay, "wave replay across interleaved topics") + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_LiveTotalOrderPerKind pins live order: live (mutate_id 0) group messages +// across all live topics on the stream arrive in ascending group id order, and live +// welcomes across all live installations in ascending welcome id order — each kind totally +// ordered on its own id sequence, independent of the other's. +func TestSubscribe_LiveTotalOrderPerKind(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + groupA := []byte(test.RandomString(32)) + groupB := []byte(test.RandomString(32)) + instA := []byte(test.RandomString(32)) + instB := []byte(test.RandomString(32)) + hpke := []byte(test.RandomString(32)) + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(groupA), 0), + addSub(groupTopic(groupB), 0), + addSub(welcomeTopic(instA), 0), + addSub(welcomeTopic(instB), 0), + }, + MutateId: 1, + })) + waitForResponses(t, stream, 10*time.Second, "CatchupComplete", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + + // Interleave the kinds so their publishes overlap in time; the two id sequences are + // independent, and each must ascend on its own. + const perGroup = 5 + for i := 0; i < perGroup; i++ { + publishGroup(t, ctx, svc, validationSvc, groupA, fmt.Sprintf("a-live-%d", i)) + populateWelcomeMessages(t, ctx, svc, instA, hpke, 1, fmt.Sprintf("wa-live-%d", i)) + publishGroup(t, ctx, svc, validationSvc, groupB, fmt.Sprintf("b-live-%d", i)) + populateWelcomeMessages(t, ctx, svc, instB, hpke, 1, fmt.Sprintf("wb-live-%d", i)) + } + resps := waitForResponses(t, stream, 10*time.Second, "all live messages", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsTagged(rs, 0)) >= 2*perGroup && + len(welcomeMsgsTagged(rs, 0)) >= 2*perGroup + }) + + live := groupMsgsTagged(resps, 0) + require.Len(t, live, 2*perGroup) + requireAscendingGroupIds(t, live, "live lane across topics") + + liveWelcomes := welcomeMsgsTagged(resps, 0) + require.Len(t, liveWelcomes, 2*perGroup) + requireAscendingWelcomeIds(t, liveWelcomes, "live welcome lane across installations") + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_SeamHoldsLiveUntilCatchupComplete pins the seam: while a wave replays a +// topic, that topic speaks live (mutate_id 0) only after the wave's CatchupComplete — +// messages published mid-wave are folded into the wave (tagged), delivered exactly once — +// while live frames for topics of other subscriptions keep flowing throughout. +func TestSubscribe_SeamHoldsLiveUntilCatchupComplete(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + sentinelId := []byte(test.RandomString(32)) + groupA := []byte(test.RandomString(32)) + mockValidateGroupMessages(validationSvc, groupA) + populateGroupMessages(t, ctx, svc, groupA, 5, "histA") + + // Gate ONLY groupA's wave, so the sentinel's wave and live tail flow freely. + gate := make(chan struct{}) + svc.readOnlyStore = &fakeReadStore{ + ReadMlsStore: svc.readOnlyStore, + groupGate: gate, + gateGroup: groupA, + } + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(sentinelId), 0), + }, + MutateId: 1, + })) + waitForResponses(t, stream, 10*time.Second, "sentinel live", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + + // Wave 9 starts (gated mid-fetch); groupA messages published now are gated behind it, + // while the sentinel keeps receiving live frames. + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + MutateId: 9, + })) + publishGroup(t, ctx, svc, validationSvc, groupA, "a-mid-wave") + publishGroup(t, ctx, svc, validationSvc, sentinelId, "s-live") + resps := waitForResponses(t, stream, 10*time.Second, "sentinel live during the wave", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsWithData(groupMsgsTagged(rs, 0), "s-live")) >= 1 + }) + require.Empty(t, groupMsgsTagged(resps, 9), "wave 9 is still gated") + require.Empty( + t, + groupMsgsWithData(groupMsgsTagged(resps, 0), "a-mid-wave"), + "no live frame for the wave's topic before its CatchupComplete", + ) + + // Release the wave: history and the folded mid-wave message arrive tagged 9, in + // ascending id order, then CatchupComplete; only frames after it speak live for groupA. + close(gate) + resps = waitForResponses(t, stream, 10*time.Second, "wave 9 complete", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 2 }) + replay := groupMsgsTagged(resps, 9) + require.Len(t, replay, 6, "5 history + 1 folded mid-wave message, all tagged 9") + require.Len(t, groupMsgsWithData(replay, "a-mid-wave"), 1, "folded exactly once") + requireAscendingGroupIds(t, replay, "wave 9 replay including the folded tail") + require.Empty(t, groupMsgsWithData(groupMsgsTagged(resps, 0), "a-mid-wave"), + "the folded message must not also be delivered live") + + // Live tail for groupA now flows tagged 0, strictly after the wave's CatchupComplete. + publishGroup(t, ctx, svc, validationSvc, groupA, "a-live") + resps = waitForResponses(t, stream, 10*time.Second, "groupA live tail", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsWithData(groupMsgsTagged(rs, 0), "a-live")) >= 1 + }) + ccIdx := -1 + for i, r := range resps { + if cc := r.GetV1().GetCatchupComplete(); cc != nil && cc.GetMutateId() == 9 { + ccIdx = i + } + } + require.GreaterOrEqual(t, ccIdx, 0) + liveIdx := frameIndex(resps, func(r *mlsv1.SubscribeResponse) bool { + msgs := r.GetV1().GetMessages() + return msgs.GetMutateId() == 0 && + len(groupMsgsWithData(msgs.GetGroupMessages(), "a-live")) > 0 + }) + require.Greater(t, liveIdx, ccIdx, "groupA's live tail begins after CatchupComplete(9)") + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// 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) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, _, cleanup := newTestService(t, ctx) + defer cleanup() + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic([]byte(test.RandomString(32))), 0), + }, + })) + + select { + case err := <-errCh: + require.Equal(t, codes.InvalidArgument, status.Code(err), + "adds with mutate_id 0 must be rejected") + case <-time.After(5 * time.Second): + t.Fatal("a Mutate with adds and mutate_id 0 was not rejected") + } +} + +// TestSubscribe_MutateAddsCapRejected verifies the per-Mutate adds cap (maxMutateAdds): a +// Mutate carrying more adds than the cap fails the stream with ResourceExhausted, and the +// rejection is atomic — it lands before the Mutate's removes or adds take any effect, so +// the offending Mutate produces no frames at all. A fresh stream with exactly the cap +// catches up to CatchupComplete. +func TestSubscribe_MutateAddsCapRejected(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, _, cleanup := newTestService(t, ctx) + defer cleanup() + svc.maxMutateAdds = 3 // tiny: a handful of adds exceeds it + + // Stream 1: subscribe a live topic, then send a Mutate that pairs removes:[liveTopic] + // with 4 adds (one over the cap). + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + liveTopic := groupTopic([]byte(test.RandomString(32))) + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(liveTopic, 0)}, + MutateId: 1, + })) + waitForResponses( + t, + stream, + 5*time.Second, + "wave 1 CatchupComplete", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }, + ) + + overCap := &mlsv1.SubscribeRequest_V1_Mutate{Removes: [][]byte{liveTopic}, MutateId: 2} + for i := 0; i < 4; i++ { + overCap.Adds = append(overCap.Adds, addSub(groupTopic([]byte(test.RandomString(32))), 0)) + } + stream.send(subReqMutate(overCap)) + + select { + case err := <-errCh: + require.Equal(t, codes.ResourceExhausted, status.Code(err), + "a Mutate with more adds than maxMutateAdds must be rejected") + case <-time.After(5 * time.Second): + t.Fatal("over-cap Mutate was not rejected") + } + // Atomicity: the rejection precedes the remove and the adds, so the offending Mutate + // delivered nothing — the stream's frames are exactly wave 1's (its TopicsLive and its + // CatchupComplete), with no CatchupComplete echoing mutate_id 2 and no later TopicsLive. + resps := stream.responses() + require.Equal(t, []uint64{1}, catchupCompletesFrom(resps), + "the rejected Mutate must not produce a CatchupComplete") + topicsLiveFrames := 0 + for _, r := range resps { + if r.GetV1().GetTopicsLive() != nil { + topicsLiveFrames++ + } + } + require.Equal(t, 1, topicsLiveFrames, + "the rejected Mutate must not produce a TopicsLive frame") + + // Stream 2: exactly the cap is allowed and catches up. + stream2 := newFakeSubscribeStream(ctx) + errCh2 := make(chan error, 1) + go func() { errCh2 <- svc.Subscribe(stream2) }() + + atCap := &mlsv1.SubscribeRequest_V1_Mutate{MutateId: 3} + for i := 0; i < 3; i++ { + atCap.Adds = append(atCap.Adds, addSub(groupTopic([]byte(test.RandomString(32))), 0)) + } + stream2.send(subReqMutate(atCap)) + resps2 := waitForResponses( + t, + stream2, + 10*time.Second, + "at-cap wave CatchupComplete", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }, + ) + require.Equal(t, []uint64{3}, catchupCompletesFrom(resps2), + "a Mutate with exactly maxMutateAdds adds must catch up normally") + + stream2.closeSend() + require.NoError(t, <-errCh2) +} + +// TestSubscribe_ResetMidWaveHandsTopicToNewWave pins the seam when a topic moves between +// overlapping waves: a remove+re-add mid-wave hands the topic (gate ownership, pending +// buffer, cursor floor) to the new wave — the old wave replays only its surviving topics, +// and the reset topic's full history plus its discarded mid-wave live message arrive +// exactly once, under the new wave's tag. +func TestSubscribe_ResetMidWaveHandsTopicToNewWave(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + t1 := []byte(test.RandomString(32)) + t2 := []byte(test.RandomString(32)) + sentinelId := []byte(test.RandomString(32)) + for i := 0; i < 5; i++ { + publishGroup(t, ctx, svc, validationSvc, t1, fmt.Sprintf("h1-%d", i)) + publishGroup(t, ctx, svc, validationSvc, t2, fmt.Sprintf("h2-%d", i)) + } + + // Gate ONLY scans covering t1 (both waves below include it), so the sentinel's wave + // and live tail flow freely. + gate := make(chan struct{}) + svc.readOnlyStore = &fakeReadStore{ + ReadMlsStore: svc.readOnlyStore, + groupGate: gate, + gateGroup: t1, + } + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(sentinelId), 0), + }, + MutateId: 11, + })) + waitForResponses(t, stream, 10*time.Second, "sentinel live", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + + // Wave 1 (gated mid-fetch) owns both topics; a live t1 message lands in its pending. + // The dbWorker dispatches strictly in id order, so the sentinel's arrival proves "mid" + // has been routed into the gated buffer. + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(t1), 0), + addSub(groupTopic(t2), 0), + }, + MutateId: 1, + })) + publishGroup(t, ctx, svc, validationSvc, t1, "mid") + publishGroup(t, ctx, svc, validationSvc, sentinelId, "s-live") + waitForResponses(t, stream, 10*time.Second, "sentinel live during wave 1", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsWithData(groupMsgsTagged(rs, 0), "s-live")) >= 1 + }) + + // Reset t1 mid-wave: wave 1 drops it (pending discarded), wave 2 owns it gated. + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Removes: [][]byte{groupTopic(t1)}, + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(t1), 0)}, + MutateId: 2, + })) + + close(gate) + resps := waitForResponses(t, stream, 10*time.Second, "both waves complete", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 3 }) + + // Wave 1 replays only its surviving topic. + wave1 := groupMsgsTagged(resps, 1) + require.Len(t, wave1, 5, "wave 1 delivers only t2's history") + for _, m := range wave1 { + require.Equal(t, string(t2), string(m.GetV1().GetGroupId()), + "no t1 frame may carry wave 1's tag") + } + requireAscendingGroupIds(t, wave1, "wave 1 replay") + + // Wave 2 delivers ALL of t1 — the history plus the reset-discarded "mid" message, which + // its own scan (whose ceiling postdates the publish) picks back up — exactly once. + wave2 := groupMsgsTagged(resps, 2) + require.Len(t, wave2, 6, "wave 2 delivers t1's full history plus the mid-wave message") + for _, m := range wave2 { + require.Equal(t, string(t1), string(m.GetV1().GetGroupId())) + } + require.Len(t, groupMsgsWithData(wave2, "mid"), 1, "the mid-wave message arrives exactly once") + requireAscendingGroupIds(t, wave2, "wave 2 replay") + + // t1 never speaks live before wave 2 completes (it was gated throughout). + for _, m := range groupMsgsTagged(resps, 0) { + require.NotEqual(t, string(t1), string(m.GetV1().GetGroupId()), + "no t1 frame may be tagged live") + } + // No message is delivered under two tags. + seen := make(map[uint64]bool) + for _, tag := range []uint64{0, 1, 2, 11} { + for _, m := range groupMsgsTagged(resps, tag) { + require.False(t, seen[m.GetV1().GetId()], "a message must appear under exactly one tag") + seen[m.GetV1().GetId()] = true + } + } + + // Each wave's TopicsLive announces only what it still owned at completion (the marker + // is sent in the same batch as its CatchupComplete, so it is the frame just before). + cc1 := frameIndex(resps, func(r *mlsv1.SubscribeResponse) bool { + cc := r.GetV1().GetCatchupComplete() + return cc != nil && cc.GetMutateId() == 1 + }) + cc2 := frameIndex(resps, func(r *mlsv1.SubscribeResponse) bool { + cc := r.GetV1().GetCatchupComplete() + return cc != nil && cc.GetMutateId() == 2 + }) + require.Greater(t, cc1, 0) + require.Greater(t, cc2, 0) + require.Equal(t, [][]byte{groupTopic(t2)}, resps[cc1-1].GetV1().GetTopicsLive().GetTopics(), + "wave 1 announces only t2 live") + require.Equal(t, [][]byte{groupTopic(t1)}, resps[cc2-1].GetV1().GetTopicsLive().GetTopics(), + "wave 2 announces only t1 live") + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_RemoveOneTopicMidMultiTopicWave verifies removing one of a wave's topics +// mid-flight: the wave completes with only the survivor — the removed topic's history and +// buffered live messages are dropped — and the removes-only Mutate itself is acked +// immediately, echoing its own mutate_id. +func TestSubscribe_RemoveOneTopicMidMultiTopicWave(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + t1 := []byte(test.RandomString(32)) + t2 := []byte(test.RandomString(32)) + sentinelId := []byte(test.RandomString(32)) + for i := 0; i < 5; i++ { + publishGroup(t, ctx, svc, validationSvc, t1, fmt.Sprintf("g1-%d", i)) + publishGroup(t, ctx, svc, validationSvc, t2, fmt.Sprintf("g2-%d", i)) + } + + gate := make(chan struct{}) + svc.readOnlyStore = &fakeReadStore{ + ReadMlsStore: svc.readOnlyStore, + groupGate: gate, + gateGroup: t1, + } + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(sentinelId), 0), + }, + MutateId: 1, + })) + waitForResponses(t, stream, 10*time.Second, "sentinel live", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + + // Wave 3 (gated mid-fetch) owns both topics; put a live t1 message in its pending + // (sentinel-sequenced), then remove t1 while the wave is still in flight. + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(t1), 0), + addSub(groupTopic(t2), 0), + }, + MutateId: 3, + })) + publishGroup(t, ctx, svc, validationSvc, t1, "g1-mid") + publishGroup(t, ctx, svc, validationSvc, sentinelId, "s-live") + waitForResponses(t, stream, 10*time.Second, "sentinel live during wave 3", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsWithData(groupMsgsTagged(rs, 0), "s-live")) >= 1 + }) + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Removes: [][]byte{groupTopic(t1)}, + MutateId: 4, + })) + + close(gate) + resps := waitForResponses(t, stream, 10*time.Second, "wave 3 complete", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 3 }) + require.Equal(t, []uint64{1, 4, 3}, catchupCompletesFrom(resps), + "the remove is acked immediately; the wave completes exactly once") + + wave := groupMsgsTagged(resps, 3) + require.Len(t, wave, 5, "the wave delivers only the surviving topic's history") + for _, m := range wave { + require.Equal(t, string(t2), string(m.GetV1().GetGroupId())) + } + requireAscendingGroupIds(t, wave, "wave 3 replay") + + // The removed topic must not surface at all — not its history, not its pending. + for _, r := range resps { + for _, m := range r.GetV1().GetMessages().GetGroupMessages() { + require.NotEqual(t, string(t1), string(m.GetV1().GetGroupId()), + "no t1 message may be delivered under any tag") + } + require.False(t, topicsLiveHas(r, groupTopic(t1)), "no marker may announce t1 live") + } + cc3 := frameIndex(resps, func(r *mlsv1.SubscribeResponse) bool { + cc := r.GetV1().GetCatchupComplete() + return cc != nil && cc.GetMutateId() == 3 + }) + require.Greater(t, cc3, 0) + require.Equal(t, [][]byte{groupTopic(t2)}, resps[cc3-1].GetV1().GetTopicsLive().GetTopics(), + "the wave announces only the surviving topic") + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_WelcomeFoldAndMergedOrder pins the welcome side of wave completion: a +// two-installation welcome wave replays merged in ascending welcome id order, live welcomes +// gated mid-wave fold into the wave (tagged, exactly once, in the merged order), and a +// post-completion welcome is live tail. +func TestSubscribe_WelcomeFoldAndMergedOrder(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, _, cleanup := newTestService(t, ctx) + defer cleanup() + + instA := []byte(test.RandomString(32)) + instB := []byte(test.RandomString(32)) + instC := []byte(test.RandomString(32)) // live sentinel: sequences the gated publishes + hpke := []byte(test.RandomString(32)) + // Alternate publishes so the two installations' ids interleave globally. + for i := 0; i < 3; i++ { + populateWelcomeMessages(t, ctx, svc, instA, hpke, 1, fmt.Sprintf("wA-%d", i)) + populateWelcomeMessages(t, ctx, svc, instB, hpke, 1, fmt.Sprintf("wB-%d", i)) + } + + // Gate ONLY scans covering instA (the wave below), so the sentinel's wave flows freely. + // scanParked pins the sequencing: once the wave parks, its ceiling is snapshotted, so + // welcomes published after it MUST reach the client through the pending fold. + gate := make(chan struct{}) + parked := make(chan struct{}) + svc.readOnlyStore = &fakeReadStore{ + ReadMlsStore: svc.readOnlyStore, + welcomeGate: gate, + gateInstallation: instA, + scanParked: parked, + } + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(welcomeTopic(instC), 0), + }, + MutateId: 1, + })) + waitForResponses(t, stream, 10*time.Second, "sentinel live", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(welcomeTopic(instA), 0), + addSub(welcomeTopic(instB), 0), + }, + MutateId: 5, + })) + select { + case <-parked: + case <-time.After(5 * time.Second): + t.Fatal("the welcome wave never reached its gated scan") + } + + // One live welcome per installation, gated into the wave's pending — instB FIRST, so + // its welcome gets the lower id and the fold's wave-topic-order concatenation + // [instA's, instB's] is descending: only the fold's sort restores the merged order. + // The welcome dispatch is strictly id-ordered, so the sentinel's arrival proves both + // were routed. + populateWelcomeMessages(t, ctx, svc, instB, hpke, 1, "wB-live") + populateWelcomeMessages(t, ctx, svc, instA, hpke, 1, "wA-live") + populateWelcomeMessages(t, ctx, svc, instC, hpke, 1, "wC-live") + waitForResponses(t, stream, 10*time.Second, "sentinel welcome during the wave", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(welcomesForKey(welcomeMsgsTagged(rs, 0), instC)) >= 1 + }) + + close(gate) + resps := waitForResponses(t, stream, 10*time.Second, "wave 5 complete", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 2 }) + + wave := welcomeMsgsTagged(resps, 5) + require.Len(t, wave, 8, "6 history + 2 folded live welcomes, all tagged 5") + require.Len(t, welcomesForKey(wave, instA), 4) + require.Len(t, welcomesForKey(wave, instB), 4) + requireAscendingWelcomeIds(t, wave, "welcome wave replay including the folded tail") + + // Before the wave's CatchupComplete, no welcome for its installations may be live. + cc5 := frameIndex(resps, func(r *mlsv1.SubscribeResponse) bool { + cc := r.GetV1().GetCatchupComplete() + return cc != nil && cc.GetMutateId() == 5 + }) + require.Greater(t, cc5, 0) + for _, r := range resps[:cc5] { + if msgs := r.GetV1().GetMessages(); msgs != nil && msgs.GetMutateId() == 0 { + for _, m := range msgs.GetWelcomeMessages() { + key := welcomeInstallationKey(m) + require.NotEqual(t, string(instA), string(key), + "no live frame for a wave installation before its CatchupComplete") + require.NotEqual(t, string(instB), string(key), + "no live frame for a wave installation before its CatchupComplete") + } + } + } + + // A post-completion welcome flows live, tagged 0. + populateWelcomeMessages(t, ctx, svc, instA, hpke, 1, "wA-after") + waitForResponses(t, stream, 10*time.Second, "post-completion live welcome", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(welcomesForKey(welcomeMsgsTagged(rs, 0), instA)) >= 1 + }) + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_FoldSortsAcrossTopics pins the fold's merge order on the group side: live +// messages gated for DIFFERENT topics of one wave land in per-topic pending buffers, and +// completeWave concatenates those buffers in wave-topic order — so when the publish order +// inverts the topic order, only the fold's sort restores the wave's ascending-id replay. +func TestSubscribe_FoldSortsAcrossTopics(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + t1 := []byte(test.RandomString(32)) + t2 := []byte(test.RandomString(32)) + sentinelId := []byte(test.RandomString(32)) + for i := 0; i < 3; i++ { + publishGroup(t, ctx, svc, validationSvc, t1, fmt.Sprintf("h1-%d", i)) + publishGroup(t, ctx, svc, validationSvc, t2, fmt.Sprintf("h2-%d", i)) + } + + // Gate ONLY scans covering t1 (the two-topic wave below), so the sentinel's wave and + // live tail flow freely. scanParked pins the sequencing: once the wave parks, its + // ceiling is snapshotted, so messages published after it MUST reach the client through + // the pending fold. + gate := make(chan struct{}) + parked := make(chan struct{}) + svc.readOnlyStore = &fakeReadStore{ + ReadMlsStore: svc.readOnlyStore, + groupGate: gate, + gateGroup: t1, + scanParked: parked, + } + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(sentinelId), 0), + }, + MutateId: 1, + })) + waitForResponses(t, stream, 10*time.Second, "sentinel live", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + + // Wave 6 owns t1 THEN t2, so completeWave folds pending as [t1's, t2's]. + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(t1), 0), + addSub(groupTopic(t2), 0), + }, + MutateId: 6, + })) + select { + case <-parked: + case <-time.After(5 * time.Second): + t.Fatal("the two-topic wave never reached its gated scan") + } + + // Publish to t2 FIRST, then t1: t2's gated message gets the LOWER id, so the fold's + // wave-topic-order concatenation [t1's, t2's] is descending — only the sort restores + // the replay's order. The dbWorker dispatches strictly in id order, so the sentinel's + // arrival proves both were routed into the gated pending buffers. + publishGroup(t, ctx, svc, validationSvc, t2, "t2-mid") + publishGroup(t, ctx, svc, validationSvc, t1, "t1-mid") + publishGroup(t, ctx, svc, validationSvc, sentinelId, "s-live") + waitForResponses(t, stream, 10*time.Second, "sentinel live during the wave", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsWithData(groupMsgsTagged(rs, 0), "s-live")) >= 1 + }) + + close(gate) + resps := waitForResponses(t, stream, 10*time.Second, "wave 6 complete", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 2 }) + + wave := groupMsgsTagged(resps, 6) + require.Len(t, wave, 8, "6 history + 2 folded live messages, all tagged 6") + require.Len(t, groupMsgsWithData(wave, "t1-mid"), 1, "t1's folded message arrives exactly once") + require.Len(t, groupMsgsWithData(wave, "t2-mid"), 1, "t2's folded message arrives exactly once") + requireAscendingGroupIds(t, wave, "wave 6 replay folded across topics") + require.Empty(t, groupMsgsWithData(groupMsgsTagged(resps, 0), "t1-mid"), + "the folded messages must not also be delivered live") + require.Empty(t, groupMsgsWithData(groupMsgsTagged(resps, 0), "t2-mid"), + "the folded messages must not also be delivered live") + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_WelcomeWaveScanSkipsUnparseableRows pins the welcome fetcher's paging +// contract at the Subscribe level: a row the store cannot parse (an unknown message_type +// from a future writer) still consumes a raw LIMIT slot, so the fetcher must advance its +// cursor from the raw scan position (lastRawID) and terminate on a short RAW page — +// paging by the parsed slice would silently truncate the wave at the first skipped row +// inside a full page. +func TestSubscribe_WelcomeWaveScanSkipsUnparseableRows(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, _, cleanup := newTestService(t, ctx) + defer cleanup() + svc.scanPageLimit = 4 // small pages so the corrupt row sits inside a full page + + inst := []byte(test.RandomString(32)) + hpke := []byte(test.RandomString(32)) + const total = 10 + populateWelcomeMessages(t, ctx, svc, inst, hpke, total, "wel") + + // Learn the real ids, then mark one INSIDE the first full raw page (not the final + // page) as unparseable: it keeps its raw slot but drops from the parsed slice. + all, _, rawCount, err := svc.readOnlyStore.QueryWelcomeMessagesWaveScan( + ctx, + []mlsstore.WelcomeCatchup{{InstallationKey: inst}}, + 0, + math.MaxUint64, + total, + ) + require.NoError(t, err) + require.Equal(t, total, rawCount) + corruptID := welcomeID(all[2]) + svc.readOnlyStore = &fakeReadStore{ + ReadMlsStore: svc.readOnlyStore, + corruptWelcomeIDs: map[uint64]bool{corruptID: true}, + } + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(welcomeTopic(inst), 0)}, + MutateId: 6, + })) + resps := waitForResponses(t, stream, 10*time.Second, "CatchupComplete", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + require.Equal(t, []uint64{6}, catchupCompletesFrom(resps)) + + wave := welcomeMsgsTagged(resps, 6) + require.Len(t, wave, total-1, + "every parseable welcome must arrive; the wave must not truncate at the skipped row") + requireAscendingWelcomeIds(t, wave, "welcome wave replay around the skipped row") + for _, m := range wave { + require.NotEqual(t, corruptID, welcomeID(m), "the unparseable row must not be delivered") + } + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_EveryHistoryPageCarriesWaveTag pins multi-page tag stamping: every +// data-bearing frame before a wave's CatchupComplete carries the wave's mutate_id — no +// page of a paginated replay may leak out tagged live. +func TestSubscribe_EveryHistoryPageCarriesWaveTag(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + svc.scanPageLimit = 50 // small pages so a modest history spans several + + groupA := []byte(test.RandomString(32)) + mockValidateGroupMessages(validationSvc, groupA) + const history = 117 + populateGroupMessages(t, ctx, svc, groupA, history, "hist") + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + MutateId: 1, + })) + resps := waitForResponses(t, stream, 15*time.Second, "full history + CatchupComplete", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsFrom(rs)) >= history && len(catchupCompletesFrom(rs)) >= 1 + }) + require.Equal(t, []uint64{1}, catchupCompletesFrom(resps)) + + ccIdx := frameIndex(resps, func(r *mlsv1.SubscribeResponse) bool { + return r.GetV1().GetCatchupComplete() != nil + }) + dataFrames := 0 + for _, r := range resps[:ccIdx] { + msgs := r.GetV1().GetMessages() + if msgs == nil || + (len(msgs.GetGroupMessages()) == 0 && len(msgs.GetWelcomeMessages()) == 0) { + continue + } + dataFrames++ + require.Equal(t, uint64(1), msgs.GetMutateId(), + "every data-bearing frame before CatchupComplete must carry the wave's tag") + } + require.Greater(t, dataFrames, 2, "the replay must have spanned multiple pages") + + // The live tail after completion is tagged 0. + publishGroup(t, ctx, svc, validationSvc, groupA, "a-live") + resps = waitForResponses(t, stream, 5*time.Second, "live tail", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsWithData(groupMsgsFrom(rs), "a-live")) >= 1 + }) + liveIdx := frameIndex(resps, func(r *mlsv1.SubscribeResponse) bool { + return len(groupMsgsWithData(r.GetV1().GetMessages().GetGroupMessages(), "a-live")) > 0 + }) + require.Equal(t, uint64(0), resps[liveIdx].GetV1().GetMessages().GetMutateId(), + "the live frame must carry the live tag") + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_HistoryOnlyMidReplayPublishNotDelivered pins the history_only seam: a +// message published while a history_only replay is in flight is never delivered — the +// topic has no live registration and the wave's ceiling predates the publish — while live +// flow for other topics is uninterrupted. +func TestSubscribe_HistoryOnlyMidReplayPublishNotDelivered(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + groupA := []byte(test.RandomString(32)) + sentinelId := []byte(test.RandomString(32)) + mockValidateGroupMessages(validationSvc, groupA) + populateGroupMessages(t, ctx, svc, groupA, 5, "histA") + + gate := make(chan struct{}) + parked := make(chan struct{}) + svc.readOnlyStore = &fakeReadStore{ + ReadMlsStore: svc.readOnlyStore, + groupGate: gate, + gateGroup: groupA, + scanParked: parked, + } + + stream := newFakeSubscribeStream(ctx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(sentinelId), 0), + }, + MutateId: 1, + })) + waitForResponses(t, stream, 10*time.Second, "sentinel live", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(groupA), 0), + }, + HistoryOnly: true, + MutateId: 2, + })) + // The wave pins its ceiling before parking on its first page; publish only after, so + // "mid-replay" is deterministically above the ceiling. + select { + case <-parked: + case <-time.After(5 * time.Second): + t.Fatal("the history_only wave never reached its gated scan") + } + publishGroup(t, ctx, svc, validationSvc, groupA, "mid-replay") + publishGroup(t, ctx, svc, validationSvc, sentinelId, "s-live") + // The dbWorker dispatches strictly in id order, so the sentinel's arrival proves the + // dispatcher already handled (and, lacking a live registration, dropped) "mid-replay". + waitForResponses(t, stream, 10*time.Second, "sentinel live during the replay", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsWithData(groupMsgsTagged(rs, 0), "s-live")) >= 1 + }) + + close(gate) + resps := waitForResponses(t, stream, 10*time.Second, "history_only wave complete", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 2 }) + require.Equal(t, []uint64{1, 2}, catchupCompletesFrom(resps)) + + require.Len(t, groupMsgsTagged(resps, 2), 5, "exactly the history, tagged with the wave") + require.Empty(t, groupMsgsWithData(groupMsgsFrom(resps), "mid-replay"), + "a mid-replay publish to a history_only topic must not be delivered under any tag") + + markerIdx := frameIndex(resps, func(r *mlsv1.SubscribeResponse) bool { + return topicsLiveHas(r, groupTopic(groupA)) + }) + cc2 := frameIndex(resps, func(r *mlsv1.SubscribeResponse) bool { + cc := r.GetV1().GetCatchupComplete() + return cc != nil && cc.GetMutateId() == 2 + }) + require.GreaterOrEqual(t, markerIdx, 0, "the history_only wave must announce its marker") + require.Greater(t, cc2, markerIdx, "the marker must precede the wave's CatchupComplete") + require.Len(t, groupMsgsWithData(groupMsgsTagged(resps, 0), "s-live"), 1, + "the sentinel's live flow must be uninterrupted") + + stream.closeSend() + require.NoError(t, <-errCh) +} + +// TestSubscribe_DisconnectMidWaveCleansUp verifies teardown with a gated wave and nonempty +// pending: cancelling the client context returns the handler promptly, and the goroutines +// the stream spawned (sender, frame reader, fetcher) all exit. +func TestSubscribe_DisconnectMidWaveCleansUp(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + svc, _, validationSvc, cleanup := newTestService(t, ctx) + defer cleanup() + + groupA := []byte(test.RandomString(32)) + sentinelId := []byte(test.RandomString(32)) + mockValidateGroupMessages(validationSvc, groupA) + populateGroupMessages(t, ctx, svc, groupA, 5, "hist") + + gate := make(chan struct{}) + svc.readOnlyStore = &fakeReadStore{ + ReadMlsStore: svc.readOnlyStore, + groupGate: gate, + gateGroup: groupA, + } + + before := runtime.NumGoroutine() + + streamCtx, streamCancel := context.WithCancel(ctx) + defer streamCancel() + stream := newFakeSubscribeStream(streamCtx) + errCh := make(chan error, 1) + go func() { errCh <- svc.Subscribe(stream) }() + + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{ + addSub(groupTopic(sentinelId), 0), + }, + MutateId: 1, + })) + waitForResponses(t, stream, 10*time.Second, "sentinel live", + func(rs []*mlsv1.SubscribeResponse) bool { return len(catchupCompletesFrom(rs)) >= 1 }) + + // Wave 2 parks on the gate; a live message lands in its pending (sentinel-sequenced). + stream.send(subReqMutate(&mlsv1.SubscribeRequest_V1_Mutate{ + Adds: []*mlsv1.SubscribeRequest_V1_Mutate_Subscription{addSub(groupTopic(groupA), 0)}, + MutateId: 2, + })) + publishGroup(t, ctx, svc, validationSvc, groupA, "a-mid") + publishGroup(t, ctx, svc, validationSvc, sentinelId, "s-live") + waitForResponses(t, stream, 10*time.Second, "sentinel live during the wave", + func(rs []*mlsv1.SubscribeResponse) bool { + return len(groupMsgsWithData(groupMsgsTagged(rs, 0), "s-live")) >= 1 + }) + + // The client disconnects while the fetcher is parked and pending is nonempty. + streamCancel() + select { + case err := <-errCh: + require.NoError(t, err, "a client cancellation is not a server error") + case <-time.After(2 * time.Second): + t.Fatal("Subscribe did not return after the client disconnected mid-wave") + } + close(gate) // release the parked fetcher only after the handler returned + + // Every stream goroutine must exit. No leak-check helper exists in this repo, so bound + // the goroutine count instead: generous slack, retried for a few seconds. + require.Eventually(t, func() bool { + return runtime.NumGoroutine() <= before+2 + }, 5*time.Second, 50*time.Millisecond, "stream goroutines did not exit after disconnect") +} diff --git a/pkg/mls/store/readStore.go b/pkg/mls/store/readStore.go index d6a6ec07..519bbea1 100644 --- a/pkg/mls/store/readStore.go +++ b/pkg/mls/store/readStore.go @@ -360,33 +360,35 @@ func clampCursor(c uint64) int64 { return int64(c) } -const queryGroupMessagesBatch = ` -SELECT m.id, m.created_at, m.group_id, m.data, m.is_commit, m.sender_hmac, m.should_push -FROM unnest($1::BYTEA[], $2::BIGINT[]) AS f (group_id, id_cursor) -CROSS JOIN LATERAL ( - SELECT g.id, g.created_at, g.group_id, g.data, g.is_commit, g.sender_hmac, g.should_push - FROM group_messages g - WHERE g.group_id = f.group_id AND g.id > f.id_cursor - ORDER BY g.id ASC - LIMIT $3 -) m -ORDER BY m.group_id ASC, m.id ASC` - -// QueryGroupMessagesBatch fetches catch-up messages for many groups in one round-trip -// (XIP-83). Each group resumes from its own cursor and is capped at perGroupLimit rows; -// a group returning fewer than perGroupLimit rows is fully caught up. Ordered by -// (group_id, id). The composite index (group_id, id) makes each per-group seek cheap. +const queryGroupMessagesWaveScan = ` +SELECT g.id, g.created_at, g.group_id, g.data, g.is_commit, g.sender_hmac, g.should_push +FROM group_messages g +JOIN unnest($1::BYTEA[], $2::BIGINT[]) AS f (group_id, id_cursor) + ON g.group_id = f.group_id +WHERE g.id > f.id_cursor AND g.id > $3 AND g.id <= $4 +ORDER BY g.id ASC +LIMIT $5` + +// QueryGroupMessagesWaveScan fetches one page of a catch-up wave's replay: the +// messages of many groups merged into a single ascending-id scan (XIP-83 server +// requirement 4 — a wave's replay is one cursor-ordered pass across its topics, +// not per-topic bursts). Each group's rows start above that group's own cursor; +// the page as a whole starts above scanCursor (the previous page's last id) and +// never exceeds ceiling (the newest id at wave start, so the scan terminates +// under sustained publishing). A page shorter than limit means the wave is fully +// replayed up to the ceiling. // -// Filter keys MUST be unique: unnest preserves duplicates and CROSS JOIN LATERAL runs the -// subquery once per occurrence, so a repeated key returns its rows more than once and -// breaks the caller's per-key pagination count. The Subscribe handler deduplicates adds -// before calling this. -func (s *ReadStore) QueryGroupMessagesBatch( +// Filter keys MUST be unique: unnest preserves duplicates and the join would +// return a repeated group's rows more than once. The Subscribe handler +// deduplicates adds before calling this. +func (s *ReadStore) QueryGroupMessagesWaveScan( ctx context.Context, filters []GroupCatchup, - perGroupLimit int32, + scanCursor uint64, + ceiling uint64, + limit int32, ) ([]*mlsv1.GroupMessage, error) { - if len(filters) == 0 { + if len(filters) == 0 || ceiling == 0 { return nil, nil } groupIds := make([][]byte, len(filters)) @@ -397,10 +399,12 @@ func (s *ReadStore) QueryGroupMessagesBatch( } rows, err := s.db.QueryContext( ctx, - queryGroupMessagesBatch, + queryGroupMessagesWaveScan, pq.Array(groupIds), pq.Array(cursors), - perGroupLimit, + clampCursor(scanCursor), + clampCursor(ceiling), + limit, ) if err != nil { return nil, err @@ -438,26 +442,30 @@ func (s *ReadStore) QueryGroupMessagesBatch( return out, rows.Err() } -const queryWelcomeMessagesBatch = ` -SELECT m.id, m.created_at, m.installation_key, m.data, m.hpke_public_key, m.wrapper_algorithm, m.welcome_metadata, m.message_type -FROM unnest($1::BYTEA[], $2::BIGINT[]) AS f (installation_key, id_cursor) -CROSS JOIN LATERAL ( - SELECT w.id, w.created_at, w.installation_key, w.data, w.hpke_public_key, w.wrapper_algorithm, w.welcome_metadata, w.message_type - FROM welcome_messages w - WHERE w.installation_key = f.installation_key AND w.id > f.id_cursor - ORDER BY w.id ASC - LIMIT $3 -) m -ORDER BY m.installation_key ASC, m.id ASC` - -// QueryWelcomeMessagesBatch is the welcome-topic analogue of QueryGroupMessagesBatch. -func (s *ReadStore) QueryWelcomeMessagesBatch( +const queryWelcomeMessagesWaveScan = ` +SELECT w.id, w.created_at, w.installation_key, w.data, w.hpke_public_key, w.wrapper_algorithm, w.welcome_metadata, w.message_type +FROM welcome_messages w +JOIN unnest($1::BYTEA[], $2::BIGINT[]) AS f (installation_key, id_cursor) + ON w.installation_key = f.installation_key +WHERE w.id > f.id_cursor AND w.id > $3 AND w.id <= $4 +ORDER BY w.id ASC +LIMIT $5` + +// QueryWelcomeMessagesWaveScan is the welcome-topic analogue of QueryGroupMessagesWaveScan. +// A row with an unknown message_type is skipped from the parsed result but still consumed a +// LIMIT slot, so the scan position is reported from the RAW rows: lastRawID is the last row +// the query visited and rawCount how many. Callers MUST advance their cursor from lastRawID +// and treat rawCount < limit as end-of-scan; paging by the parsed slice would silently +// truncate the replay at the first skipped row inside a full page. +func (s *ReadStore) QueryWelcomeMessagesWaveScan( ctx context.Context, filters []WelcomeCatchup, - perGroupLimit int32, -) ([]*mlsv1.WelcomeMessage, error) { - if len(filters) == 0 { - return nil, nil + scanCursor uint64, + ceiling uint64, + limit int32, +) ([]*mlsv1.WelcomeMessage, uint64, int, error) { + if len(filters) == 0 || ceiling == 0 { + return nil, 0, 0, nil } keys := make([][]byte, len(filters)) cursors := make([]int64, len(filters)) @@ -467,17 +475,21 @@ func (s *ReadStore) QueryWelcomeMessagesBatch( } rows, err := s.db.QueryContext( ctx, - queryWelcomeMessagesBatch, + queryWelcomeMessagesWaveScan, pq.Array(keys), pq.Array(cursors), - perGroupLimit, + clampCursor(scanCursor), + clampCursor(ceiling), + limit, ) if err != nil { - return nil, err + return nil, 0, 0, err } defer func() { _ = rows.Close() }() var out []*mlsv1.WelcomeMessage + var lastRawID uint64 + rawCount := 0 for rows.Next() { var ( id int64 @@ -490,8 +502,10 @@ func (s *ReadStore) QueryWelcomeMessagesBatch( messageType int16 ) if err := rows.Scan(&id, &createdAt, &installationKey, &data, &hpkePublicKey, &wrapperAlgorithm, &welcomeMetadata, &messageType); err != nil { - return nil, err + return nil, 0, 0, err } + lastRawID = uint64(id) + rawCount++ switch messageType { case 0: out = append(out, &mlsv1.WelcomeMessage{ @@ -529,7 +543,7 @@ func (s *ReadStore) QueryWelcomeMessagesBatch( s.log.Error("unknown welcome message type", zap.Int16("message_type", messageType)) } } - return out, rows.Err() + return out, lastRawID, rawCount, rows.Err() } func (s *ReadStore) QueryCommitLog( diff --git a/pkg/mls/store/store.go b/pkg/mls/store/store.go index d4fc61c2..510ba552 100644 --- a/pkg/mls/store/store.go +++ b/pkg/mls/store/store.go @@ -56,16 +56,20 @@ type ReadMlsStore interface { ctx context.Context, query *mlsv1.QueryWelcomeMessagesRequest, ) (*mlsv1.QueryWelcomeMessagesResponse, error) - QueryGroupMessagesBatch( + QueryGroupMessagesWaveScan( ctx context.Context, filters []GroupCatchup, - perGroupLimit int32, + scanCursor uint64, + ceiling uint64, + limit int32, ) ([]*mlsv1.GroupMessage, error) - QueryWelcomeMessagesBatch( + QueryWelcomeMessagesWaveScan( ctx context.Context, filters []WelcomeCatchup, - perGroupLimit int32, - ) ([]*mlsv1.WelcomeMessage, error) + scanCursor uint64, + ceiling uint64, + limit int32, + ) ([]*mlsv1.WelcomeMessage, uint64, int, error) QueryCommitLog( ctx context.Context, query *mlsv1.QueryCommitLogRequest, @@ -420,20 +424,24 @@ func (s *Store) QueryWelcomeMessagesV1( return s.readstore.QueryWelcomeMessagesV1(ctx, req) } -func (s *Store) QueryGroupMessagesBatch( +func (s *Store) QueryGroupMessagesWaveScan( ctx context.Context, filters []GroupCatchup, - perGroupLimit int32, + scanCursor uint64, + ceiling uint64, + limit int32, ) ([]*mlsv1.GroupMessage, error) { - return s.readstore.QueryGroupMessagesBatch(ctx, filters, perGroupLimit) + return s.readstore.QueryGroupMessagesWaveScan(ctx, filters, scanCursor, ceiling, limit) } -func (s *Store) QueryWelcomeMessagesBatch( +func (s *Store) QueryWelcomeMessagesWaveScan( ctx context.Context, filters []WelcomeCatchup, - perGroupLimit int32, -) ([]*mlsv1.WelcomeMessage, error) { - return s.readstore.QueryWelcomeMessagesBatch(ctx, filters, perGroupLimit) + scanCursor uint64, + ceiling uint64, + limit int32, +) ([]*mlsv1.WelcomeMessage, uint64, int, error) { + return s.readstore.QueryWelcomeMessagesWaveScan(ctx, filters, scanCursor, ceiling, limit) } func (s *Store) GetNewestGroupMessageMetadata( diff --git a/pkg/mls/store/store_test.go b/pkg/mls/store/store_test.go index 6d6c0c00..acb3a966 100644 --- a/pkg/mls/store/store_test.go +++ b/pkg/mls/store/store_test.go @@ -1269,6 +1269,68 @@ func TestQueryWelcomeMessagesV1_Paginate(t *testing.T) { require.Equal(t, []byte("content8"), resp.Messages[1].GetV1().Data) } +// TestQueryWelcomeMessagesWaveScan_UnknownTypeRowsAdvanceRawCursor pins the wave scan's +// paging contract: a row with an unknown message_type is skipped from the parsed result +// but still consumes a LIMIT slot, so the raw scan position (lastRawID / rawCount) — not +// the parsed slice — must drive the cursor. Paging by parsed rows would end the scan on +// the first short page and silently truncate the replay. +func TestQueryWelcomeMessagesWaveScan_UnknownTypeRowsAdvanceRawCursor(t *testing.T) { + store, cleanup := NewTestStore(t) + defer cleanup() + ctx := context.Background() + + installationKey := []byte("installation") + var ids []int64 + for i := 0; i < 10; i++ { + msg, err := store.InsertWelcomeMessage( + ctx, + installationKey, + []byte(fmt.Sprintf("data%d", i)), + []byte("hpke"), + types.AlgorithmCurve25519, + []byte("metadata"), + ) + require.NoError(t, err) + ids = append(ids, msg.ID) + } + // Corrupt a middle row the way a future writer would: a message_type this reader does + // not understand. It still occupies its id slot in the scan. + _, err := store.db.Exec("UPDATE welcome_messages SET message_type = 99 WHERE id = ?", ids[4]) + require.NoError(t, err) + + ceiling, err := store.queries.GetLatestWelcomeMessageID(ctx) + require.NoError(t, err) + + // Page exactly as the Subscribe welcome fetcher does: advance from the raw scan + // position, terminate on a short raw page. + const limit = 4 + filters := []WelcomeCatchup{{InstallationKey: installationKey, IdCursor: 0}} + var got []*mlsv1.WelcomeMessage + scanCursor := uint64(0) + done := false + for i := 0; i < 10 && !done; i++ { // bound the loop; 3 pages suffice + msgs, lastRawID, rawCount, err := store.QueryWelcomeMessagesWaveScan( + ctx, + filters, + scanCursor, + uint64(ceiling), + limit, + ) + require.NoError(t, err) + if rawCount > 0 { + scanCursor = lastRawID + } + got = append(got, msgs...) + done = rawCount < limit + } + require.True(t, done, "the scan must terminate on a short raw page") + require.Len(t, got, 9, "every parseable row must be returned despite the skipped row") + for i, msg := range got { + require.NotEqual(t, uint64(ids[4]), msg.GetV1().GetId(), + "the unknown-type row must be skipped, not parsed (index %d)", i) + } +} + func TestGetNewestGroupMessage(t *testing.T) { store, cleanup := NewTestStore(t) defer cleanup() diff --git a/pkg/proto/mls/api/v1/mls.pb.go b/pkg/proto/mls/api/v1/mls.pb.go index f538c576..22f11870 100644 --- a/pkg/proto/mls/api/v1/mls.pb.go +++ b/pkg/proto/mls/api/v1/mls.pb.go @@ -2998,17 +2998,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 + 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 messages arrive on no lane of this stream. Combined with + // half-closing the request stream, this is the // bounded catch-up ("sync") mode: the server finishes the wave and then // closes the stream itself. Removals in the Mutate are unaffected. 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 + // (Messages.mutate_id). MUST be nonzero when adds are present (0 is the + // live tag), and MUST NOT match the mutate_id of a wave still in flight + // on the stream (an in-flight collision would make two waves' frames and + // completions indistinguishable) — either violation fails the stream + // with INVALID_ARGUMENT. SHOULD be unique per stream so completed waves + // stay attributable too. MutateId uint64 `protobuf:"varint,4,opt,name=mutate_id,json=mutateId,proto3" json:"mutate_id,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -3257,11 +3263,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_Messages_) isSubscribeResponse_V1_Response() {} @@ -3277,13 +3283,20 @@ func (*SubscribeResponse_V1_TopicsLive_) isSubscribeResponse_V1_Response() {} func (*SubscribeResponse_V1_CatchupComplete_) isSubscribeResponse_V1_Response() {} // A batch of new messages; group and welcome messages share the stream, -// depending on which subscriptions are active. +// depending on which subscriptions are active. A frame belongs to exactly +// one catch-up wave or to live — the server never mixes lanes, or two +// waves, in one frame — and each lane is delivered in ascending id order +// per message kind (live: across all live topics on the stream; a wave: +// across the wave's topics, one merged cursor-ordered pass). type SubscribeResponse_V1_Messages struct { state protoimpl.MessageState `protogen:"open.v1"` GroupMessages []*GroupMessage `protobuf:"bytes,1,rep,name=group_messages,json=groupMessages,proto3" json:"group_messages,omitempty"` WelcomeMessages []*WelcomeMessage `protobuf:"bytes,2,rep,name=welcome_messages,json=welcomeMessages,proto3" json:"welcome_messages,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // The catch-up wave that produced this frame: the Mutate's mutate_id + // for wave replay, 0 for live tail. + MutateId uint64 `protobuf:"varint,3,opt,name=mutate_id,json=mutateId,proto3" json:"mutate_id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *SubscribeResponse_V1_Messages) Reset() { @@ -3330,6 +3343,13 @@ func (x *SubscribeResponse_V1_Messages) GetWelcomeMessages() []*WelcomeMessage { return nil } +func (x *SubscribeResponse_V1_Messages) 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"` @@ -3389,11 +3409,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 } @@ -3436,14 +3460,15 @@ func (x *SubscribeResponse_V1_CatchupComplete) GetMutateId() uint64 { } // Emitted when topics finish catch-up, AFTER the last history frame for -// them — including any live messages that queued up behind the catch-up, -// which were equally historical from the client's perspective — so every -// later frame for a listed topic is live tail. Informational only: delivery +// them — including messages that arrived mid-wave and were folded into it, +// which were equally historical from the client's perspective — so no +// further replay for a listed topic follows; its live (mutate_id 0) frames +// begin after the wave's CatchupComplete. Informational only: delivery // correctness (no duplicates, no gaps) never depends on it. Re-adding a // topic re-runs catch-up and re-emits it; receivers treat it idempotently. 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 } @@ -3685,9 +3710,9 @@ const file_mls_api_v1_mls_proto_rawDesc = "" + "\x05topic\x18\x01 \x01(\fR\x05topic\x12\x1b\n" + "\tid_cursor\x18\x02 \x01(\x04R\bidCursorB\t\n" + "\arequestB\t\n" + - "\aversion\"\xcb\a\n" + + "\aversion\"\xe8\a\n" + "\x11SubscribeResponse\x127\n" + - "\x02v1\x18\x01 \x01(\v2%.xmtp.mls.api.v1.SubscribeResponse.V1H\x00R\x02v1\x1a\xf1\x06\n" + + "\x02v1\x18\x01 \x01(\v2%.xmtp.mls.api.v1.SubscribeResponse.V1H\x00R\x02v1\x1a\x8e\a\n" + "\x02V1\x12L\n" + "\bmessages\x18\x01 \x01(\v2..xmtp.mls.api.v1.SubscribeResponse.V1.MessagesH\x00R\bmessages\x12I\n" + "\astarted\x18\x02 \x01(\v2-.xmtp.mls.api.v1.SubscribeResponse.V1.StartedH\x00R\astarted\x12+\n" + @@ -3695,10 +3720,11 @@ const file_mls_api_v1_mls_proto_rawDesc = "" + "\x04pong\x18\x04 \x01(\v2\x15.xmtp.mls.api.v1.PongH\x00R\x04pong\x12S\n" + "\vtopics_live\x18\x05 \x01(\v20.xmtp.mls.api.v1.SubscribeResponse.V1.TopicsLiveH\x00R\n" + "topicsLive\x12b\n" + - "\x10catchup_complete\x18\x06 \x01(\v25.xmtp.mls.api.v1.SubscribeResponse.V1.CatchupCompleteH\x00R\x0fcatchupComplete\x1a\x9c\x01\n" + + "\x10catchup_complete\x18\x06 \x01(\v25.xmtp.mls.api.v1.SubscribeResponse.V1.CatchupCompleteH\x00R\x0fcatchupComplete\x1a\xb9\x01\n" + "\bMessages\x12D\n" + "\x0egroup_messages\x18\x01 \x03(\v2\x1d.xmtp.mls.api.v1.GroupMessageR\rgroupMessages\x12J\n" + - "\x10welcome_messages\x18\x02 \x03(\v2\x1f.xmtp.mls.api.v1.WelcomeMessageR\x0fwelcomeMessages\x1a\x93\x01\n" + + "\x10welcome_messages\x18\x02 \x03(\v2\x1f.xmtp.mls.api.v1.WelcomeMessageR\x0fwelcomeMessages\x12\x1b\n" + + "\tmutate_id\x18\x03 \x01(\x04R\bmutateId\x1a\x93\x01\n" + "\aStarted\x122\n" + "\x15keepalive_interval_ms\x18\x01 \x01(\rR\x13keepaliveIntervalMs\x12T\n" + "\fcapabilities\x18\x02 \x03(\x0e20.xmtp.mls.api.v1.SubscribeResponse.V1.CapabilityR\fcapabilities\x1a.\n" + diff --git a/pkg/proto/openapi/mls/api/v1/mls.swagger.json b/pkg/proto/openapi/mls/api/v1/mls.swagger.json index 19bb1bb1..4ae0f9a6 100644 --- a/pkg/proto/openapi/mls/api/v1/mls.swagger.json +++ b/pkg/proto/openapi/mls/api/v1/mls.swagger.json @@ -122,7 +122,7 @@ "200": { "description": "A successful response.", "schema": { - "$ref": "#/definitions/xmtpmlsapiv1GetIdentityUpdatesResponse" + "$ref": "#/definitions/v1GetIdentityUpdatesResponse" } }, "default": { @@ -138,7 +138,7 @@ "in": "body", "required": true, "schema": { - "$ref": "#/definitions/xmtpmlsapiv1GetIdentityUpdatesRequest" + "$ref": "#/definitions/v1GetIdentityUpdatesRequest" } } ], @@ -401,13 +401,13 @@ "type": "object", "properties": { "result": { - "$ref": "#/definitions/apiv1GroupMessage" + "$ref": "#/definitions/v1GroupMessage" }, "error": { "$ref": "#/definitions/rpcStatus" } }, - "title": "Stream result of apiv1GroupMessage" + "title": "Stream result of v1GroupMessage" } }, "default": { @@ -510,6 +510,16 @@ } }, "definitions": { + "FetchKeyPackagesResponseKeyPackage": { + "type": "object", + "properties": { + "keyPackageTlsSerialized": { + "type": "string", + "format": "byte" + } + }, + "title": "An individual key package" + }, "GetIdentityUpdatesResponseNewInstallationUpdate": { "type": "object", "properties": { @@ -563,6 +573,21 @@ }, "title": "A wrapper for the updates for a single wallet" }, + "MutateSubscription": { + "type": "object", + "properties": { + "topic": { + "type": "string", + "format": "byte" + }, + "idCursor": { + "type": "string", + "format": "uint64", + "description": "Deliver ids greater than this; 0 = from the beginning. For a newly\njoined group, a client SHOULD seed this from the welcome's encrypted\nWelcomeMetadata.message_cursor so a new membership does not refetch\npre-join history it cannot decrypt; for a new installation's welcome\ntopic, 0 is how pending welcomes are collected." + } + }, + "description": "A topic to subscribe, with the cursor to resume from." + }, "SignatureECDSACompact": { "type": "object", "properties": { @@ -595,14 +620,112 @@ }, "description": "ECDSA signature bytes and the recovery bit\nproduced by xmtp-js::PublicKey.signWithWallet function, i.e.\nEIP-191 signature of a \"Create Identity\" message with the key embedded.\nUsed to sign identity keys." }, - "apiv1GroupMessage": { + "V1Capability": { + "type": "string", + "enum": [ + "CAPABILITY_UNSPECIFIED" + ], + "default": "CAPABILITY_UNSPECIFIED", + "description": "Optional per-stream protocol features (none defined yet; future\nrevisions add values, e.g. fetch-over-stream lookups answered with the\nsame read view that feeds the stream, or new streamable topic kinds)." + }, + "V1CatchupComplete": { "type": "object", "properties": { - "v1": { - "$ref": "#/definitions/v1GroupMessageV1" + "mutateId": { + "type": "string", + "format": "uint64", + "title": "echoes the Mutate; 0 only if a waveless Mutate carried 0" } }, - "title": "Full representation of a group message" + "description": "Sent once per Mutate: at wave completion (after the wave's last\nTopicsLive) for a Mutate that started a catch-up \"wave\", immediately for\none that did not (nothing added — removes-only or empty — or every add\na no-op). Also the catch-up\nseam: live frames (mutate_id 0) for the wave's topics begin only after\nthis frame." + }, + "V1Messages": { + "type": "object", + "properties": { + "groupMessages": { + "type": "array", + "items": { + "type": "object", + "$ref": "#/definitions/v1GroupMessage" + } + }, + "welcomeMessages": { + "type": "array", + "items": { + "type": "object", + "$ref": "#/definitions/v1WelcomeMessage" + } + }, + "mutateId": { + "type": "string", + "format": "uint64", + "description": "The catch-up wave that produced this frame: the Mutate's mutate_id\nfor wave replay, 0 for live tail." + } + }, + "description": "A batch of new messages; group and welcome messages share the stream,\ndepending on which subscriptions are active. A frame belongs to exactly\none catch-up wave or to live — the server never mixes lanes, or two\nwaves, in one frame — and each lane is delivered in ascending id order\nper message kind (live: across all live topics on the stream; a wave:\nacross the wave's topics, one merged cursor-ordered pass)." + }, + "V1Mutate": { + "type": "object", + "properties": { + "adds": { + "type": "array", + "items": { + "type": "object", + "$ref": "#/definitions/MutateSubscription" + }, + "title": "begin delivering these topics" + }, + "removes": { + "type": "array", + "items": { + "type": "string", + "format": "byte" + }, + "title": "stop delivering; clears the topic's cursor floor so a re-add replays" + }, + "historyOnly": { + "type": "boolean", + "description": "Catch this Mutate's adds up — history, TopicsLive markers, and the\nwave's CatchupComplete — but do NOT register them for live delivery.\nThe markers then mean \"you have everything as of the wave's start\";\nlater messages arrive on no lane of this stream. Combined with\nhalf-closing the request stream, this is the\nbounded catch-up (\"sync\") mode: the server finishes the wave and then\ncloses the stream itself. Removals in the Mutate are unaffected." + }, + "mutateId": { + "type": "string", + "format": "uint64", + "description": "Client-chosen correlation id: echoed on this wave's CatchupComplete,\nand stamped on every delivery frame of the wave's catch-up replay\n(Messages.mutate_id). MUST be nonzero when adds are present (0 is the\nlive tag), and MUST NOT match the mutate_id of a wave still in flight\non the stream (an in-flight collision would make two waves' frames and\ncompletions indistinguishable) — either violation fails the stream\nwith INVALID_ARGUMENT. SHOULD be unique per stream so completed waves\nstay attributable too." + } + }, + "description": "Add and/or remove subscriptions in place (applied atomically per frame).\nTopics use the kind-prefixed binary representation shared with the\ndecentralized backend (XIP-49 §3.3.2): the first byte is the topic kind,\nthe remainder is the identifier. This RPC initially serves\nTOPIC_KIND_GROUP_MESSAGES_V1 (0x00, identifier = group_id) and\nTOPIC_KIND_WELCOME_MESSAGES_V1 (0x01, identifier = installation_key);\na topic whose kind the node does not serve fails the stream with\nINVALID_ARGUMENT. Future kinds (key packages, identity updates) are\nadopted via the capabilities advertised on Started." + }, + "V1Started": { + "type": "object", + "properties": { + "keepaliveIntervalMs": { + "type": "integer", + "format": "int64", + "description": "The server's ping cadence (ms): the basis for the client's staleness\nthreshold and the server's reap deadline." + }, + "capabilities": { + "type": "array", + "items": { + "$ref": "#/definitions/V1Capability" + }, + "description": "Optional protocol features the node supports on this stream. The node\nsilently ignores request types it does not understand, so a client\nMUST NOT send an optional request type whose capability the node did\nnot advertise (it would hang waiting on a response that never comes)." + } + }, + "description": "The first frame on every stream." + }, + "V1TopicsLive": { + "type": "object", + "properties": { + "topics": { + "type": "array", + "items": { + "type": "string", + "format": "byte" + }, + "title": "kind-prefixed topics done replaying" + } + }, + "description": "Emitted when topics finish catch-up, AFTER the last history frame for\nthem — including messages that arrived mid-wave and were folded into it,\nwhich were equally historical from the client's perspective — so no\nfurther replay for a listed topic follows; its live (mutate_id 0) frames\nbegin after the wave's CatchupComplete. Informational only: delivery\ncorrectness (no duplicates, no gaps) never depends on it. Re-adding a\ntopic re-runs catch-up and re-emits it; receivers treat it idempotently." }, "associationsRecoverableEd25519Signature": { "type": "object", @@ -657,33 +780,6 @@ "description": "- WELCOME_WRAPPER_ALGORITHM_SYMMETRIC_KEY: Only used for WelcomePointee's", "title": "Describes the algorithm used to encrypt the Welcome Wrapper" }, - "mlsapiv1PagingInfo": { - "type": "object", - "properties": { - "direction": { - "$ref": "#/definitions/mlsapiv1SortDirection" - }, - "limit": { - "type": "integer", - "format": "int64" - }, - "idCursor": { - "type": "string", - "format": "uint64" - } - }, - "title": "Pagination config for queries" - }, - "mlsapiv1SortDirection": { - "type": "string", - "enum": [ - "SORT_DIRECTION_UNSPECIFIED", - "SORT_DIRECTION_ASCENDING", - "SORT_DIRECTION_DESCENDING" - ], - "default": "SORT_DIRECTION_UNSPECIFIED", - "title": "Sort direction for queries" - }, "protobufAny": { "type": "object", "properties": { @@ -769,22 +865,42 @@ "type": "array", "items": { "type": "object", - "$ref": "#/definitions/v1FetchKeyPackagesResponseKeyPackage" + "$ref": "#/definitions/FetchKeyPackagesResponseKeyPackage" }, "description": "Returns one key package per installation in the original order of the\nrequest. If any installations are missing key packages, an empty entry is\nleft in their respective spots in the array." } }, "title": "The response to a FetchKeyPackagesRequest" }, - "v1FetchKeyPackagesResponseKeyPackage": { + "v1GetIdentityUpdatesRequest": { "type": "object", "properties": { - "keyPackageTlsSerialized": { + "accountAddresses": { + "type": "array", + "items": { + "type": "string" + } + }, + "startTimeNs": { "type": "string", - "format": "byte" + "format": "uint64" } }, - "title": "An individual key package" + "title": "Get all updates for an identity since the specified time" + }, + "v1GetIdentityUpdatesResponse": { + "type": "object", + "properties": { + "updates": { + "type": "array", + "items": { + "type": "object", + "$ref": "#/definitions/GetIdentityUpdatesResponseWalletUpdates" + }, + "title": "A list of updates (or empty objects if no changes) in the original order\nof the request" + } + }, + "title": "Used to get any new or revoked installations for a list of wallet addresses" }, "v1GetNewestGroupMessageResponse": { "type": "object", @@ -803,11 +919,20 @@ "type": "object", "properties": { "groupMessage": { - "$ref": "#/definitions/apiv1GroupMessage", + "$ref": "#/definitions/v1GroupMessage", "title": "If no message is found on the topic, will be nil" } } }, + "v1GroupMessage": { + "type": "object", + "properties": { + "v1": { + "$ref": "#/definitions/v1GroupMessageV1" + } + }, + "title": "Full representation of a group message" + }, "v1GroupMessageInput": { "type": "object", "properties": { @@ -879,6 +1004,43 @@ "description": "This would be a serialized MLS key package that the node would\n parse, validate, and then store.", "title": "A wrapper around the Key Package bytes" }, + "v1PagingInfo": { + "type": "object", + "properties": { + "direction": { + "$ref": "#/definitions/v1SortDirection" + }, + "limit": { + "type": "integer", + "format": "int64" + }, + "idCursor": { + "type": "string", + "format": "uint64" + } + }, + "title": "Pagination config for queries" + }, + "v1Ping": { + "type": "object", + "properties": { + "nonce": { + "type": "string", + "format": "uint64" + } + }, + "description": "Liveness challenge/response, shared across versions. Either peer MAY send a\nPing; the receiver MUST reply with a Pong echoing the nonce. The sender closes\nthe stream if no Pong arrives within its deadline — how a node reaps a vanished\npeer (e.g. a mobile client the OS suspended behind a proxy that still ACKs the\ntransport)." + }, + "v1Pong": { + "type": "object", + "properties": { + "nonce": { + "type": "string", + "format": "uint64", + "title": "echoes the nonce of the Ping it answers" + } + } + }, "v1PublishCommitLogRequest": { "type": "object", "properties": { @@ -903,7 +1065,7 @@ "format": "byte" }, "pagingInfo": { - "$ref": "#/definitions/mlsapiv1PagingInfo" + "$ref": "#/definitions/v1PagingInfo" } } }, @@ -922,7 +1084,7 @@ } }, "pagingInfo": { - "$ref": "#/definitions/mlsapiv1PagingInfo" + "$ref": "#/definitions/v1PagingInfo" } } }, @@ -934,7 +1096,7 @@ "format": "byte" }, "pagingInfo": { - "$ref": "#/definitions/mlsapiv1PagingInfo" + "$ref": "#/definitions/v1PagingInfo" } }, "title": "Request for group message queries" @@ -946,11 +1108,11 @@ "type": "array", "items": { "type": "object", - "$ref": "#/definitions/apiv1GroupMessage" + "$ref": "#/definitions/v1GroupMessage" } }, "pagingInfo": { - "$ref": "#/definitions/mlsapiv1PagingInfo" + "$ref": "#/definitions/v1PagingInfo" } }, "title": "Response for group message queries" @@ -963,7 +1125,7 @@ "format": "byte" }, "pagingInfo": { - "$ref": "#/definitions/mlsapiv1PagingInfo" + "$ref": "#/definitions/v1PagingInfo" } }, "title": "Request for welcome message queries" @@ -979,7 +1141,7 @@ } }, "pagingInfo": { - "$ref": "#/definitions/mlsapiv1PagingInfo" + "$ref": "#/definitions/v1PagingInfo" } }, "title": "Response for welcome message queries" @@ -1047,6 +1209,16 @@ }, "title": "Send a batch of welcome messages" }, + "v1SortDirection": { + "type": "string", + "enum": [ + "SORT_DIRECTION_UNSPECIFIED", + "SORT_DIRECTION_ASCENDING", + "SORT_DIRECTION_DESCENDING" + ], + "default": "SORT_DIRECTION_UNSPECIFIED", + "title": "Sort direction for queries" + }, "v1SubscribeGroupMessagesRequest": { "type": "object", "properties": { @@ -1074,6 +1246,59 @@ }, "title": "Subscription filter" }, + "v1SubscribeRequestV1": { + "type": "object", + "properties": { + "mutate": { + "$ref": "#/definitions/V1Mutate" + }, + "ping": { + "$ref": "#/definitions/v1Ping", + "title": "liveness challenge (e.g. probe the link after resuming)" + }, + "pong": { + "$ref": "#/definitions/v1Pong", + "title": "answer to a server Ping" + } + } + }, + "v1SubscribeResponse": { + "type": "object", + "properties": { + "v1": { + "$ref": "#/definitions/v1SubscribeResponseV1" + } + }, + "description": "Server -\u003e client." + }, + "v1SubscribeResponseV1": { + "type": "object", + "properties": { + "messages": { + "$ref": "#/definitions/V1Messages" + }, + "started": { + "$ref": "#/definitions/V1Started", + "title": "sent once, immediately on open, before any catch-up" + }, + "ping": { + "$ref": "#/definitions/v1Ping", + "title": "idle liveness challenge; receiver MUST answer with Pong" + }, + "pong": { + "$ref": "#/definitions/v1Pong", + "title": "answer to a client Ping" + }, + "topicsLive": { + "$ref": "#/definitions/V1TopicsLive", + "title": "no more replay for these topics; live begins after CatchupComplete" + }, + "catchupComplete": { + "$ref": "#/definitions/V1CatchupComplete", + "title": "acks a Mutate; wave completion if it started one" + } + } + }, "v1SubscribeWelcomeMessagesRequest": { "type": "object", "properties": { @@ -1269,36 +1494,6 @@ } }, "description": "Signature represents a generalized public key signature,\ndefined as a union to support cryptographic algorithm agility." - }, - "xmtpmlsapiv1GetIdentityUpdatesRequest": { - "type": "object", - "properties": { - "accountAddresses": { - "type": "array", - "items": { - "type": "string" - } - }, - "startTimeNs": { - "type": "string", - "format": "uint64" - } - }, - "title": "Get all updates for an identity since the specified time" - }, - "xmtpmlsapiv1GetIdentityUpdatesResponse": { - "type": "object", - "properties": { - "updates": { - "type": "array", - "items": { - "type": "object", - "$ref": "#/definitions/GetIdentityUpdatesResponseWalletUpdates" - }, - "title": "A list of updates (or empty objects if no changes) in the original order\nof the request" - } - }, - "title": "Used to get any new or revoked installations for a list of wallet addresses" } } }