xmtp_proto: XIP-83 d14n binding — QueryApi.Subscribe (bidi mutable subscription)#338
Merged
Merged
Conversation
…bscription; vector-cursor topics; OriginatorEnvelope delivery; mutate_id waves; Started/CatchupComplete/TopicsLive; ping/pong; history_only) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
Contributor
ApprovabilityVerdict: Approved Purely additive protobuf schema changes defining a new bidirectional subscription API. No existing definitions modified, and the author is the designated code owner for both files. You can customize Macroscope's approvability policy. Learn more. |
tylerhawkes
added a commit
to xmtp/xmtpd
that referenced
this pull request
Jun 26, 2026
…2020) ## XIP-83 d14n binding — bidirectional `QueryApi.Subscribe` Implements the decentralized (d14n) binding of [XIP-83 mutable subscription streams](xmtp/XIPs#139) on xmtpd. A single long-lived bidirectional stream lets a client **mutate its topic set in place** (add/remove subscriptions) and keep the connection alive with ping/pong — instead of tearing down and reopening a stream every time group membership changes. This is the xmtpd counterpart to the node-go v3 `MlsApi.Subscribe` binding. Same control protocol (`Mutate` / `Started` / `CatchupComplete` / `TopicsLive` / `Ping` / `Pong`); the d14n binding delivers `OriginatorEnvelope`s and uses per-originator vector cursors. ### Design: gated async catch-up - **Single-writer + single-sender actor** per subscription. All topic-set mutation happens on the writer goroutine; all stream sends happen on the sender goroutine (drained via a bounded queue). - **Per-topic gate + pending buffer.** A newly-added topic is caught up by an async fetcher goroutine while already-live topics keep flowing — a slow catch-up never head-of-line-blocks live delivery. - **No missing messages across the switch.** A topic is registered *live* **before** its catch-up snapshot is taken, and delivery is deduped by a monotonic per-originator cursor (`advanceTopicCursors`). Anything that lands during catch-up is either in the snapshot or arrives live; the dedup drops the overlap. - **Sparse vs. filled cursors.** The writer keeps only the originator cursors the client provided (growing them as new originators are seen); the fetcher holds a filled copy purely for the query. Keeps memory bounded at the active-topic ceiling. - **1M active-topic ceiling** (`maxActiveSubscribeTopics`) — ~16–32MB for a consumer that large, deliberately favored over forcing clients into multiple connections (which would invite rate-limiting and wreck some topologies). - **Liveness.** Either peer may `Ping`; the receiver must `Pong` the nonce. No pong within the deadline → the node reaps the vanished peer (e.g. a mobile client the OS suspended behind a proxy that still ACKs the transport). ### Commits 1. `feat(api): add mutableSubscription handle to subscribeWorker` — in-place topic add/remove on the existing subscribe worker, guarded by `topicsMu`. 2. `feat(api): XIP-83 bidirectional Subscribe handler on QueryApi + xmtpv4 protos` — the handler (`subscribe.go`) + regenerated xmtpv4 protos. 3. `test(api): Subscribe handler tests + configurable keepalive interval for tests` — 5 race-tested scenarios + a test-only knob to shorten the keepalive cadence. ### Tests `go test -race ./pkg/api/message` — all green: - `TestSubscribe_CatchUpThenLive` - `TestSubscribe_MutateRemoveStopsDelivery` - `TestSubscribe_HalfCloseHistoryOnlyDrains` - `TestSubscribe_HistoryOnlyOnLiveRejected` - `TestSubscribe_NoPongIsReaped` ### Dependency & merge ordering Built on [xmtp/proto#338](xmtp/proto#338) (`QueryApi.Subscribe`). The committed `.pb.go` here is generated from that branch via `dev/gen/protos`; once #338 merges, regenerating from proto `main` produces byte-identical output. 🤖 Generated with [Claude Code](https://claude.com/claude-code)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds the decentralized-backend binding of XIP-83 — companion to the v3
MlsApi.Subscribebinding (#337, merged). A new bidirectionalSubscribe(stream SubscribeRequest) returns (stream SubscribeResponse)RPC onQueryApi, additive alongside the existing server-streamingSubscribeTopics.The control protocol is identical to the v3 binding —
Mutate(add/remove topics in place,history_only,mutate_id),Started,CatchupComplete,TopicsLive,Ping/Pong— adapted to the decentralized data model:Subscriptionresumes from axmtp.xmtpv4.envelopes.Cursor(the same typeSubscribeTopicsuses), not v3's scalarid_cursor.OriginatorEnvelopes; the client demultiplexes by topic.Bidirectional streaming requires HTTP/2; browser/connect-web clients stay on
SubscribeTopics.Spec: xmtp/XIPs#139 (the "Decentralized (d14n) binding" section).
🤖 Generated with Claude Code
Note
Add bidirectional
QueryApi.SubscribeRPC with mutable topic subscriptions and ping/pong livenessSubscribeRequestandSubscribeResponsemessages to message_api.proto defining the client→node and node→client framing for the new bidi subscription protocol (XIP-83 d14n binding).SubscribeRequest.V1.Mutatesupports atomic topic adds/removes, ahistory_onlycatch-up flag, and amutate_idfor correlatingCatchupCompleteresponses.SubscribeResponse.V1carries batchedEnvelopes, stream init metadata (Started), topic live-boundary markers (TopicsLive), per-mutation completion signals (CatchupComplete), andPing/Pongliveness frames.Subscribebidi-streaming RPC on theQueryApiservice in query_api.proto, alongside the existing server-streamingSubscribeTopics.Macroscope summarized 6ccd578.