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

[tylerhawkes]

Implements XIP-83 server-tagged deliveries on the d14n QueryApi.Subscribe handler. Spec revision: xmtp/XIPs#139; protos: xmtp/proto#340.

Protocol behavior

• Every Envelopes frame is stamped with the wave that produced it (mutate_id, 0 = live); a frame never mixes lanes or waves.
• Live frames deliver each originator's envelopes in ascending originator_sequence_id, stream-wide; a wave's replay does the same across the wave's topics.
• The seam: no live frame for a wave's topic before that wave's CatchupComplete; live for other topics keeps flowing; envelopes are delivered exactly once across the seam.
• A Mutate with adds must carry a nonzero mutate_id (INVALID_ARGUMENT otherwise); every Mutate is acked by exactly one CatchupComplete.

Implementation

• Per-topic chunked catch-up is replaced by one merged keyset scan (SelectGatewayEnvelopesWaveScan: per-(topic, originator) floors, row-value (originator, sequence) keyset progression, per-originator ceilings from the new SelectOriginatorCeilings pinned at wave start).
• Live arrivals for a wave's topics are gated and folded into the wave at completion (sorted per originator, tagged, deduped via the topic cursor), then TopicsLive, then CatchupComplete, then the gates open.
• The scan position advances from raw rows; page fetches retry with backoff and send only after a fully successful fetch.

Testing

White-box pins for the fold (tag/order/seam frame sequence, scan-vs-pending dedup, ceiling-less new originator, remove-mid-wave pending disposal, DB or send error → no CatchupComplete), plus end-to-end: overlapping waves, per-originator order on both lanes, the seam with a live sentinel topic, reset mid-wave across scan pages, keyset pagination across a mid-originator page boundary, duplicate adds, and the empty-Mutate ack. Internal tests run green under -race.

Depends on xmtp/proto#340 — the vendored generated code here matches that branch and regenerates identically from proto@main once it lands.

🤖 Generated with Claude Code

Note

Tag subscribe delivery frames with wave mutate_id and enforce per-wave ordering guarantees

• Catch-up (replay) frames are now stamped with the wave's non-zero mutate_id; live frames carry mutate_id=0, giving clients a reliable way to distinguish replay from live delivery.
• On wave completion, buffered live envelopes are merged across all topics, sorted by (originator, sequence), and emitted under the wave's mutate_id before TopicsLive and CatchupComplete are sent, enforcing per-originator ordering at the seam.
• Catch-up history is now scanned as a single merged (originator, sequence) ordered stream bounded by per-originator ceilings captured at wave start, replacing per-topic chunking; new SQL queries SelectGatewayEnvelopesWaveScan and SelectOriginatorCeilings support this.
• handleMutate now requires a non-zero mutate_id for adds, rejects reuse of an in-flight mutate_id, and enforces caps on raw add count (maxMutateAdds=100,000) and total cursor entries (maxMutateCursorEntries=1,000,000).
• Oversized single-envelope frames that exceed the gRPC payload limit are skipped with a warning rather than aborting the stream.

^{Macroscope summarized 095b0b1.}

[octane-security-app]

Summary by Octane

New Contracts

No new contracts were added.

Updated Contracts

• subscribe.go: The smart contract now includes wave-stamped envelope frames, wave and live lane separation, and enhanced error handling and retries.
• subscribe_topics.go: Removed the advanceCursorsFromRows function which advanced pagination cursors using raw query rows, ensuring forward progress.
• db.go: Added queries for selectGatewayEnvelopesWaveScan and selectOriginatorCeilings to prepare and close functions in smart contract logic.
• envelopes_v2.sql.go: Added wave scan and originator ceiling functions for structured data querying and processing.
• types.go: The update introduces wave-scan functionalities with functions to flatten topic cursors and ceilings for improved data querying.
• message_api.pb.go: The smart contract changes focus on better topic replay handling, mandatory unique nonzero IDs for mutation, and wave-synced catch-up.

---

🔗 Commit Hash: 574d699

[macroscopeapp]

Approvability

Verdict: Needs human review

This PR implements XIP-83 Subscribe ordering guarantees - a protocol-level feature with significant runtime behavior changes including new wave tagging, merged keyset scanning, and ordering enforcement. The scope and complexity of these changes, plus the fact that all files are owned by xmtp/backend, warrants review by the designated owners.

