feat(api): XIP-83 d14n binding — bidirectional Subscribe on QueryApi (#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)

Author: tylerhawkes

pkg/api/message/listener.go

@@ -11,8 +11,12 @@ import (
 )
 
 type listener struct {
-	ctx         context.Context
-	ch          chan<- []*envelopes.OriginatorEnvelope
+	ctx context.Context
+	ch  chan<- []*envelopes.OriginatorEnvelope
+	// topicsMu guards topics, which is fixed for a server-stream listener but mutated in
+	// place by a mutableSubscription (the XIP-83 bidi Subscribe). The worker reads it when
+	// reaping the listener (closeListener), so concurrent mutation needs the lock.
+	topicsMu    sync.Mutex
 	closed      bool
 	topics      map[string]struct{}
 	originators map[uint32]struct{}
@@ -101,7 +105,7 @@ func (lm *listenersMap[K]) removeListener(keys map[K]struct{}, l *listener) {
 	for key := range keys {
 		value, ok := lm.data.Load(key)
 		if !ok || value == nil {
-			return // Key doesn't exist, nothing to do
+			continue // This key is already gone (e.g. reaped by the worker); keep unregistering the rest
 		}
 		set := value.(*listenerSet)
 		set.removeListener(l)

pkg/api/message/mutable_subscription.go

@@ -0,0 +1,80 @@
+package message
+
+import (
+	"context"
+
+	"github.com/xmtp/xmtpd/pkg/envelopes"
+)
+
+// mutableSubscription is an in-place-mutable topic subscription on the subscribeWorker, used by
+// the XIP-83 bidirectional Subscribe RPC. Unlike subscribeWorker.listen (a fixed query), its
+// topic set is grown and shrunk over the life of the stream via addTopics / removeTopics. It is
+// never global: an empty topic set delivers nothing (a fresh stream that has subscribed to
+// nothing yet), in contrast to newListener, which treats an empty query as "all envelopes".
+//
+// All mutation methods are intended to be called from the single owning goroutine (the Subscribe
+// handler's writer loop). The worker's reap path (closeListener) may also touch the underlying
+// listener's topic set concurrently; both sides take listener.topicsMu, so that is safe.
+type mutableSubscription struct {
+	worker *subscribeWorker
+	l      *listener
+	// ch is the receive end of the listener channel. The worker pushes envelope batches here;
+	// it CLOSES the channel when it reaps the listener (ctx done or the consumer fell behind),
+	// which the writer loop observes as "torn down, reconnect from cursors".
+	ch <-chan []*envelopes.OriginatorEnvelope
+}
+
+// newMutableSubscription registers an empty, non-global topic subscription on the worker and
+// returns a handle the caller mutates over the stream's lifetime. Call close when done.
+func (s *subscribeWorker) newMutableSubscription(ctx context.Context) *mutableSubscription {
+	ch := make(chan []*envelopes.OriginatorEnvelope, subscriptionBufferSize)
+	l := &listener{
+		ctx:         ctx,
+		ch:          ch,
+		topics:      make(map[string]struct{}),
+		originators: make(map[uint32]struct{}),
+		isGlobal:    false, // empty topic set => delivers nothing, NOT all-envelopes
+	}
+	return &mutableSubscription{worker: s, l: l, ch: ch}
+}
+
+// addTopics begins delivering the given topic keys to this subscription. Idempotent: a key
+// already subscribed is a no-op. A no-op once the worker has reaped the listener.
+func (m *mutableSubscription) addTopics(keys map[string]struct{}) {
+	if len(keys) == 0 {
+		return
+	}
+	m.l.topicsMu.Lock()
+	defer m.l.topicsMu.Unlock()
+	if m.l.closed {
+		return
+	}
+	m.worker.topicListeners.addListener(keys, m.l)
+	for k := range keys {
+		m.l.topics[k] = struct{}{}
+	}
+}
+
+// removeTopics stops delivering the given topic keys. Idempotent.
+func (m *mutableSubscription) removeTopics(keys map[string]struct{}) {
+	if len(keys) == 0 {
+		return
+	}
+	m.l.topicsMu.Lock()
+	defer m.l.topicsMu.Unlock()
+	m.worker.topicListeners.removeListener(keys, m.l)
+	for k := range keys {
+		delete(m.l.topics, k)
+	}
+}
+
+// close unregisters the subscription from every topic it still holds. Safe to call even after
+// the worker has already reaped the listener (removeListener is idempotent).
+func (m *mutableSubscription) close() {
+	m.l.topicsMu.Lock()
+	defer m.l.topicsMu.Unlock()
+	if len(m.l.topics) > 0 {
+		m.worker.topicListeners.removeListener(m.l.topics, m.l)
+		m.l.topics = make(map[string]struct{})
+	}
+}