Skip to content

Commit 4f4e1e5

Browse files
bluestreak01claude
bluestreak01
and
claude
committed
docs: cursor SF — durability & reconnect spec
Captures the design for closing the WS sender's reliability gap that landed when we collapsed onto the cursor engine: flush()/close() no longer wait for ACKs (in either mode), and memory mode can drop data on close-then-exit. Spec covers: - flush()/close() contracts (close gets a 5s drain timeout with fast-close opt-out) - Reconnect with bounded per-outage retry budget (default 5 min) and schema-reset machinery (volatile connectionGeneration counter to close the encode-mid-reconnect race) - Slot directory model: sf_dir is the parent, sender_id picks the slot, foreground sender + opt-in background drainers for orphan recovery - Server-side dedup contract (assumed) 13 decisions locked, no open items. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 36263c4 commit 4f4e1e5

1 file changed

Lines changed: 162 additions & 0 deletions

File tree

‎design/qwp-cursor-durability.md‎

Lines changed: 162 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,162 @@
1+
# QWP WebSocket sender — durability & reconnect spec
2+
3+
Status: **draft v3**, working notes for the cursor SF refactor on `vi_sf`.
4+
5+
## Goals
6+
- **Reduce data loss.** SF mode preserves every batch the producer has handed to the engine until the server has ACK'd it, surviving JVM crashes, process restarts, and transient network outages.
7+
- Memory mode (`ws::addr=...;` no `sf_dir`) is reliable enough for typical use under transient network blips.
8+
- SF mode (`ws::...;sf_dir=...`) survives process restarts and JVM crashes; disk does not grow under steady-state traffic (only ACK'd data is trimmed).
9+
- Failure surfaces are loud and distinguishable: "server slow" ≠ "server unreachable" ≠ "data refused".
10+
11+
## Modes
12+
| | Memory | SF |
13+
|---|---|---|
14+
| Storage | malloc'd ring | mmap'd files under sender's slot dir |
15+
| Cap | `sf_max_total_bytes` (default 128 MiB) | `sf_max_total_bytes` (default 10 GiB) |
16+
| Cap-full behavior | Producer's `flush()`/`at()` blocks up to `sf_append_deadline_millis`, then throws | Same |
17+
| Survives JVM exit | No | Yes (recovered on next startup; orphans optionally drained by another sender) |
18+
| Reconnect retries | Yes | Yes |
19+
20+
## flush() contract
21+
- Encodes accumulated rows into the cursor engine.
22+
- Returns when data is **published into the engine** (in-RAM for memory mode, on-disk for SF). **Never** waits for server ACK — ACKs are asynchronous and not every flush correlates to one.
23+
- The I/O loop drains in the background and retries on reconnect until either ACK or the cap forces backpressure → hard error to the producer.
24+
25+
## close() contract
26+
- One knob: `close_flush_timeout_millis`.
27+
- **Default `5000`**: close() blocks waiting for `engine.ackedFsn() >= engine.publishedFsn()` (server ACK'd everything published) for up to 5 s, then logs WARN and proceeds with stop.
28+
- **`0` or `-1`**: close() does not flush at all — fast exit. Pending data is lost (memory mode) or recovered by next sender (SF mode).
29+
- Any other positive value: that timeout in millis.
30+
31+
## Reconnect policy (both modes)
32+
- I/O loop catches any wire error (send fail, recv fail, server close, ACK timeout). Logs WARN and enters reconnect.
33+
- Backoff: exponential with jitter. Reuse `LineSenderBuilder.maxBackoffMillis` (initial 100 ms, cap as configured).
34+
- **Budget: `reconnect_max_duration_millis`** — per-outage time cap (resets on each successful reconnect). Once total elapsed time since the first failure of *this* outage exceeds the cap, the I/O loop gives up.
35+
- **Default 300_000 ms (5 min).** Long enough to ride out most server restarts and brief outages where the cause needs investigation; short enough that a permanently-gone server surfaces within minutes.
36+
- **Auth failure on reconnect (401, 403, non-101 upgrade reject) is terminal** — don't burn the retry budget on errors that won't fix themselves.
37+
- On successful reconnect: I/O loop restarts `nextWireSeq=0`, sets `fsnAtZero = engine.ackedFsn() + 1`, walks segments forward from there, and replays. Producer thread is signaled (volatile counter bump) so the next encoded batch carries full schema definitions instead of refs.
38+
- On budget exhaustion: connection error recorded → next user-thread API call throws.
39+
40+
### Initial connect
41+
- **Default: terminal.** Initial-connect failures (DNS, refused, bad auth, version mismatch) usually mean misconfig; throw immediately so the user sees the error, not a 5-minute hang.
42+
- **Opt-in: `initial_connect_retry=true`** uses the same backoff + `reconnect_max_duration_millis` cap as reconnect. Useful for "publisher comes up before server" scenarios (k8s ordering, dev environments).
43+
44+
### Logging cadence
45+
- WARN at first failure of an outage: `"disconnected from <addr>, reconnecting"`.
46+
- WARN throttled to once per `BACKPRESSURE_LOG_THROTTLE_NANOS` (5 s) during the retry storm — not one per backoff sleep, otherwise a 5-min outage at 100 ms backoff = 3000 lines.
47+
- INFO on each successful reconnect: `"reconnected to <addr> after <Xms>, <Y> attempts"`.
48+
- ERROR on budget exhaustion: `"giving up reconnecting to <addr> after <Xs>, <Y> attempts"`.
49+
50+
## Backpressure semantics
51+
- Engine cap full → `appendBlocking` spins for `sf_append_deadline_millis` (default 30 s) → throws.
52+
- Error message must distinguish:
53+
- `"backpressured for Xms — wire path is not draining (server slow?)"` (engine published, but server hasn't ACKed)
54+
- `"backpressured for Xms — Y reconnect attempts in progress (server unreachable since Z)"` (the I/O loop is in retry-backoff)
55+
56+
## Schema state on reconnect
57+
- Single volatile counter, single writer (I/O thread), shared across two roles:
58+
```java
59+
private volatile long connectionGeneration; // bumped by I/O loop on every successful reconnect AND on initial recovery from disk
60+
```
61+
- Producer's `flushPendingRows` does:
62+
```java
63+
int retries = 0;
64+
while (true) {
65+
long genBefore = connectionGeneration;
66+
if (genBefore != lastSeenGeneration) {
67+
resetSchemaStateForNewConnection();
68+
lastSeenGeneration = genBefore;
69+
}
70+
encoder.beginMessage(...); /* encode all tables */
71+
int messageSize = encoder.finishMessage();
72+
if (connectionGeneration == genBefore) break; // common case
73+
if (++retries >= MAX_SCHEMA_RACE_RETRIES /* =10 */) throw new LineSenderException("schema-reset race exceeded retry limit");
74+
// gen advanced mid-encode → bytes are poisoned, discard + loop.
75+
// Table buffers are NOT reset until after this loop, so source rows are intact.
76+
}
77+
```
78+
- **On initial open with on-disk recovery** (SF mode, non-empty slot): `connectionGeneration` starts at 1, not 0. Recovered FSNs were never seen by *this* server connection, so the first batch must publish full schemas.
79+
80+
## Slot directory model
81+
82+
**`sf_dir` is a parent (group root)**, not a slot. The actual slot is `<sf_dir>/<sender_id>/`.
83+
84+
### Identity
85+
- **`sender_id` defaults to `"default"`.** Single-sender users get zero-config: their slot is `<sf_dir>/default/`.
86+
- **Multi-sender users must set `sender_id` explicitly.** Two senders trying to use the default name will collide on the lock — surfaced loudly as `"sf slot already in use by PID X"`.
87+
- The slot dir holds segments + `.lock` (advisory exclusive `FileChannel.tryLock`).
88+
- Lock released on `engine.close()` or OS-level process exit (kernel releases `fcntl`/`LockFileEx` locks automatically on crash).
89+
90+
### Foreground sender
91+
- Locks `<sf_dir>/<sender_id>/.lock`.
92+
- Recovers segments via `SegmentRing.openExisting`. Recovery is per-slot, in baseSeq order — preserves publishing order trivially.
93+
- Seeds `SegmentManager.fileGeneration` to `max(existing sf-<gen>.sfa hex) + 1` to avoid filename collisions with recovered files.
94+
95+
### Background drainers (orphan adoption)
96+
- **Opt-in: `drain_orphans=true`** (default false).
97+
- At foreground sender startup, scan `<sf_dir>/*/` for sibling slots that are (a) unlocked and (b) contain unacked segments.
98+
- For each orphan, spawn a background drainer:
99+
- Locks the orphan's `.lock`
100+
- Opens its own `WebSocketClient` (separate connection from the foreground sender)
101+
- Recovers segments, drains them in baseSeq order
102+
- Releases lock and exits when the slot is fully ACK'd and empty
103+
- **Drain-only**: no user appends, no public API for writing.
104+
- **Cap concurrent drainers: `max_background_drainers=4`** (default). Excess orphans are queued and started as earlier drainers finish.
105+
- **Drain failure policy**: drainer's reconnect cap exhausts, or auth fails, or segments are corrupt → drainer drops a `.failed` sentinel in the slot, releases the lock, exits. Future foreground startups skip slots with `.failed` until the user clears the sentinel manually. Bounded automatic retry, then human-in-the-loop.
106+
- **No automatic cleanup of empty slot dirs.** Goal is data preservation; only ACK'd data is trimmed (within a slot, by the segment manager). Empty slot dirs are cheap and stay forever unless the user removes them.
107+
108+
### Visibility
109+
- WS-only accessor `sender.getBackgroundDrainers()` returns a snapshot list: `{dir, framesPending, framesAcked, lastError, isFailed}`.
110+
- Lets users observe orphan-drain progress without parsing logs.
111+
112+
### Per-sender threading cost
113+
- Each engine (foreground + each background drainer) has its own `SegmentManager`. That's 1 manager thread + 1 I/O thread per engine. With `max_background_drainers=4`, worst case is 1 (foreground) + 4 (drainers) = 5 engines = 10 threads + 5 sockets per `Sender.fromConfig` call. Acceptable for typical deployments; users with hundreds of senders per JVM should set `max_background_drainers` low.
114+
115+
## Configuration knobs (connect string)
116+
| Key | Default | Mode | Status |
117+
|---|---|---|---|
118+
| `sf_dir` | unset | both | existing (semantics: now a parent dir) |
119+
| `sender_id` | `"default"` | SF | **NEW** |
120+
| `sf_max_bytes` | 4 MiB | both | existing |
121+
| `sf_max_total_bytes` | 128 MiB / 10 GiB | both | existing |
122+
| `sf_durability` | `memory` | SF | existing (`flush`/`append` reserved) |
123+
| `sf_append_deadline_millis` | 30000 | both | **NEW** (currently a constant) |
124+
| `reconnect_max_duration_millis` | 300000 | both | **NEW** |
125+
| `reconnect_initial_backoff_millis` | 100 | both | **NEW** |
126+
| `max_backoff_millis` | already exists | both | reuse existing |
127+
| `initial_connect_retry` | `false` | both | **NEW** |
128+
| `close_flush_timeout_millis` | 5000 (0/-1 = fast close) | both | **NEW** |
129+
| `drain_orphans` | `false` | SF | **NEW** |
130+
| `max_background_drainers` | 4 | SF | **NEW** |
131+
132+
Each new knob also gets a `LineSenderBuilder` setter.
133+
134+
## Counter accessors (WS-only, on QwpWebSocketSender)
135+
- `getTotalBackpressureStalls()` — already exists
136+
- `getTotalReconnectAttempts()`
137+
- `getTotalReconnectsSucceeded()`
138+
- `getTotalFramesReplayed()`
139+
- `getBackgroundDrainers()` — list of `{dir, framesPending, framesAcked, lastError, isFailed}`
140+
141+
## Stated assumptions (server contract)
142+
- Server **dedups** replayed batches by `messageSequence`. Replay-after-reconnect produces duplicates; without server-side dedup, every reconnect = double-write. Legacy code already relied on this; the new design continues to.
143+
- Server's dedup window must be ≥ a sender's `sf_max_total_bytes` worth of FSNs (else replay = double-write under sustained outage + full cap).
144+
- Coordination/testing of the recovery + dedup contract is **outside this repo's scope**.
145+
146+
## Decisions locked
147+
1. ✅ flush() never waits for ACK (ACKs are async).
148+
2. ✅ Reconnect cap is per-outage time-based, default 300s.
149+
3. ✅ close() drains by default with 5s timeout; `close_flush_timeout_millis=0|-1` opts out for fast close.
150+
4. ✅ Schema-reset is also fired on disk recovery (recovered state == post-reconnect state).
151+
5. ✅ Encode-mid-reconnect race closed via single volatile `connectionGeneration` counter + retry loop in `flushPendingRows`.
152+
6. ✅ Slot dir model: `sf_dir` is parent; per-sender slots `<sf_dir>/<sender_id>/`; default `sender_id="default"`.
153+
7. ✅ Orphan adoption is opt-in (`drain_orphans=true`); foreground sender spawns background drainers per orphan, capped at `max_background_drainers`.
154+
8. ✅ Drain failure → `.failed` sentinel; bounded retry + human-in-the-loop.
155+
9. ✅ Initial connect terminal by default; opt-in retry via `initial_connect_retry=true`.
156+
10. ✅ Auth failures (401/403/non-101) terminal even on reconnect.
157+
11. ✅ Logging: WARN on outage entry/exit-attempt, INFO on reconnect success, ERROR on budget exhaustion; throttled.
158+
12. ✅ Counters and orphan-drainer visibility on `QwpWebSocketSender` (WS-only).
159+
13. ✅ No automatic cleanup of empty slot dirs — preserve goal of data-loss reduction.
160+
161+
## Open
162+
None. Ready to implement.

0 commit comments

Comments
 (0)