fix(qwp): recycle the connection after a server NACK

[mtopolnik]

Fixes the ingestion stall behind the qwp-fuzz failures on #66: TestQwpFuzzIngressOraclePoisonErrorHandler timed out with only a fraction of the clean rows landing (e.g. 30/790, then 0/235 on rerun with a different seed), and a workflow_dispatch baseline on untouched main failed identically — the stall is pre-existing and independent of the rename PR.

Root cause. After the server rejects a frame it silently refuses every further frame on that connection (QwpIngressUpgradeProcessor.handleBinaryMessage: an unresolved sequence means "the client replays it from its acked watermark on a fresh connection"). The go client's PolicyDropAndContinue path advanced the watermark past the rejected frame and then kept draining on the same connection — waiting forever on acks the server would never send. Ingestion stalled without erroring after the first NACK. The server log shows the signature: one rejected [status=SCHEMA_MISMATCH...] per connection, then silence with the connection still open.

Fix. The receiver still advances the watermark past the dropped frame, then returns a plain error so run() recycles the wire through the normal reconnect path (connectWithBackoff → swapClient), replaying the refused tail on a fresh connection starting past the drop. Pre-send rejections with non-halt policies recycle the same way. This mirrors the Java client's post-rejection recycle (handleServerRejection: "recycling connection, will replay from fsn ..."), which is why the server's own Java-client e2e never hit this.

Tests.

• The mock server's rejectFirstNFrames now models the server contract: N error ACKs total, at most one per connection, with the connection's pipelined tail refused silently.
• TestQwpSfSendLoopDropAndContinue asserts the recycle; DropStreakThenHalt got a pipeline-faithful custom mock; DispatcherDrainTimeoutCap uses per-connection rejections with a tighter close bound.
• The poison test's dec256-overflow expectation follows the server's SCHEMA_MISMATCH classification of deterministic data-shape rejections (WRITE_ERROR was stale; both resolve to the Drop policy).

Validation. Local run of the CI race job's exact filter (go test -race -run '^TestQwp(Integration|FuzzIngress)') against a freshly built server master jar: 73/73 pass, no data races. The previously stalling seed now finishes in ~4s: all 235 clean rows land and handlerCalls == poisonedChunks == 21. Full -short suite green.

Merging this should turn #66's qwp-fuzz check green (that branch will pick the fix up via rebase or merge once this lands).

🤖 Generated with Claude Code

Summary by CodeRabbit

• Bug Fixes
  • Improved resilience when server rejections interrupt ordered delivery by recycling the connection and replaying from the affected point.
  • Updated rejection classification for schema-mismatch responses to correctly drop-and-continue behavior.
  • Reduced worst-case shutdown time during error handling.
• Tests
  • Tightened reconnection, replay, and capped-drop scenarios to better stress shutdown timing and rejection handling.
  • Updated fuzz expectations to match deterministic schema-mismatch behavior.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: cfa7be90-af2a-4703-925a-d7204822ae35

📥 Commits

Reviewing files that changed from the base of the PR and between 46d6b7d and b5c5c94.

📒 Files selected for processing (2)

• qwp_error_resilience_test.go
• qwp_sf_send_loop.go

🚧 Files skipped from review as they are similar to previous changes (2)

• qwp_sf_send_loop.go
• qwp_error_resilience_test.go

---

📝 Walkthrough

Walkthrough

The send loop now recycles connections after non-halting rejection responses, and test servers model silent refusal of subsequent frames. Resilience tests validate replay, drain timing, and halt behavior. Poison-row ingress tests now expect schema-mismatch errors with drop-and-continue handling.

Changes

Connection recycling

Layer / File(s)	Summary
Recycle connections after rejected frames qwp_sf_send_loop.go	Non-halting rejection paths return recycling errors after sender-error handling and dropped-frame acknowledgement.
Model capped rejection across connections qwp_sf_send_loop_test.go	The fake server tracks rejection ACKs globally and silently refuses later frames on each connection after its capped rejection.
Validate recycling and halt sequences qwp_error_resilience_test.go, qwp_sf_send_loop_test.go	Tests model per-connection refusal, replay, reduced rejection loads, tighter drain timing, and halt after three dropped write errors.

Schema mismatch validation