^{You can customize Macroscope's approvability policy. Learn more.}

[octane-security-app]

Overview

Vulnerabilities found:	10
Severity breakdown:	4 High, 3 Medium, 1 Low, 2 Informational

Warnings found:	1

Detailed findings

pkg/api/message/subscribe.go

• Unbounded T×N floor materialization and per-wave O(N) ceilings in Subscribe catch-up cause node/DB DoS. See more
• Ungated mutable Subscribe waves with PR-introduced merged replay in ReplicationApi cause significant single-node DoS. See more
• Orphaned catch-up waves with added per-wave ceilings and merged scan in QueryApi.Subscribe cause significant redundant DB load and node availability degradation. See more
• Ceilings built only from cached originators in wave-scan catch-up (Subscribe) causes silent omission of persisted envelopes. See more
• Unnegotiated mutate_id enforcement in Subscribe V1 Mutate handling causes client subscription failures and stream aborts. See more
• Wave-start ceiling without capability negotiation in Subscribe history_only catch-up causes clients to miss mid-wave messages. See more
• Whole-wave fold-and-sort in QueryApi.Subscribe catch-up completion causes writer blocking and node-local DoS. See more

pkg/indexer/app_chain/contracts/group_message_storer.go

• Single global (originator,sequence) resume in merged catch-up scan causes omission of valid envelopes across topics. See more

pkg/interceptors/server/rate_limit.go

• Unconditional per-wave all-originator ceiling query and unchunked merged-scan flattening in QueryApi.Subscribe cause node DoS/Crash risk. See more

pkg/proto/xmtpv4/message_api/message_api.pb.go

• Missing live-backlog drain at wave completion in Subscribe causes mid‑wave envelopes to be misattributed or (with half‑close) dropped. See more

Warnings

pkg/api/message/subscribe.go

• mutate_id overhead in bidi Subscribe replay frames can exceed 25 MiB send cap causing catch-up stream abort. See more

---

🔗 Commit Hash: 574d699
🛡️ Octane Dashboard: All vulnerabilities

[tylerhawkes]

Octane scan triage. The scan ran against 574d699, two force-pushes behind; current head is 75f7bc5c4. Per-finding adjudication:

Hardened in response — finding 1 (T×N floor materialization / per-wave ceilings): the adds-count cap (maxMutateAdds) did not bound the cursor entries inside adds — each add's cursor is a per-originator vector, so a single transport-limit-sized Mutate could name ~1.5M (topic, originator) floor pairs, all riding every page query plus the wave's ceilings query. Added maxMutateCursorEntries = 1_000_000 (total entries per Mutate, rejected with ResourceExhausted like the adds cap, never truncated) with a white-box test; XIP-83 requirement 8 updated to name this bound.

Fixed before the scan posted — finding 4 (ceilings built only from cached originators): the current head builds the ceiling set from the originator cache ∪ every cursor-named originator (ceilingOriginators), regression-tested by TestSubscribe_CursorNamedUnknownOriginatorReplayed. The residual (an originator neither cached nor named in any cursor) is documented in-code and matches the pre-existing live-registration semantics.

Bounded by existing guards — findings 2, 3: concurrent waves are capped per stream (maxInflightSubscribeWaves = 256), topics per wave (maxMutateAdds), projected active topics (maxActiveSubscribeTopics), buffered pending bytes (64 MiB), and scan pages (LIMIT 500) with ceilings fetched once per wave. Disconnect cancels the stream context, which propagates into the scan query and every channel emit, so waves cannot outlive their stream ("orphaned waves" do not persist).

By design per XIP-83 — findings 5, 6: the nonzero mutate_id requirement is deliberately not capability-negotiated pre-GA (servers deploy before clients adopt; 0 is the live tag). history_only replay is specified as "as of the wave's start"; mid-wave publishes reach live subscriptions exactly once via the pending fold or the live lane.

Not reproducible — findings 7–10 and the warning:

• 7: the wave-completion fold/sort runs on the stream's own goroutine over its own ≤64 MiB pending buffer; the publish/worker path never blocks on it (non-blocking dispatch with drop-on-backpressure).
• 8: attributed to pkg/indexer/app_chain/contracts/group_message_storer.go, which this PR does not touch. The merged scan's keyset resume matches ORDER BY (originator_node_id, originator_sequence_id) — the table's primary key — exactly, so no row can be skipped across pages; covered by TestSubscribe_WaveScanPaginatesPastPageLimit.
• 9: attributed to rate_limit.go, which this PR does not touch; the substance duplicates finding 1, addressed above.
• 10: attributed to generated code (message_api.pb.go). The wave-completion drain exists: fold → sort → tagged send → TopicsLive → CatchupComplete → gate open, and half-close drains in-flight waves to their acks before close; covered by TestSubscribeSessionFoldTagsAndOrders, TestSubscribe_SeamLiveWaitsForCatchupComplete, and TestSubscribe_HalfCloseHistoryOnlyDrains.
• Warning (25 MiB cap): mutate_id is a single uint64 per frame (≤11 wire bytes), and frames split at 2 MiB (maxSubscribeFrameBytes), far below the 25 MiB send cap — the tag cannot push a frame over the limit.