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

[nikhiln64]

Fixes #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 the reset entry stays the last and only entry in pendingMarkDeleteOps for the duration of the reset and cannot be clobbered. The reset flag is set before the reset entry is ever enqueued and stays set until the reset completes, so the guard covers the whole window during which the reset entry is buffered. 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.

Verification

I ran only the new test in the managed-ledger module rather than a full build. I confirmed it passes with the fix in place and fails without it, which isolates the displacement as the defect. This is a concurrency correctness fix and a clean local run is weak evidence for timing sensitive behaviour, so the test is written to be deterministic by controlling the read completion and the queue state directly rather than relying on a natural race.

The root cause is structurally proven from the code. The one narrowing factor worth stating honestly is the residual skip branch inside internalMarkDelete that #26304 also mentions, where an in flight ack completing between the reset nulling persistentMarkDeletePosition and the reset persisting can re arm that field. This change removes the dominant displacement path completely by keeping any new individual delete out of the queue while a reset is in progress, which also stops any new ack from re arming that field mid reset. If maintainers would like the in flight skip branch hardened as well I am happy to follow up on that in this PR or a separate one, whichever is preferred.