Layer / File(s)	Summary
Assert schema-mismatch delivery qwp_ingress_oracle_fuzz_test.go	Poison-row assertions now expect CategorySchemaMismatch, the corresponding server message, and PolicyDropAndContinue.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Sequence Diagram(s)

sequenceDiagram
  participant Sender
  participant ReceiverLoop
  participant QWPServer
  Sender->>QWPServer: send ordered frame
  QWPServer-->>ReceiverLoop: non-OK rejection ACK
  ReceiverLoop->>Sender: offer SenderError and return recycle error
  Sender->>QWPServer: reconnect
  Sender->>QWPServer: replay frames after dropped frame

Loading

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly describes the main change: recycling the connection after a server NACK.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
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

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch mt_qwp-nack-recycle

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

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

[mtopolnik]

Review: PR #67

Overview

After a server NACK, the real server silently refuses every further frame on that connection, so the old code's "keep draining on the same connection" stalled without erroring. The fix makes receiverLoop return a plain (non-terminal) error after a Drop-policy rejection — post-send after advancing the watermark past the dropped frame, pre-send without touching the watermark — which routes run() into connectWithBackoff → swapClient and replays the refused tail on a fresh connection. The mock server is updated to model the one-NACK-then-silence contract, and tests are adjusted accordingly.

Findings

1. Inaccurate comments (minor — addressed in a follow-up commit on this branch)

• qwp_sf_send_loop.go:1268 called connectWithBackoff "the same paced path a transport failure takes" — the round-walk backs off only after failed dials, so a healthy server that just NACKed is redialed immediately. Reworded to state that explicitly.
• qwp_error_resilience_test.go:899 — rejectFirstNFrames: 50, // 50 drops on conn 1 was stale: it's one drop per connection across 50 connections.
• qwp_error_resilience_test.go:460-462 (DropAcrossReconnect) — the described choreography ("conn 1 reads frame 1 then closes... One ACK delivered, so the silent-drop guard does not fire") no longer happens: the NACK itself recycles conn 1 before frame 1 is sent on it, so closeAfterFrames never fires. Comment now describes the recycle-driven flow.

2. Small nits (take or leave)

• rejectsIssued: the Load()-then-Add(1) pair isn't atomic; benign today (the client holds one connection at a time) but a CompareAndSwap/Add-based claim would keep the fixture safe if a test ever drives two senders at one server.
• assert.Less(t, elapsed, 750*time.Millisecond) in DispatcherDrainTimeoutCap: fine in principle (drain cap is 100 ms), but it now sits closer to CI-runner noise than the old 2 s bound. Not blocking; just be ready to loosen if it flakes.

Observation: the recycle is unpaced — superseded by pending policy work

Backoff engages only on failed dials, so a server NACKing every frame drives full recycle cycles at NACK rate (measured against a mock NACKing pre-send on every connection: ~59 reconnects/sec, indefinitely). This is deliberately not a fix-item for this PR: the Java client and spec have since moved to NACK policy v2 (design/qwp-nack-policy-v2.md — no drop policy, RETRIABLE/RETRIABLE_OTHER/TERMINAL tiers, FSN-keyed poison-frame detector, paced recycles, removal of the user-configurable on_*_error surface), and a Go port of v2 is pending as separate follow-up work — rough size a few focused days, a diff in the low thousands of lines, mostly tests. Pacing arrives there as an integral component rather than as a patch to the Drop model. This PR's recycle mechanism is exactly v2's RETRIABLE recycle, so it merges as-is and the port builds on it.

Test coverage, performance, security

• Coverage is good: the mock now enforces the real contract, DropAndContinue explicitly asserts the recycle happened, DropStreakThenHalt got a pipeline-faithful custom mock with correct per-connection seq-0 attribution (I traced the FSN math: drops at 0,1,2, halt at 3), and the fuzz test's WRITE_ERROR → SCHEMA_MISMATCH expectation change is behavior-neutral (both resolve to Drop) and tracks the floating server master, consistent with how that CI is built.
• Hot path untouched; allocations only on the error path. No security concerns — the untrusted rejection seq is still clamped, and the new error strings carry only status/category.

Verdict

The fix is correct and well-evidenced (root cause, Java cross-check, live-server validation) — approve. The comment corrections from finding 1 are in a follow-up commit on this branch; the NACK-policy-v2 port follows as its own body of work.