feat: XIP-83 bidirectional Subscribe handler (mutable subscriptions + ping/pong liveness)

[tylerhawkes]

Server side of XIP-83: one bidirectional Subscribe stream on the v3 MLS API that replaces repeated server-streaming subscriptions.

Depends on xmtp/proto#337 (regenerated pkg/proto/mls/api/v1 comes from that branch) — draft until it merges.

What's in here

• Single-writer handler (pkg/mls/api/v1/subscribe.go): the select loop is the sole owner of all stream state and the sole caller of stream.Send — no mutexes; catch-up fetchers and the frame reader are pure producers over channels.
• Mutate-in-place: id-based add/remove with per-topic cursors; (*Subscription).Add/Remove are O(1) under the dispatch lock (pkg/subscriptions).
• Batched catch-up: chunked (256) bounded-pool (4) unnest(...) CROSS JOIN LATERAL queries with per-group pagination (pkg/mls/store/readStore.go); 2MB frame splitting; 64MB pending-buffer cap.
• Ordering guarantees: live-gate before dispatcher Add + per-topic high-water mark ⇒ history-before-live, no duplicates, no gaps.
• Live-boundary signals: TopicsLive after each topic's history (including the drained pending buffer); one CATCHUP_COMPLETE per adding Mutate (wave), after the wave's last marker.
• Bounded catch-up: history_only Mutates never register with the dispatcher; client half-close drains in-flight waves then the server closes OK (no pings post-half-close).
• Liveness: idle-triggered Ping (≤30s, resets on traffic), reap on missed Pong (DeadlineExceeded), answers client pings.

Testing

8 handler tests against real Postgres via a hand-written MlsApi_SubscribeServer fake: catch-up-then-live no-dupes, mutate-remove, ping/pong keepalive, reap-on-missed-pong, multi-identity multiplexing, TopicsLive boundary ordering, history-only non-delivery, half-close drain. All race-clean across 3 consecutive -race runs; golangci-lint 0 issues.

🤖 Generated with Claude Code

Note

Add bidirectional MlsApi.Subscribe gRPC handler with mutable subscriptions and ping/pong liveness

• Implements the XIP-83 Subscribe bidirectional streaming RPC in pkg/mls/api/v1/subscribe.go: clients send Mutate requests to add/remove topics in-place, and the server delivers ordered history (catch-up) followed by live messages per topic.
• Uses a single-writer goroutine model: a dedicated sender owns all mutable state, enforces frame size limits (maxFrameBytes), and aborts with ResourceExhausted when the pending buffer exceeds maxPendingBytes.
• Emits TopicsLive markers after catch-up per topic and CatchupComplete frames echoing mutate_id once a full wave completes; supports history_only adds that fetch bounded history without live registration.
• Adds QueryGroupMessagesBatch and QueryWelcomeMessagesBatch to pkg/mls/store/readStore.go for single-round-trip batched catch-up queries with per-topic cursors and limits.
• Extends Subscription in pkg/subscriptions/subscription.go with Add/Remove methods and SubscriptionDispatcher with NewSubscription to support in-place topic mutation under the dispatcher lock.
• Risk: streams are reaped on missed pong (configurable pongDeadline) or send stall; non-reading clients receive DeadlineExceeded rather than backpressure.

^{Macroscope summarized ce3f169.}

[macroscopeapp]

🟡 Medium store/readStore.go:424

The SQL query queryWelcomeMessagesBatch duplicates rows when the same installation_key appears multiple times in filters because unnest preserves duplicates and CROSS JOIN LATERAL executes the subquery once per duplicate. The caller in catchUpWelcomes uses counts[installationKey] == catchUpPerGroupLimit to decide pagination is complete, so duplicate keys can make the count exceed the limit on the first page and incorrectly mark the topic as caught up, causing later welcome messages to be dropped silently.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @pkg/mls/store/readStore.go around line 424:

The SQL query `queryWelcomeMessagesBatch` duplicates rows when the same `installation_key` appears multiple times in `filters` because `unnest` preserves duplicates and `CROSS JOIN LATERAL` executes the subquery once per duplicate. The caller in `catchUpWelcomes` uses `counts[installationKey] == catchUpPerGroupLimit` to decide pagination is complete, so duplicate keys can make the count exceed the limit on the first page and incorrectly mark the topic as caught up, causing later welcome messages to be dropped silently.

[macroscopeapp]

🟡 Medium store/readStore.go:510

QueryWelcomeMessagesBatch drops DB rows with unknown message_type values and returns success. The caller in subscribe.go counts returned proto messages to decide whether more pages exist; when unknown rows are dropped, it receives fewer than perGroupLimit results and incorrectly concludes the installation key is fully caught up. This silently truncates catch-up history after the first batch containing an unknown row.

-		default:
-			// Parity with the live dispatch path (worker.go): surface, don't swallow.
-			s.log.Error("unknown welcome message type", zap.Int16("message_type", messageType))
+		default:
+			return nil, fmt.Errorf("unknown welcome message type: %d", messageType)

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @pkg/mls/store/readStore.go around lines 510-513:

`QueryWelcomeMessagesBatch` drops DB rows with unknown `message_type` values and returns success. The caller in `subscribe.go` counts returned proto messages to decide whether more pages exist; when unknown rows are dropped, it receives fewer than `perGroupLimit` results and incorrectly concludes the installation key is fully caught up. This silently truncates catch-up history after the first batch containing an unknown row.

[macroscopeapp]

Approvability

Verdict: Needs human review

3 blocking correctness issues found. This PR introduces a significant new feature (XIP-83 bidirectional Subscribe handler) with complex streaming and concurrency logic. The author does not own any of the modified files, and there are multiple unresolved medium-severity review comments identifying potential bugs that could cause silent data loss.

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

[macroscopeapp]

🟢 Low v1/subscribe.go:567

lastActivity is only updated inside send at line 172, so inbound client frames like Ping, Pong, or Mutate that don't trigger a response frame never reset the idle timer. On an otherwise idle stream where the client is actively sending keepalive traffic, the server still emits its own Ping after s.pingInterval and can eventually reap the stream with DeadlineExceeded despite active client communication. Consider updating lastActivity whenever any valid client frame is received.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @pkg/mls/api/v1/subscribe.go around line 567:

`lastActivity` is only updated inside `send` at line 172, so inbound client frames like `Ping`, `Pong`, or `Mutate` that don't trigger a response frame never reset the idle timer. On an otherwise idle stream where the client is actively sending keepalive traffic, the server still emits its own `Ping` after `s.pingInterval` and can eventually reap the stream with `DeadlineExceeded` despite active client communication. Consider updating `lastActivity` whenever any valid client frame is received.