Keep final retry batches recoverable until processing completes

[robjarawan]

I reproduced the final disk-backed retry batch disappearing if the process stops after dequeue and before processing finishes.

Basically, DiskQueue.get() unlinks queue_file as soon as msg_count reaches zero. At that point Retry.after_accept() has only handed a download retry to the work phase, or Retry.after_work() has only handed a post retry to the posting phase. Neither operation has completed yet.

Reproduction

I tested fork development at 24de015ccbdd20419ff4cc58e560478009d6ed20 with the real Retry callbacks, disk queues, synthetic messages and temporary local state. Current upstream 446a826bc3c30e6523775809dd85d1544b583c67 has the same unlink path.

I committed two retry messages, retrieved a batch through the callback, stopped before its outcome callback, recreated Retry, and checked the recovered queue.

Case	Dequeued before stop	Recovered after restart
No dequeue control	0/2	2/2
Partial dequeue control	1/2	2/2
Final batch through Retry.after_accept()	2/2	0/2
Final batch through Retry.after_work()	2/2	0/2

2 failed, 2 passed

The controls rule out queue creation, housekeeping, serialization and normal close/reopen. The failure occurs only when DiskQueue.get() returns the final available batch and removes its file.

The download path can lose the retry before transfer completes. The post path can lose it before publication completes. Retry messages have already had ack_id removed, so a restart cannot recover these obligations from the original broker delivery.

I plan to keep the disk queue state until the corresponding retry outcome has been recorded, then remove it at an explicit completion boundary. I will cover download retry, post retry, successful completion and requeue-on-failure behavior in the regression tests.

Upstream #1746 is related to retry expiry but does not cover this dequeue/completion window. I did not find an existing fork issue for it.

[robjarawan]

I put the regression and fix in draft #149.

The completed regression has 11 cases. On fork development, the three final-batch restart cases and both retry-write failure cases fail, while six lifecycle controls pass: 5 failed, 6 passed. With the fix, all 11 pass.

The focused DiskQueue/Retry/RedisQueue run passes 60 passed, 1 skipped, and the broader unit run passes 365 passed, 1 skipped with Azure's external-emulator test excluded. I also combined this locally with #98 and #100; their focused retry/housekeeping tests pass 49 passed, 9 skipped.

The runtime commit is cb7bf9049. It leaves the original disk queue file recoverable until after_work() or after_post() records completion, and it does not retire the file if writing the retry outcome fails.

[robjarawan]

I extended the proof around the retry queue's failure boundaries and updated draft #149.

The same 46 regression cases produce 10 failed, 36 passed on the previous PR head and 46 passed on the current branch. The additional failures covered a retry-store exception that was caught before the in-flight batch was retired, a mid-read exception that advanced queue state, partial and flush-failed appends, cleanup failure, and false-empty retry metrics.

The current focused run is 79 passed, 1 skipped, and the broader unit run is 376 passed, 1 skipped with Azure's external-emulator tests excluded. I also checked the local combinations with #150 (86 passed) and #98/#100 (60 passed).

The branch now contains only the two product commits on fork development; the separate #140 test-harness commits are no longer inherited. Current head: a3dc4aee5.

[robjarawan]

I found and fixed one more recovery boundary in draft #149.

After an append failed and its rollback also failed, the previous head could remain blocked even after storage recovered. Later retry obligations stayed only in memory, and a restart recovered the older prefix without them. The new regression reproduces that behavior on the previous head. The current code retains the exact pre-append byte/count boundary, restores it before the next append, then writes the complete pending batch.

Current results:

• retry/DiskQueue regression: 51 passed
• focused retry/queue/callback suite: 80 passed, 1 skipped
• broader offline suite: 381 passed, 1 skipped
• combined with fix: preserve publisher-specific post retries #150: 99 passed, 1 skipped
• combined with fix: preserve failed messages until retry persistence #98/fix: preserve retry inputs on housekeeping failure #100: 80 passed, 1 skipped

The branch is three focused commits directly on fork development and no longer carries #140. Current head: 19dbb236564dbd39c9024b3f7c0fae791d3b4334.

Pending messages remain memory-owned while every storage write and rollback attempt continues to fail. A restart cannot recover a record that never reached durable storage. Once storage accepts an append again, the saved boundary is restored and the complete pending batch is written before later work proceeds.

[robjarawan]

I built the standalone proof for this outside pytest and ran it against both revisions. So what I did was parse a real subscribe config through one_config, drive the real Retry callbacks through the real Flow dispatcher, and kill the process with SIGKILL after the final dequeue but before any completion runs. No broker involved, so recovery can only come from local retry files.

Case	Baseline recovered	Fixed recovered
Final download batch killed	0/2	2/2
Final post batch killed	0/2	2/2
Restart before dequeue	2/2	2/2
Partial batch killed	2/2	2/2
Completed work restarted	0/2	0/2

Basically the baseline loses exactly the final batch on both paths, and the fix recovers the exact records without resurrecting completed work. Full procedure, per-phase logs and state listings are in outputs/sarracenia-bugs/high-priority-20260906/retry-final-dequeue/human-reproduction/.

[robjarawan]

Correction to my last comment: that harness drives the real Retry and DiskQueue methods, but it never launches a complete SR3 flow, so I am reclassifying it as supplemental callback-level proof, not the config-level reproduction. The full-flow proof through a running flow is still being completed. I also removed the credentials fixture I had created; that filename should never appear in this work.