fix(dynostore): fix 3 partial-failure bugs found by object_store fault injection

[shortishly]

All three shared the same shape: commit a state transition durably in one object-store write, then do more non-atomic work, with the retry path keyed off "was I in the pre-transition state" rather than "did the work actually finish" - so a transient failure got silently treated as fully complete.

• produce(): the idempotent sequence number advanced durably before the record write that it's meant to guard. A failure writing the record left the sequence advanced anyway, so the client's mandatory retry was told DuplicateSequenceNumber - permanent, silent data loss. Fixed by splitting the check (read-only, unchanged position) from the durable advance, which now only commits after the record write succeeds.

• txn_end(): the transaction state moved from Begin to PrepareCommit before the per-partition end-of-transaction marker writes. A failure partway through left the retry recomputing an empty set of partitions needing a marker (since it was no longer in Begin), completing the transaction anyway with some partitions never marked - never marked, so read_committed consumers of those partitions stall at the transaction's first offset. Fixed by recomputing the set from whatever's still in txn_detail.produces whenever in a Prepare* state, not only on the Begin transition.

• delete_topic(): topic metadata was removed before the two delete_stream cleanup calls. A failure partway left a retry finding no metadata and reporting UnknownTopicOrPartition without retrying anything, orphaning the partition's record objects at a name-keyed path forever - blocking any topic later recreated with the same name (AlreadyExists, permanently). Fixed by removing metadata last, only once cleanup actually succeeds.

Each fix accepts a trade-off in exchange for closing the original permanent/silent failure:

• produce(): a duplicate instead of a loss. If the sequence advance fails after the record write succeeds (any transient failure, or a shutdown aborting the request between the two), the client's retry passes the pre-check and the batch is written a second time. This is not limited to a race. The pre-check can also pass on stale state (a cached meta.json on another broker, or a retry while the original put is still in flight). Failures of the post-write advance are now logged at error! and counted in nisshi_produce_post_write_sequence_rejected.
• txn_end(): a harmless second marker on retry. A retry whose decision contradicts the prepared state (commit, then abort) is rejected with InvalidTxnState.
• delete_topic(): a slightly larger window where a concurrent produce() or offset_commit() can land on a mid-delete topic.

Not fixed here, tracked as follow-ups:

• dynostore: idempotent produce can still duplicate a batch on retry #953: dedupe of produce retries against the partition's tail batch (or a pending-sequence CAS before the watermark bump)
• dynostore: delete_topic needs a "deleting" tombstone so produce, offset_commit and fetch reject a topic mid-delete #954: a "deleting" tombstone for delete_topic
• dynostore: transactional consumer offsets can be lost when offset_commit fails inside EndTxn #955: transactional consumer offsets lost when offset_commit fails inside EndTxn
• dynostore: init_producer abandons a transaction stuck in PrepareCommit or PrepareAbort #956: init_producer abandoning a transaction stuck in Prepare*

What does this change do, and why?

Fixes #742 (produce), and fixes the same class of bug in txn_end and delete_topic.

How was this tested?

traceforge_* fault-injection tests (each fails with the fix reverted), delete_topics, and the broker, dynostore and storage-sql suites; just clippy and cargo fmt --check are clean. The nisshi-schema berg tests need the lakehouse service and were not run.

Checklist

• Commits are signed off (git commit -s) — see CONTRIBUTING.md
• just fmt, just clippy, and just test pass locally
• Tests added or updated for the behavior changed
• Docs updated if user-facing behavior changed

[solace-aross]

Approving. All three fixes close real permanent failures, and the regression tests earn their green.

What I checked, by running it (not reading)

Check	Result
traceforge_* tests at PR head (-p nisshi-broker --features dynostore)	12/12 pass, under 2s total
Same 3 *_object_store_fault tests with dynostore.rs reverted to main's logic (keeping only the pub changes)	3/3 fail: :425 (record count), :630 (partition 1 has no marker), :273 (recreated topic can't produce)

So each test genuinely couples to its fix.

Two residuals are worth knowing about. Neither blocks, because each is less harmful than the bug it replaces. I reproduced both with throwaway tests using the same FaultInjectingObjectStore pattern (not pushed).

1. produce() now trades silent loss for a duplicate on any transient failure, not only on a race. Fault the meta.json write that advances the sequence after the record write succeeds, then retry the same batch:

