[fix][ml] Prevent durable cursor reset from being silently discarded by a concurrent individual delete

[nikhiln64]

Related to #26304

Motivation

A durable cursor reset, which is what backs pulsar-admin topics reset-cursor and a client consumer.seek(...), can have its entire state mutation dropped while still reporting success to the caller. The admin REST call returns 2xx and the client seek future completes normally echoing the requested position, but the cursor never actually moves. That turns a correctness operation into a silent no op, and the state it leaves behind is durable, so a broker restart reproduces the un reset state rather than healing it.

The reason this happens is a queue ordering problem in ManagedCursorImpl. Since #25047 the reset no longer carries its state change in its completion callback. Instead the whole mutation, which corrects messagesConsumedCounter, assigns markDeletePosition and readPosition, clears individualDeletedMessages and re seeds batchDeletedIndexes, lives inside the alignAcknowledgeStatusAfterPersisted runnable of a single MarkDeleteEntry. When a read is in flight the reset entry is not applied immediately. It is buffered in pendingMarkDeleteOps because PENDING_READ_OPS is greater than zero, and it waits there until the read completes.

The trouble starts if another mark delete lands while the reset entry is buffering. In practice this is an individual acknowledgement arriving through asyncDelete, and there are ack sources that need no connected consumer and so survive the consumer disconnect that resetCursorInternal performs first, transaction pending ack commit being the clearest example. That later entry is appended behind the reset entry in the queue. When the read finally completes, internalFlushPendingMarkDeletes persists only pendingMarkDeleteOps.getLast() and runs only that last entry's runnable, while triggerComplete still fires the callback of every entry in the group. So the later delete wins, its position is what gets persisted, the reset's runnable is dropped, and yet the reset callback runs and calls resetComplete with success. The mirror ordering, where the ack is queued first and the reset last, is harmless because then the reset is getLast() and its runnable is the one that runs.

The cumulative mark delete path already guards against exactly this. asyncMarkDelete rejects a mark delete outright when RESET_CURSOR_IN_PROGRESS is set, so a cumulative ack can never slip in behind a reset. The individual delete path in asyncDelete is simply missing the same guard, which is why an individual ack is able to enqueue behind the reset entry and displace it.

Modifications

I added the same reset in progress guard to asyncDelete that asyncMarkDelete has always had. While a reset is in progress asyncDelete now fails the delete with a clear message instead of queueing it behind the reset entry, so a delete that arrives once a reset is already in progress cannot be clobbered. Rejecting the ack is consistent with the existing cumulative behaviour and preserves at least once semantics, since a rejected ack is simply redelivered. Because the change lives entirely in the durable cursor path it does not touch non durable cursors or readers, which run alignAcknowledgeStatus synchronously and have no pending queue, and are not affected by this bug in the first place.

I also added a focused unit test in managed-ledger that reproduces the drop deterministically using hooks that already exist in the tree. It holds a data read open with a PulsarMockReadHandleInterceptor so PENDING_READ_OPS stays above zero and the mark delete queue buffers, issues an asyncResetCursor back to the first position and waits until the reset entry is queued, then issues an asyncDelete of a later position, and finally releases the read and asserts that the reset actually took effect by checking readPosition, markDeletePosition and that the later position was not left deleted. The test fails without this change because the reset is silently displaced, and passes with it.

Scope and remaining work

The root cause is structurally proven from the code and the test isolates the displacement. On review Denovo1998 correctly pointed out that this change narrows rather than fully closes the race. The reset in progress flag is checked before the delete is enqueued, but that check and the enqueue are not atomic with the reset setting the flag and enqueuing its own entry, so a delete that reads the flag as false and is then preempted can still enqueue behind a reset that starts in between. The same check then act shape exists in asyncMarkDelete, and the internalMarkDelete backward position skip that #26304 also describes can still re arm persistentMarkDeletePosition. A complete fix needs the reset and the ack paths to share a real serialization barrier rather than a flag check, either by performing the reset in progress check under the same pendingMarkDeleteOps monitor that enqueues the entry so the check and the enqueue are atomic, or by draining accepted and in flight mark deletes before the reset is submitted. I have changed this from Fixes to Related so #26304 stays open for that work, and I am discussing the preferred approach on the review thread.

[Denovo1998]

There also appears to be a second silent-success path from #26304 that remains unresolved. An acknowledgment already in flight before the reset can complete after internalResetCursor clears persistentMarkDeletePosition, re-arming that field. The reset may then hit the backward-position skip in internalMarkDelete, which calls triggerComplete() without alignAcknowledgeStatus(), again reporting a successful reset without applying the reset state.

The PR description already notes this remaining path, but currently uses "Fixes #26304," which would close an issue that still contains this correctness case. Could we either harden that path in this PR—with a regression test—or narrow the scope and keep #26304 open for the remaining case?

[Denovo1998]

