feat: Plan 7 PR A — long-lived SSE edge sync

[messagesgoel-blip]

Summary

• Long-lived GET /v1/sync/events SSE: heartbeats, poll loop, bootstrap cursor, bounded write queue (frames + bytes) with event: shutdown control lane
• Snapshot-consistent backlog probe → 429 + X-Verilink-Sync-Reason: backlog-cap (not 410); strict raw-string Last-Event-ID parsing
• Snapshot-route gzip only; SSE drain before server close on SIGTERM/SIGINT
• Unit tests for parseLastEventId, write queue, registry

Test plan

• cd control-plane && npm run test:unit
• cd control-plane && npx tsc --noEmit
• Manual: connect SSE, verify : ping and cursor events
• PR B follow-up: integration suite + sync_cursors + lag endpoint

@coderabbitai review

Summary by CodeRabbit

• New Features

  • Added real-time event streaming with Server-Sent Events, including heartbeats, cursor tracking, reconnection support, and tenant-scoped updates.
  • Added connection limits, backlog protection, slow-client handling, and graceful stream shutdown.
  • Added gzip compression for snapshot downloads.
  • Added clearer error responses for oversized requests and temporarily unavailable services.

• Bug Fixes

  • Improved validation and handling of event-replay cursors, including very large values.

• Tests

  • Added coverage for cursor parsing, streaming queues, connection management, and shutdown behavior.

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[cursor]

Bugbot is not enabled for your account, so this pull request was not reviewed.

Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs.

[coderabbitai]

Warning

Review limit reached

You’ve reached a temporary PR review limit under our Fair Usage Limits Policy.

Your recent review volume is higher than typical usage, so adaptive limits are currently applied.

Next review available in: 47 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: 78fe28e0-af39-45b2-a073-19ed5578af30

📥 Commits

Reviewing files that changed from the base of the PR and between 3bda00f and 08d6b0a.

📒 Files selected for processing (7)

• control-plane/src/domains/sync/parseLastEventId.test.ts
• control-plane/src/domains/sync/parseLastEventId.ts
• control-plane/src/domains/sync/sseSession.ts
• control-plane/src/domains/sync/sseWriteQueue.test.ts
• control-plane/src/domains/sync/sseWriteQueue.ts
• control-plane/src/domains/sync/syncRepository.ts
• control-plane/src/index.ts

Walkthrough

Changes

SSE edge sync