attempt1        = Err(ObjectStore(Generic { .. "meta advance failed" }))
attempt2 (retry) = Ok(1)
fetched records = 2

The pre-check still sees the old sequence, so the retry is accepted and written again at a new offset. On main the same fault writes nothing and the retry is fine, so this is a new mode. Duplicate beats loss, so I agree with the direction. The code comment at :1040 and the commit message describe the residual as "duplicate-on-race" only, though, and that undersells it. Worth correcting the wording and tracking a follow-up. One way to close it: on a sequence mismatch, check whether the tail batch at the partition already carries this producer_id/base_sequence before accepting the retry. That's roughly how Kafka rebuilds producer state from the log.

2. txn_end accepts a retry with the opposite decision. Commit with partition 1's marker faulted, then txn_end(.., committed: false):

abort retry = Ok(None)
partition 0 control markers = 2   (commit, then abort)
partition 1 control markers = 1   (abort)

Partition 0 commits and partition 1 aborts, which breaks atomicity. This is pre-existing: main also returns Ok(None) there (partition 0 commit, partition 1 no marker). But this PR makes Prepare* an intentional retry path, so it's the natural place to reject a decision that contradicts the prepared direction. Kafka returns INVALID_TXN_STATE for this.

Smaller notes

• DynoStore and its builder are now pub in a published crate, only so tests can inject an ObjectStore. #[doc(hidden)] or a test-support feature would keep that out of the public API.
• About 2,400 of the added test lines (traceforge_spike, _group_join, _offset_commit, _produce_idempotent) don't relate to the three fixes. Seven module docs also point at docs/traceforge-test-plan.md, which isn't in the repo. Either add the plan or drop the references, so the next reader isn't sent to a missing file.
• The commit has no Signed-off-by line (CONTRIBUTING asks for DCO sign-off).

[sgamelin]

Approving. The three reorderings are right, and I agree with @solace-aross that a duplicate beats a loss. I've left three inline comments on points the existing review doesn't cover. Two more are about txn_end code outside the diff:

1. Transactional consumer offsets have the same bug, about 100 lines below the fix. The CAS that completes the transaction sets the state to Committed (:2726) and clears txn_detail.offsets (:2761). Only after that does offset_commit run (:2792). Its per-partition result is thrown away (_ = ...), and offset_commit reports a failed put as UnknownServerError inside Ok (:1496). So EndTxn returns success even when the sendOffsetsToTransaction offsets were never written. A retry finds the transaction Committed with no offsets left, so it has nothing to redo. In a consume-transform-produce app, that means input gets reprocessed and output duplicated. This predates the PR, so a follow-up is fine. The fix is to check the codes, and to clear offsets in a CAS only after the writes succeed.

2. A transaction stuck in Prepare* recovers only if the same epoch retries EndTxn. A producer that restarts after a failed EndTxn calls InitProducerId instead. init_producer rolls back only Begin (:2169), so it bumps the epoch and abandons the old transaction with its markers still missing. offset_stage counts Prepare* as open, so read_committed consumers on those partitions stall at the transaction's first offset. Could init_producer finish a Prepare* transaction before bumping the epoch, or could that go on a follow-up? Also, the description says the old bug left records "invisible to read_committed consumers forever". That doesn't quite match dynostore, which computes the last stable offset from transaction state, not from markers. The symptom to expect now is a stalled consumer.

One addition to the thread at :1054: on shutdown, accept_loop breaks and drops its JoinSet (nisshi-service/src/stream.rs:342), which aborts in-flight produces. So a rolling deploy under load can routinely stop a produce between the record put and the sequence advance. That makes the dedupe follow-up more urgent than a fix for a rare fault.

[solace-aross]