Can this test confirm that the delete fails specifically due to a reset in progress? Currently, both deleteComplete and deleteFailed count down the same latch, so the test does not validate the behavior introduced by this patch.

More importantly, we need a deterministic branch where the delete passes the reset-in-progress check before the reset sets the flag, and is then released after the reset entry is queued. In the current test, asyncDelete is always called after the reset entry already exists, so it does not cover the check-before-CAS race condition in asyncDelete.

[nikhiln64]

Agreed on both. The shared latch means the test passes whether the delete completes or fails, so it does not actually assert the reset in progress rejection, and because asyncDelete is only called after the reset entry exists it never exercises the check before the flag is set. I will split the latch so the test asserts the delete fails with the reset in progress exception specifically, and add a case that releases the delete in the window before internalResetCursor sets the flag, which is the race that needs the serialization fix. I will bring both in with that change so the test fails on the current code and passes on the fixed code.

[Denovo1998]

This check does not serialize the delete with the start of a reset. asyncDelete can observe FALSE, get preempted, then internalResetCursor can CAS the flag to TRUE and enqueue the reset entry. When the delete resumes, it can still mutate the cursor state and enqueue its MarkDeleteEntry behind the reset. If a read is pending, internalFlushPendingMarkDeletes() would then pick that delete as getLast(), so the same reset-displacement/silent-success scenario remains possible.

asyncMarkDelete appears to have the same check-then-act issue. We need an atomic reset/ack barrier, or we must drain all accepted and in-flight mark-delete operations before submitting the reset—not just checking the flag before the operation starts.

[nikhiln64]

You are right, and this is the important one. The guard I added mirrors the existing asyncMarkDelete check, so it inherits the same check then act shape you describe. The flag read in asyncDelete and the enqueue in internalAsyncMarkDelete are not atomic with internalResetCursor setting the flag and enqueuing the reset entry, so a delete that observes false and is then preempted can still land behind a reset that started in between, and asyncMarkDelete has the same gap. So this change narrows the dominant window, the transaction ack arriving well after the reset, but it does not serialize the two paths.

I see two ways to actually close it. One is to move the reset in progress check inside the synchronized pendingMarkDeleteOps block in internalAsyncMarkDelete, and to have internalResetCursor set the flag and enqueue the reset entry under that same monitor, so a delete either observes the flag before the reset holds the lock or is ordered after it with no interleaving, and this covers asyncMarkDelete in the same place. The other is the drain you mention, holding new acks and waiting for pendingMarkDeleteOps to empty before the reset is submitted. The lock scoped check is the smaller change and reuses the monitor that already orders the queue, while the drain gives a stronger barrier. Which would you prefer for this area, and I will implement it with a deterministic test for the check before CAS race.

[nikhiln64]

Thanks for the careful review Denovo1998, these are all correct. On the closing keyword you are right that Fixes 26304 would close an issue that still holds the internalMarkDelete re arm case and the check before CAS window, so I have changed the description from Fixes to Related and 26304 stays open. I would rather harden the remaining path in this PR than narrow it out, using whichever serialization approach you prefer from the thread on asyncDelete, and I will bring the deterministic test for the check before CAS race in with it.

[nikhiln64]

I went with the lock scoped serialization we discussed. The reset in progress reject now lives inside internalAsyncMarkDelete, under the same pendingMarkDeleteOps monitor that appends to the queue, so the check and the enqueue are one indivisible step for both the cumulative asyncMarkDelete path and the individual asyncDelete path, and the rate limiter fast paths that skip the queue carry the same check through updateLastMarkDeleteEntryToLatest.

The subtlety worth calling out is where the flag is armed. Arming it early, before the reset entry is enqueued, rejects an ack that was already accepted just ahead of the reset, and that regressed the compaction cursor case where a mark delete carrying CompactedTopicLedger needs its properties to survive a concurrent reset. So the reset now arms RESET_CURSOR_IN_PROGRESS exactly where it enqueues its own entry inside internalAsyncMarkDelete. The flag being set is then equivalent to the reset entry being committed to the queue. An ack that reaches the monitor before that observes the flag unset, is accepted, and orders ahead of the reset, which stays getLast and inherits the ack properties. An ack that arrives after observes the flag set and is rejected. Either way the reset cannot be displaced, and an ack accepted before the reset keeps its properties rather than being dropped.

On the test, I rewrote it to the deterministic check before enqueue race. It parks a delete on the cursor write lock at the point just after the old early check, arms the reset while it is parked, then releases so the delete reaches the guarded enqueue only afterwards, and it asserts the delete fails specifically with the reset in progress error rather than sharing a latch with success. The full ManagedCursorTest is green including the existing compaction property test, which the early arming approach had broken.

I kept Related to 26304 rather than Fixes, since this targets the displacement race and you separately raised the internalMarkDelete backward position skip, where an ack ordered before the reset can re arm persistentMarkDeletePosition after the reset clears it. That path is not closed by this change. I am glad to take it on in this PR or a follow up, whichever you prefer.