Layer / File(s)	Summary
Sync cursors and transactional batches control-plane/src/domains/sync/parseLastEventId.ts, control-plane/src/domains/sync/syncRepository.ts, control-plane/src/domains/sync/syncService.ts	Adds BigInt-safe cursor parsing and transactional initial/poll batch queries with high-water marks and backlog detection.
SSE queue and connection lifecycle control-plane/src/domains/sync/sseWriteQueue.ts, control-plane/src/domains/sync/sseRegistry.ts, control-plane/src/domains/sync/*.test.ts	Adds bounded output queues, backpressure handling, control frames, connection limits, shutdown draining, and focused tests.
SSE session and HTTP integration control-plane/src/domains/sync/sseSession.ts, control-plane/src/routes/sync.ts, control-plane/src/config.ts, control-plane/src/shared/errors/AppError.ts, control-plane/package.json	Adds long-lived SSE sessions with polling and heartbeats, snapshot compression, structured errors, and configurable stream limits.
Graceful shutdown coordination control-plane/src/index.ts, docs/superpowers/plans/HANDOVER.md	Drains registered SSE connections during shutdown and updates the handover status for the SSE implementation.

Estimated code review effort: 4 (Complex) | ~60 minutes

Sequence Diagram(s)

sequenceDiagram
  participant Client
  participant SyncRoute
  participant SseRegistry
  participant syncRepository
  participant runSseSession
  participant SseWriteQueue
  Client->>SyncRoute: request /events with cursor
  SyncRoute->>SseRegistry: check admission
  SyncRoute->>syncRepository: load initial batch and high-water
  SyncRoute->>runSseSession: start SSE session
  runSseSession->>SseWriteQueue: enqueue initial events and cursor
  runSseSession->>syncRepository: poll for new events
  syncRepository-->>runSseSession: return events and high-water
  runSseSession->>SseWriteQueue: enqueue events and heartbeat frames
  SseWriteQueue-->>Client: stream SSE frames

Loading

Possibly related PRs

• Numeracode/verilink#4: Provides the control-plane sync foundation extended by this SSE implementation.
• Numeracode/verilink#12: Modifies overlapping sync repository cursor and query behavior.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 16.67% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly reflects the primary change: implementing long-lived SSE edge sync for Plan 7 PR A.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches 💡 1

📝 Generate docstrings 💡

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/sse-edge-sync

---

_{Comment @coderabbitai help to get the list of available commands.}

[qodo-code-review]

Code Review by Qodo

🐞 Bugs (0) 📘 Rule violations (0) 📜 Skill insights (0)

Context used

✅ Compliance rules (platform): 18 rules

1. Abort listener leak ✓ Resolved 🐞 Bug ☼ Reliability

Description

sleep() in runSseSession adds an abort listener each poll iteration but never removes it when
the timeout resolves normally, so long-lived SSE connections accumulate listeners and leak memory
over time.

Code

control-plane/src/domains/sync/sseSession.ts[R14-26]

+function sleep(ms: number, signal: AbortSignal): Promise<void> {
+  return new Promise((resolve, reject) => {
+    if (signal.aborted) {
+      reject(new Error('aborted'));
+      return;
+    }
+    const t = setTimeout(resolve, ms);
+    const onAbort = () => {
+      clearTimeout(t);
+      reject(new Error('aborted'));
+    };
+    signal.addEventListener('abort', onAbort, { once: true });
+  });

Relevance

●●● Strong

Very similar listener-accumulation bug was previously accepted and fixed (losing listeners not
cleaned up).

PR-#12

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

sleep() installs an abort listener each time it’s called, but only clears the timeout on abort;
there is no removal on successful timeout. The poll loop calls sleep() continuously, so listeners
accumulate for long-lived sessions.

control-plane/src/domains/sync/sseSession.ts[14-26]
control-plane/src/domains/sync/sseSession.ts[160-167]
PR-#12

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
`control-plane/src/domains/sync/sseSession.ts::sleep()` registers `signal.addEventListener('abort', ...)` but does not remove that listener when the timer resolves successfully. Because `sleep()` is called on every poll cycle, the session’s `AbortSignal` accumulates listeners for the lifetime of the connection.

## Issue Context
This is a long-lived SSE loop (`while (!abort.signal.aborted)`) with a short poll interval, so leaks compound quickly.

## Fix Focus Areas
- control-plane/src/domains/sync/sseSession.ts[14-26]
- control-plane/src/domains/sync/sseSession.ts[160-167]

## Suggested fix
Refactor `sleep()` to use a shared `cleanup()` that always removes the abort listener.
- On timeout: call `cleanup()` then `resolve()`.
- On abort: clear the timeout, call `cleanup()`, then `reject(...)`.
(Keeping `{ once: true }` is fine, but still remove the listener on the non-abort path.)

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Backpressure bypasses queue ✓ Resolved 🐞 Bug ☼ Reliability

Description

SseWriteQueue continues to start new pumps after res.write() returns false because pump()
does not guard on waitingForDrain, and it emits space/reduces dataBytes even while the
underlying socket buffer is still full, defeating the bounded-queue design and risking unhandled
res.write() errors on closed responses.

Code

control-plane/src/domains/sync/sseWriteQueue.ts[R59-191]

+  /** Enqueue a data frame (durable event, cursor, or ping). */
+  enqueueData(frame: string, cursor?: bigint): EnqueueResult {
+    if (this.closed) return 'closed';
+    const size = Buffer.byteLength(frame);
+    if (size > this.opts.maxFrameBytes) return 'oversized';
+    if (
+      this.data.length >= this.opts.maxFrames ||
+      this.dataBytes + size > this.opts.maxBytes
+    ) {
+      return 'full';
+    }
+    this.data.push({ frame, cursor });
+    this.dataBytes += size;
+    queueMicrotask(() => void this.pump());
+    return 'ok';
+  }
+
+  /** Control lane — always accepted (overwrites prior pending shutdown). */
+  enqueueControl(frame: string): EnqueueResult {
+    if (this.closed) return 'closed';
+    this.control = frame;
+    queueMicrotask(() => void this.pump());
+    return 'ok';
+  }
+
+  /** Wait until there is capacity for a frame of `size` bytes, or closed. */
+  async waitForSpace(size: number, signal?: AbortSignal): Promise<'ok' | 'closed'> {
+    for (;;) {
+      if (this.closed) return 'closed';
+      if (
+        this.data.length < this.opts.maxFrames &&
+        this.dataBytes + size <= this.opts.maxBytes
+      ) {
+        return 'ok';
+      }
+      try {
+        await new Promise<void>((resolve, reject) => {
+          const onSpace = () => {
+            cleanup();
+            resolve();
+          };
+          const onClose = () => {
+            cleanup();
+            resolve();
+          };
+          const onAbort = () => {
+            cleanup();
+            reject(new Error('aborted'));
+          };
+          const cleanup = () => {
+            this.off('space', onSpace);
+            this.off('closed', onClose);
+            signal?.removeEventListener('abort', onAbort);
+          };
+          this.on('space', onSpace);
+          this.on('closed', onClose);
+          signal?.addEventListener('abort', onAbort, { once: true });
+        });
+      } catch {
+        return 'closed';
+      }
+    }
+  }
+
+  close(): void {
+    if (this.closed) return;
+    this.closed = true;
+    this.clearSlowTimer();
+    this.emit('closed');
+    this.emit('space');
+  }
+
+  private clearSlowTimer(): void {
+    if (this.slowTimer) {
+      clearTimeout(this.slowTimer);
+      this.slowTimer = null;
+    }
+  }
+
+  private armSlowTimer(): void {
+    if (this.slowTimer || this.closed) return;
+    this.slowTimer = setTimeout(() => {
+      this.slowTimer = null;
+      this.opts.onSlowClient?.();
+      this.close();
+      try {
+        this.res.end();
+      } catch {
+        /* ignore */
+      }
+    }, this.opts.slowClientMs);
+  }
+
+  private async pump(): Promise<void> {
+    if (this.pumping || this.closed) return;
+    this.pumping = true;
+    try {
+      while (!this.closed) {
+        if (this.control !== null) {
+          const next = this.control;
+          this.control = null;
+          const ok = this.res.write(next);
+          this.emit('space');
+          if (!ok) {
+            this.waitingForDrain = true;
+            this.armSlowTimer();
+            break;
+          }
+          continue;
+        }
+        const item = this.data[0];
+        if (!item) break;
+        this.data.shift();
+        this.dataBytes -= Buffer.byteLength(item.frame);
+        const ok = this.res.write(item.frame);
+        if (item.cursor !== undefined) {
+          if (this.lastWrittenCursor === null || item.cursor > this.lastWrittenCursor) {
+            this.lastWrittenCursor = item.cursor;
+          }
+        }
+        this.emit('space');
+        if (!ok) {
+          this.waitingForDrain = true;
+          this.armSlowTimer();
+          break;
+        }
+      }
+    } finally {
+      this.pumping = false;
+      if (!this.waitingForDrain && (this.control || this.data.length > 0)) {
+        void this.pump();
+      }
+    }

Relevance

●●● Strong

Backpressure correctness is a stated SSE safety rail; queue must pause pumps until drain to preserve
bounds.

PR-#14

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

After backpressure (res.write returns false), waitingForDrain is set, but enqueues still
schedule new pump runs via microtasks. Since pump() doesn’t check waitingForDrain, later pumps
can keep writing before drain, and dataBytes is decremented before the write while space is
emitted regardless, letting producers enqueue more than intended.

control-plane/src/domains/sync/sseWriteQueue.ts[59-82]
control-plane/src/domains/sync/sseWriteQueue.ts[39-45]
control-plane/src/domains/sync/sseWriteQueue.ts[152-191]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
`SseWriteQueue` aims to be bounded (frames + bytes) and respect backpressure. Today:
- When `res.write(...)` returns `false`, `waitingForDrain` is set, but future `enqueueData/enqueueControl` still schedule `pump()`.
- `pump()` does not early-return when `waitingForDrain` is true, so a *later* scheduled pump can write additional frames before a `drain` event, pushing data into Node’s internal buffers beyond the queue’s byte budget.
- `pump()` also calls `res.write(...)` without try/catch, so a closed/destroyed response can turn into an unhandled rejected promise (because callers use `void this.pump()`).

## Issue Context
This code is used on long-lived SSE connections where slow clients/backpressure are expected.

## Fix Focus Areas
- control-plane/src/domains/sync/sseWriteQueue.ts[59-82]
- control-plane/src/domains/sync/sseWriteQueue.ts[39-45]
- control-plane/src/domains/sync/sseWriteQueue.ts[152-191]

## Suggested fix
1) Prevent pumping while waiting for drain:
- Add `if (this.waitingForDrain) return;` near the top of `pump()` (in addition to `pumping/closed`), and/or avoid scheduling `pump()` from `enqueue*` while `waitingForDrain` is true.

