feat(api): XIP-83 Subscribe — tag deliveries with wave mutate_id and enforce ordering guarantees

pkg/api/message/subscribe_topics.go

@@ -357,25 +357,6 @@ func advanceTopicCursors(
 	return result
 }
 
-// advanceCursorsFromRows advances per-topic pagination cursors from the RAW query rows (not just the
-// rows that successfully unmarshal). A catch-up paginator that advanced only from unmarshaled
-// envelopes would never move past a row whose envelope bytes fail to parse: since pagination
-// terminates on raw row count, a single bad row in an otherwise-full page would re-fetch that page
-// forever and the wave would never complete. Advancing from raw rows guarantees forward progress.
-func advanceCursorsFromRows(cursors db.TopicCursors, rows []queries.GatewayEnvelopesView) {
-	for i := range rows {
-		vc, ok := cursors[string(rows[i].Topic)]
-		if !ok {
-			continue
-		}
-		origID := uint32(rows[i].OriginatorNodeID)
-		seqID := uint64(rows[i].OriginatorSequenceID)
-		if cur, seen := vc[origID]; !seen || cur < seqID {
-			vc[origID] = seqID
-		}
-	}
-}
-
 func newSubscriptionStatusMessage(
 	status message_api.SubscribeTopicsResponse_SubscriptionStatus,
 ) *message_api.SubscribeTopicsResponse {

pkg/db/sqlc/envelopes_v2.sql

@@ -261,6 +261,58 @@ CROSS JOIN LATERAL (
 ) AS bl
 ORDER BY bl.originator_node_id, bl.originator_sequence_id;
 
+-- name: SelectOriginatorCeilings :many
+-- Newest sequence id per originator: the per-originator replay ceiling a Subscribe
+-- catch-up wave pins at wave start (XIP-83 server requirement 4). One backward probe
+-- of the (originator_node_id, originator_sequence_id) primary key per originator.
+SELECT o.node_id::INT AS originator_node_id,
+       COALESCE((SELECT max(m.originator_sequence_id)
+                 FROM gateway_envelopes_meta m
+                 WHERE m.originator_node_id = o.node_id), 0)::BIGINT AS max_sequence_id
+FROM unnest(@node_ids::INT[]) AS o(node_id);
+
+-- name: SelectGatewayEnvelopesWaveScan :many
+-- One page of a Subscribe catch-up wave's replay: the wave's per-(topic, originator)
+-- cursor floors merged into a single (originator, sequence) keyset scan, bounded per
+-- originator by the ceiling pinned at wave start — so the wave's replay is delivered
+-- in total cursor order per originator across ALL of its topics, not per-topic
+-- bursts, and the scan terminates under sustained publishing (XIP-83 server
+-- requirement 4). The page starts strictly after (scan_node_id, scan_sequence_id)
+-- (the previous page's last row); a page shorter than row_limit means the wave is
+-- fully replayed up to its ceilings.
+--
+-- Cursor keys MUST be unique per (topic, originator): unnest preserves duplicates
+-- and the join would return a repeated pair's rows more than once.
+WITH cursor_entries AS (
+    SELECT t.topic, n.node_id, s.seq_id
+    FROM unnest(@cursor_topics::BYTEA[]) WITH ORDINALITY AS t(topic, ord)
+    JOIN unnest(@cursor_node_ids::INT[]) WITH ORDINALITY AS n(node_id, ord) USING (ord)
+    JOIN unnest(@cursor_sequence_ids::BIGINT[]) WITH ORDINALITY AS s(seq_id, ord) USING (ord)
+),
+ceilings AS (
+    SELECT x.node_id, y.seq_id
+    FROM unnest(@ceiling_node_ids::INT[]) WITH ORDINALITY AS x(node_id, ord)
+    JOIN unnest(@ceiling_sequence_ids::BIGINT[]) WITH ORDINALITY AS y(seq_id, ord) USING (ord)
+)
+SELECT m.originator_node_id,
+       m.originator_sequence_id,
+       m.gateway_time,
+       m.topic,
+       b.originator_envelope
+FROM gateway_envelopes_meta AS m
+JOIN cursor_entries AS ce
+    ON m.topic = ce.topic AND m.originator_node_id = ce.node_id
+JOIN ceilings AS cl
+    ON cl.node_id = m.originator_node_id
+JOIN gateway_envelopes_blob AS b
+    ON b.originator_node_id = m.originator_node_id
+   AND b.originator_sequence_id = m.originator_sequence_id
+WHERE m.originator_sequence_id > ce.seq_id
+  AND m.originator_sequence_id <= cl.seq_id
+  AND (m.originator_node_id, m.originator_sequence_id) > (@scan_node_id::INT, @scan_sequence_id::BIGINT)
+ORDER BY m.originator_node_id, m.originator_sequence_id
+LIMIT @row_limit::INT;
+
 -- name: InsertGatewayEnvelopeBatchV2 :one
 -- Pre-rename batch insert. Calls the v2 stored function which still
 -- references the legacy gateway_envelope_blobs table. No production code

pkg/db/types.go

@@ -163,6 +163,63 @@ func TransformRowsByPerTopicCursors(
 	return result
 }
 
+// SetWaveScanCursors flattens TopicCursors into the parallel floor arrays required by
+// SelectGatewayEnvelopesWaveScan. Each (topic, nodeID, seqID) triple produces one entry
+// in the three arrays.
+func SetWaveScanCursors(
+	q *queries.SelectGatewayEnvelopesWaveScanParams,
+	tc TopicCursors,
+) {
+	total := 0
+	for _, vc := range tc {
+		total += len(vc)
+	}
+
+	q.CursorTopics = make([][]byte, 0, total)
+	q.CursorNodeIds = make([]int32, 0, total)
+	q.CursorSequenceIds = make([]int64, 0, total)
+
+	for topicKey, vc := range tc {
+		topicBytes := []byte(topicKey)
+		for nodeID, seqID := range vc {
+			if nodeID > math.MaxInt32 || seqID > uint64(math.MaxInt64) {
+				continue
+			}
+			q.CursorTopics = append(q.CursorTopics, topicBytes)
+			q.CursorNodeIds = append(q.CursorNodeIds, int32(nodeID))
+			q.CursorSequenceIds = append(q.CursorSequenceIds, int64(seqID))
+		}
+	}
+}
+
+// SetWaveScanCeilings flattens the per-originator ceiling vector into the parallel
+// arrays required by SelectGatewayEnvelopesWaveScan.
+func SetWaveScanCeilings(
+	q *queries.SelectGatewayEnvelopesWaveScanParams,
+	ceilings VectorClock,
+) {
+	q.CeilingNodeIds = make([]int32, 0, len(ceilings))
+	q.CeilingSequenceIds = make([]int64, 0, len(ceilings))
+	for nodeID, seqID := range ceilings {
+		if nodeID > math.MaxInt32 || seqID > uint64(math.MaxInt64) {
+			continue
+		}
+		q.CeilingNodeIds = append(q.CeilingNodeIds, int32(nodeID))
+		q.CeilingSequenceIds = append(q.CeilingSequenceIds, int64(seqID))
+	}
+}
+
+// TransformRowsWaveScan converts wave-scan rows to the common GatewayEnvelopesView type.
+func TransformRowsWaveScan(
+	rows []queries.SelectGatewayEnvelopesWaveScanRow,
+) []queries.GatewayEnvelopesView {
+	result := make([]queries.GatewayEnvelopesView, len(rows))
+	for i, row := range rows {
+		result[i] = queries.GatewayEnvelopesView(row)
+	}
+	return result
+}
+
 // CalculateRowsPerEntry computes the per-(topic, originator) sub-limit
 // for the per-topic cursor query. Returns at least 10 to avoid starving
 // low-volume originators.