All three fixes share the same root cause (a durable state transition committed before the non-atomic work it's meant to guard) and all three are fixed the same way: reorder so the durable commit only happens after the work it depends on has actually succeeded.

Checked the produce() split directly: the pre-check is now read-only (.with, not .with_mut) and the durable sequence advance happens after the record write succeeds, re-checking rather than blindly re-applying, which correctly handles a concurrent write landing in between. delete_topic() now removes metadata last, after both cleanup calls, so a retry after a transient failure finds the topic again instead of orphaning its records. txn_end() recomputes the pending-partition set from txn_detail.produces on every Prepare* pass, not just the Begin transition, so a partial failure gets retried instead of silently completing with partitions unmarked.

The new pub(crate) -> pub visibility bump on DynoStore looks like it's there to let the new traceforge_* fault-injection tests in nisshi-broker/tests/it/ construct one directly - makes sense for an integration test, not a concern.

CI is green including the new fault-injection tests for two of the three bugs. Approving.

[solace-aross]

Re-approving after c94e529; my earlier approval was dismissed by the push. Both of my notes are addressed.

Checked in the delta:

• The produce comment now states the residual accurately: a failed sequence advance after the record write duplicates on retry for any transient failure, not only a race. The dedupe against the partition tail batch is a named follow-up.
• txn_end returns InvalidTxnState when a retry contradicts the prepared state (PrepareCommit with abort, PrepareAbort with commit), before any markers are written. end_txn_retry_with_opposite_decision_is_rejected covers the faulted-marker, opposite-decision, then same-decision sequence.
• Post-write Duplicate/OutOfOrder rejections now log offset, location, producer_id and base_sequence at error and increment nisshi_produce_post_write_sequence_rejected.
• The assert_matches crate swap to std::assert_matches touches only manifests and test code.

Notes:

1. Open from sgamelin and still worth a tracked issue before this ships widely: the pre-check can pass on stale state (the 5s metadata cache across brokers), so a fenced producer's batch can be written before the later check rejects it. On main the CAS ran first. The author's reply agrees and defers it, which I accept for the dynostore engine as it stands, since the new ordering removes a permanent loss.
2. The offset_commit window against a topic that is mid-delete, and the transactional offsets in txn_end being dropped when the write fails, are pre-existing and already acknowledged as follow-ups.
3. No test covers the new counter.

I did not rerun the fault-injection tests locally this pass; the new check and logging are read from the diff, and CI is green on 1e5db7a.

[solace-aross]

Re-approving at 739bb9a. Since 1e5db7a the branch gained two merges of main and b22512e, which moves the traceforge produce tests from acks=0 to acks=1. That matches #845, where an acks=0 produce now fails the request instead of returning per-partition error codes. The PR's own diff is otherwise unchanged apart from a comment-context shift from #889.

My earlier note stands: sgamelin's stale pre-check point is still deferred by the author. CI is green on this head.

[sgamelin]

Approving at d080ea7. The PR's own diff is identical to 739bb9a, the head of the last approval. Since then the branch was re-signed and merged with main, and all seven commits now carry Signed-off-by. The only change on main in the same code is #887, which touches init_producer and doesn't overlap the three fixes. I reran the traceforge_* and delete_topics tests on this head: 26 of 26 pass.

The commits since my earlier review read well. The produce comment now describes the residual accurately. A retry with the opposite decision returns InvalidTxnState, which matches Kafka 3.9.1 (TransactionCoordinator.scala#L554-L563). Gating the constructor behind test-support keeps it out of the published API.

Two asks, neither blocking:

1. Could you file the follow-ups before this merges? Fixes #742 closes the issue on merge, but the "Not fixed here" items in the description have no issues yet:

   • dedupe of produce retries against the partition's tail batch, or a pending-sequence CAS before the watermark bump;
   • a "deleting" tombstone for delete_topic, checked by produce, offset_commit and fetch;
   • transactional consumer offsets lost when offset_commit fails inside EndTxn;
   • init_producer abandoning a transaction stuck in Prepare* (after #887 it still rolls back only Begin).

   Without them, nothing tracks the duplicate-on-retry residual once #742 closes.

2. One inline comment on the new counter.

[solace-aross]

Approving at 10918fa. Sam's open point is closed: nisshi_produce_post_write_advance_failed now counts every error from the post-write advance, with a bounded reason attribute.

Checked in the delta since 739bb9a:

• Ran the memory-backed suites at this head: 21/21 across the traceforge tests, delete_topics::in_memory and the new post-write test. Postgres is not reachable here, so CI covers those legs.
• Narrowing the counter back to API errors only makes the new test fail, so it guards the fix.
• git merge-tree against main is clean.

One note, not blocking: reason has two catch-all values, otherwise (an object store error the request metric does not name) and other (anything else). A line in the counter's description would save a dashboard author from reading them as a typo.