2) Preserve the bounded-memory invariant under backpressure:
- Do not `emit('space')` in the backpressured write path (only emit on `drain` / when real capacity returns).
- Track bytes written-but-not-drained (e.g., `inflightBytes`) so `waitForSpace()`/`enqueueData()` do not treat those bytes as free until `drain` fires.

3) Harden writes:
- Wrap `res.write(...)` in try/catch; on error, `close()` the queue and stop pumping.
- Optionally listen to `res.on('close'|'error')` to `close()` the queue immediately.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

3. Last-Event-ID DB overflow ✓ Resolved 🐞 Bug ≡ Correctness

Description

parseLastEventId accepts arbitrarily large digit strings as bigint, but
sync_events.sync_version is BIGINT; values outside Postgres’ 64-bit range will cause the initial
batch query to throw and the /v1/sync/events endpoint to return a 500 instead of a 400.

Code

control-plane/src/domains/sync/parseLastEventId.ts[R24-36]

+function parseNonNegativeInteger(raw: string): ParseLastEventIdResult {
+  const trimmed = raw.trim();
+  if (trimmed.length === 0) {
+    return { ok: false, message: 'last_event_id must not be blank' };
+  }
+  if (!/^\d+$/.test(trimmed)) {
+    return { ok: false, message: 'last_event_id must be a non-negative integer' };
+  }
+  try {
+    return { ok: true, value: BigInt(trimmed) };
+  } catch {
+    return { ok: false, message: 'last_event_id is out of range' };
+  }

Relevance

●●● Strong

Plan 7 explicitly wants strict Last-Event-ID validation; 64-bit BIGINT bounds prevent 500s.

PR-#14

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The schema defines sync_version as BIGINT. The SSE route parses Last-Event-ID into an unbounded
JS bigint and passes its string form as the query parameter for sync_version > $1, which will
fail for values larger than Postgres BIGINT.

control-plane/src/domains/sync/parseLastEventId.ts[24-36]
control-plane/src/routes/sync.ts[64-85]
control-plane/src/domains/sync/syncRepository.ts[94-103]
control-plane/migrations/003_sync/migration.sql[4-6]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
`parseLastEventId()` currently treats any non-negative integer string as valid and converts it to JS `bigint`. JS `bigint` is unbounded, but `sync_events.sync_version` is PostgreSQL `BIGINT`. When an oversized cursor is supplied, Postgres will error during the `sync_version > $1` comparison, producing a server error instead of a client error.

## Issue Context
`/v1/sync/events` feeds `parsed.value` into `getInitialBatchWithHighWater()` which binds `sinceVersion.toString()` against `sync_version`.

## Fix Focus Areas
- control-plane/src/domains/sync/parseLastEventId.ts[24-36]
- control-plane/src/routes/sync.ts[64-85]
- control-plane/src/domains/sync/syncRepository.ts[94-105]
- control-plane/migrations/003_sync/migration.sql[4-6]

## Suggested fix
After parsing the `bigint`, enforce `0n <= value <= 9223372036854775807n` (Postgres signed BIGINT max) and return `{ ok: false, message: 'last_event_id is out of range' }` when exceeded. This keeps the existing BAD_REQUEST path in the route and avoids DB exceptions.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

_{To customize comments, go to the Qodo configuration screen, or learn more in the docs.}

[qodo-code-review]

PR Summary by Qodo

feat: long-lived SSE sync stream with backpressure, backlog cap, graceful drain

✨ Enhancement 🧪 Tests 🕐 40+ Minutes

AI Description

• Rewrites GET /v1/sync/events from a one-shot dump into a poll-driven, long-lived SSE session
 with heartbeats, bootstrap cursor, and shutdown control lane.
• Adds a bounded per-connection write queue (frame count + byte budget) with a dedicated control
 lane so event: shutdown is never dropped.
• Returns 429 + X-Verilink-Sync-Reason: backlog-cap for backlog overflow and strictly parses
 raw-string Last-Event-ID using BigInt.
• Applies gzip compression to /snapshot only and drains SSE connections during SIGTERM/SIGINT
 shutdown.
• Adds unit tests for cursor parsing, write queue, and registry shutdown behavior.

Diagram

sequenceDiagram
  actor Edge as Edge Client
  participant Route as "routes/sync.ts"
  participant Session as SseSession
  participant Queue as SseWriteQueue
  participant Registry as SseRegistry
  participant DB as "Postgres (sync_events)"
  Edge->>Route: GET /v1/sync/events (Last-Event-ID)
  Route->>DB: getInitialBatchWithHighWater()
  DB-->>Route: events + highWater or backlogExceeded
  Route-->>Edge: 429 backlog-cap (if exceeded)
  Route->>Session: runSseSession()
  Session->>Registry: tryRegister(queue, abort, cleanup)
  loop poll interval
    Session->>DB: getPollBatchWithHighWater()
    Session->>Queue: enqueueData(frame, cursor)
    Queue-->>Edge: res.write() (backpressure-aware)
  end
  Registry->>Queue: enqueueControl(shutdown) on SIGTERM
  Queue-->>Edge: event: shutdown

Loading

High-Level Assessment

The following are alternative approaches to this PR:

1. Use a pub/sub broker (Redis) to fan out sync events

• ➕ Reduces N-per-connection polling load
• ➕ Lower latency than poll-interval delivery
• ➕ Scales better across multiple control-plane instances
• ➖ Adds new infrastructure dependency and operational complexity
• ➖ Still needs replay/backlog semantics using sync_version/cursors
• ➖ Out of scope for the single-node v1 plan and current milestone (PR A)

2. Adopt an SSE library (e.g., better-sse / sse-channel)

• ➕ Less custom streaming/formatting code to maintain
• ➕ Library-tested edge handling for SSE basics
• ➖ Doesn't remove need for custom polling/cursor/backlog logic
• ➖ May not support a non-droppable shutdown control lane cleanly

Recommendation: The PR’s approach (poll-driven SSE with snapshot-consistent DB reads, bounded queue + control lane, and explicit shutdown draining) is a good fit for the current single-process control-plane and matches the Plan 7 decisions. The main strategic risk left for PR B is durability/observability (sync_cursors persistence and lag metrics), but for PR A the chosen design keeps semantics explicit and avoids introducing new infrastructure prematurely.

Files changed (16) +1086 / -44

Enhancement (9) +794 / -38

parseLastEventId.tsAdd strict raw-string Last-Event-ID/last_event_id parser +66/-0

Add strict raw-string Last-Event-ID/last_event_id parser

• Implements Decision 15 parsing: header wins when present (including blank), supplied-but-blank is an error, omitted both defaults to 0, and values are parsed as non-negative integers via BigInt without Number coercion.

control-plane/src/domains/sync/parseLastEventId.ts

sseRegistry.tsAdd SseRegistry for connection tracking and graceful shutdown +81/-0

Add SseRegistry for connection tracking and graceful shutdown

• Adds a connection registry singleton to track active SSE sessions, gate new connections, and broadcast shutdown control frames before aborting and cleaning up lingering sessions.

control-plane/src/domains/sync/sseRegistry.ts

sseSession.tsImplement poll-driven long-lived SSE session +209/-0

Implement poll-driven long-lived SSE session

• Runs the SSE lifecycle: sets socket no-delay, creates bounded write queue, flushes initial events, emits a bootstrap cursor, sends periodic ': ping' heartbeats, polls for new events + high-water in a snapshot, and aborts/cleans up on oversize frames or disconnect.

control-plane/src/domains/sync/sseSession.ts

sseWriteQueue.tsAdd bounded SSE write queue with backpressure + shutdown control lane +209/-0

Add bounded SSE write queue with backpressure + shutdown control lane

• Implements a per-connection bounded queue that enforces both frame-count and byte budgets for data, handles Node stream backpressure via 'drain', times out slow clients, and reserves a control lane that can always enqueue shutdown frames.

control-plane/src/domains/sync/sseWriteQueue.ts

syncRepository.tsAdd snapshot-consistent initial+poll batch queries with high-water +111/-3

Add snapshot-consistent initial+poll batch queries with high-water

• Extends 'getEventsSince' to accept bigint and an optional limit. Adds 'getInitialBatchWithHighWater' (REPEATABLE READ backlog probe returning 'backlogExceeded') and 'getPollBatchWithHighWater' for the SSE poll loop.

control-plane/src/domains/sync/syncRepository.ts

syncService.tsExpose getInitialBatchWithHighWater passthrough +8/-0

Expose getInitialBatchWithHighWater passthrough

• Adds a service-level wrapper for the repository’s snapshot-consistent initial batch + high-water query.

control-plane/src/domains/sync/syncService.ts

index.tsDrain SSE registry during graceful shutdown +7/-1

Drain SSE registry during graceful shutdown

• Integrates 'getSseRegistry().shutdownAll()' into SIGTERM/SIGINT shutdown path before stopping the recompute scheduler, protected by the forced-exit watchdog.

control-plane/src/index.ts

sync.tsRewrite /v1/sync/events as long-lived SSE with backlog 429; gzip only on /snapshot +99/-34

Rewrite /v1/sync/events as long-lived SSE with backlog 429; gzip only on /snapshot

• Applies route-scoped compression only to '/snapshot'. Replaces the '/events' handler with admission control, strict Last-Event-ID parsing, snapshot-consistent backlog probe returning 429 + reason header, SSE headers (including 'X-Accel-Buffering: no'), and session execution via 'runSseSession'.

control-plane/src/routes/sync.ts

AppError.tsAdd PAYLOAD_TOO_LARGE and SERVICE_UNAVAILABLE error codes +4/-0

Add PAYLOAD_TOO_LARGE and SERVICE_UNAVAILABLE error codes

• Extends the centralized error code mapping with HTTP 413 and 503 codes, used by the SSE handler’s error cases.

control-plane/src/shared/errors/AppError.ts

Tests (3) +212 / -0

parseLastEventId.test.tsUnit tests for parseLastEventId +68/-0

Unit tests for parseLastEventId

• Covers omitted vs supplied-but-blank behavior, header precedence over query, rejection of negative/decimal/non-numeric values, and BigInt parsing beyond MAX_SAFE_INTEGER.

control-plane/src/domains/sync/parseLastEventId.test.ts

sseRegistry.test.tsUnit tests for SseRegistry +63/-0

Unit tests for SseRegistry

• Validates admission control after 'stopAccepting()' and verifies 'shutdownAll()' delivers an 'event: shutdown' frame even when the data queue is full.

control-plane/src/domains/sync/sseRegistry.test.ts

sseWriteQueue.test.tsUnit tests for SseWriteQueue +81/-0

Unit tests for SseWriteQueue

• Tests SSE formatting helpers, oversized rejection, full-queue behavior, control-lane acceptance when data is full, and 'writtenCursor' tracking after async writes.

control-plane/src/domains/sync/sseWriteQueue.test.ts

Documentation (1) +6 / -6

HANDOVER.mdUpdate Plan 7 status to in-progress for SSE edge sync +6/-6

Update Plan 7 status to in-progress for SSE edge sync

• Updates handover metadata to reflect Plan 7 plan doc merged and Plan 7 implementation underway on 'feat/sse-edge-sync', with PR A (SSE handler) and PR B follow-up noted.

docs/superpowers/plans/HANDOVER.md

Other (3) +74 / -0

package-lock.jsonLockfile update for compression dependency tree +61/-0

Lockfile update for compression dependency tree

• Adds 'compression' and its transitive dependencies ('compressible', 'on-headers', etc.) to support snapshot-only gzip compression.

control-plane/package-lock.json

package.jsonAdd compression dependency and types +2/-0

Add compression dependency and types

• Adds 'compression' as a runtime dependency and '@types/compression' for TypeScript typing support.

control-plane/package.json

config.tsAdd syncSse configuration block +11/-0

Add syncSse configuration block

• Introduces 'config.syncSse' with tunables for poll/heartbeat intervals, connection limits, queue limits (frames/bytes), per-frame max bytes, slow-client timeout, and shutdown drain timeout.

control-plane/src/config.ts

[messagesgoel-blip]

Review: PR #15 — Plan 7 PR A implementation

Overall

High-quality implementation that faithfully follows the plan doc. The architecture (write queue with control lane, snapshot-consistent DB reads, registry for admission/shutdown) is well-thought-out. Tests exist for the new units.

Issues requiring changes

1. getInitialBatchWithHighWater — tenant filtering regression (syncRepository.ts:90-94)

When tenantId is undefined, the SELECT adds AND tenant_id IS NULL. This means non-tenant requests only return NULL-tenant events, excluding all tenant-scoped events. The old getEventsSince returned ALL events when tenantId was undefined. This is a behavior regression that could cause data loss.

Fix: only add AND (tenant_id IS NULL OR ...) when tenantId is provided. When undefined, return all events (no tenant filter).

2. Same bug in getPollBatchWithHighWater (syncRepository.ts:143-146)

Same issue: when tenantId is undefined, the query incorrectly adds AND tenant_id IS NULL.

Recommendations (non-blocking)

• SseWriteQueue.pump() uses queueMicrotask — on high throughput this could starve the event loop. Consider setImmediate or a threshold-based scheduler.
• The sendProblem https://verilink.dev/problems/... URL doesnt exist yet — fine for RFC 9457 compliance but worth documenting.
• No integration test in this PR — acceptable per the PR split plan.
• REPEATABLE READ isolation is correct but may need a retry loop under contention (deferrable).

Verdict

Two tenant-filtering correctness bugs that should be fixed. Everything else is solid.

[coderabbitai]

Actionable comments posted: 5

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@control-plane/src/domains/sync/sseSession.ts`:
- Around line 160-196: In the SSE session polling loop around
getPollBatchWithHighWater, add monitoring for per-connection polling load and
database latency, including connection/session context where available. Preserve
the existing poll-driven behavior and event delivery logic; use the metrics or
observability conventions already established in the surrounding module.

In `@control-plane/src/domains/sync/sseWriteQueue.ts`:
- Around line 62-68: Update the frame classification logic in the queue’s
enqueue/write method to return “oversized” whenever the frame size exceeds
either maxFrameBytes or maxBytes, before the “full” capacity check. Preserve
“full” only for frames that can fit the byte budget but cannot currently be
queued, so waitForSpace(size) can eventually succeed.

In `@control-plane/src/domains/sync/syncRepository.ts`:
- Around line 85-176: Extract the duplicated snapshot transaction,
tenant-filtered event query, high-water retrieval, commit/rollback, and client
release logic from getInitialBatchWithHighWater and getPollBatchWithHighWater
into a shared fetchEventsAndHighWaterInSnapshot helper. Keep
getInitialBatchWithHighWater responsible for requesting maxBatch + 1 events,
truncating the result, and setting backlogExceeded; have
getPollBatchWithHighWater delegate with limit unchanged.
- Around line 85-131: Update getInitialBatchWithHighWater and the SyncEvent
definition so BIGINT sync_version values use one consistent runtime type: either
convert query results to the existing number contract before returning events,
or change SyncEvent and all consumers to string/bigint. Ensure highWater and
event sync_version handling follow the same representation without unsafe
implicit assumptions.

In `@control-plane/src/index.ts`:
- Around line 40-49: Update the watchdog timeout in the shutdown flow around
forceExit to derive its duration from config.syncSse.shutdownDrainMs rather than
the hardcoded 10000ms value, preserving the existing forced-exit behavior while
allowing the configured SSE drain period to complete.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: 226bfe64-1aef-47b0-9ff4-721dba8ff5c9

📥 Commits

Reviewing files that changed from the base of the PR and between 998554d and 3bda00f.

⛔ Files ignored due to path filters (1)

• control-plane/package-lock.json is excluded by !**/package-lock.json

📒 Files selected for processing (15)

• control-plane/package.json
• control-plane/src/config.ts
• control-plane/src/domains/sync/parseLastEventId.test.ts
• control-plane/src/domains/sync/parseLastEventId.ts
• control-plane/src/domains/sync/sseRegistry.test.ts
• control-plane/src/domains/sync/sseRegistry.ts
• control-plane/src/domains/sync/sseSession.ts
• control-plane/src/domains/sync/sseWriteQueue.test.ts
• control-plane/src/domains/sync/sseWriteQueue.ts
• control-plane/src/domains/sync/syncRepository.ts
• control-plane/src/domains/sync/syncService.ts
• control-plane/src/index.ts
• control-plane/src/routes/sync.ts
• control-plane/src/shared/errors/AppError.ts
• docs/superpowers/plans/HANDOVER.md

[messagesgoel-blip]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[messagesgoel-blip]

Review of latest commit (08d6b0a)

All previous concerns are addressed:

1. Tenant filtering regression — FIXED — fetchEventsAndHighWaterInSnapshot only adds tenant filter when tenantId is provided. The old else { AND tenant_id IS NULL } bug is removed from both snapshot queries and getEventsSince.

2. PG_BIGINT_MAX guard — ADDED — rejects values above 9223372036854775807n with test coverage.

3. inflightBytes tracking — ADDED — SseWriteQueue now tracks bytes written to socket but not yet drained for accurate backpressure.

4. Socket error handling — ADDED — close/error listeners on response, safeWrite wrapper.

5. Sleep abort listener leak — FIXED — cleanup() runs on both resolve and reject with proper removeEventListener.

6. Shared DB helper — EXTRACTED — fetchEventsAndHighWaterInSnapshot eliminates duplication.

7. Oversized byte budget — FIXED — enqueueData now checks size > opts.maxBytes alongside maxFrameBytes.

Minor observation (non-blocking)

sleep() uses addEventListener without { once: true }. The cleanup function calls removeEventListener on resolve, so it is not a leak — but if abort fires between addEventListener and the timeout, the listener stays registered (harmlessly, since the signal is already aborted). Minor.

Verdict

All review feedback addressed. Clean commit. Ready to ship.