[Bug] Durable cursor reset (seek / reset-cursor) can be silently discarded while reporting success when another mark-delete is queued after it

## Search before reporting

- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

## Read release policy

- [x] I understand that [unsupported versions](https://pulsar.apache.org/contribute/release-policy/#supported-versions) don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

## User environment

**This is a code-analysis report, not an observed incident.** There is no production failure behind it, no reproducer has been written or run, and no test was executed. Everything below was established by reading the code on `master` and by `git log`/`git blame` on the commit that introduced the behaviour change. The reachability discussion at the end is deliberately explicit about what is structurally proven versus what remains untested.

This was noticed while reviewing PR #26299 (an unrelated `internalReadFromLedger` optimization). It is **independent of that PR** — the defect described here predates it and is not affected by it either way. The original concern raised during that review (a stale read-position hop applied across a reset) did *not* hold up: for durable cursors the `PENDING_READ_OPS` deferral at `ManagedCursorImpl.java:2394` orders the reset last. Following that thread is what surfaced the issue below.

- Code examined: `apache/pulsar` `master` at commit `cb90c12a14d`. All line numbers below are from that tree.
- Affected component: `managed-ledger` (`ManagedCursorImpl`), reached from broker `seek` and admin `reset-cursor`.
- Affected branches: `master`, `branch-5.0-M1`, `branch-4.2`, `branch-4.1`, `branch-4.0`.
- Affected releases: `4.0.9+`, `4.1.3+`, `4.2.0+`, `5.0.0-M1`. Not affected: `branch-3.0`, `branch-3.3`, and 4.0/4.1 releases predating the backports (see *Regression provenance*).
- Not affected: non-durable cursors / `Reader` seek — `NonDurableCursorImpl.internalAsyncMarkDelete` runs `mdEntry.alignAcknowledgeStatus()` synchronously (`NonDurableCursorImpl.java:116`) and has no pending queue.

## Issue Description

A durable cursor reset — `pulsar-admin topics reset-cursor`, or a client `consumer.seek(...)` — can have its **entire state mutation silently dropped while the operation reports success** to the caller, if another mark-delete entry is enqueued behind it while the cursor's mark-delete queue is buffering.

### Why the reset can be dropped

Since #25047, the reset's whole state mutation lives in a per-`MarkDeleteEntry` runnable, not in the reset's completion callback:

* `ManagedCursorImpl.java:1627-1679` — `alignAcknowledgeStatusAfterPersisted` for the reset. It corrects `messagesConsumedCounter` (`:1631-1651`), calls `individualDeletedMessages.removeAtMost` (`:1639`), assigns `markDeletePosition` (`:1652`) and `lastMarkDeleteEntry` (`:1653`), does `individualDeletedMessages.clear()` (`:1655`), clears and re-seeds `batchDeletedIndexes` (`:1656-1664`), emits the two confirming log lines (`:1671`, `:1676`), and assigns `readPosition` (`:1678`). It is handed to `internalAsyncMarkDelete` at `:1721-1732`.
* `ManagedCursorImpl.java:1681-1702` — the reset's own completion path (`finalCallback.operationComplete()`) mutates **no cursor state**. It calls `ledger.onCursorReadPositionUpdated` (`:1688`), clears `pendingMarkDeleteOps` (`:1693`), CASes `RESET_CURSOR_IN_PROGRESS` back to `FALSE` (`:1694`), and calls `callback.resetComplete(newReadPosition)` (`:1701`).
* `ManagedCursorImpl.java:2363-2409` — `internalAsyncMarkDelete`, under `synchronized (pendingMarkDeleteOps)` (`:2368`), **queues** the entry instead of executing it in three cases: `NoLedger` (`:2384`), `SwitchingLedger` (`:2390`), and `Open` with `PENDING_READ_OPS_UPDATER.get(this) > 0` (`:2394-2396`). Nothing in that block inspects whether a reset entry is already queued; `peekLast()` (`:2370`) is read only to inherit `properties`.
* `ManagedCursorImpl.java:3275-3281` — `internalFlushPendingMarkDeletes` persists **only** `pendingMarkDeleteOps.getLast()` (`:3276`), after setting that one entry's `callbackGroup` to the whole queue (`:3277`).
* `ManagedCursorImpl.java:2456-2478` — on persist completion, only `mdEntry.alignAcknowledgeStatus()` runs (`:2468`), i.e. the last entry's runnable, while `mdEntry.triggerComplete()` (`:2477`) fires the callback of **every** entry in `callbackGroup` (`:277-290`).
* `alignAcknowledgeStatus()` has exactly two invocation sites in the tree: `ManagedCursorImpl.java:2468` and `NonDurableCursorImpl.java:116`. There is no second path that replays a displaced runnable.

### The interleaving

1. A read is in flight, so `PENDING_READ_OPS > 0` (incremented at `ManagedCursorImpl.java:946`, `:1174`, `:3692`).
2. The reset enqueues entry **R** carrying the reset runnable (`:1721-1732` → `:2394-2396`).
3. Another mark-delete — in practice an individual acknowledgement via `asyncDelete` → `internalAsyncMarkDelete(..., null)` (`:2709-2720`) — enqueues entry **A** behind it. Its `null` runnable is replaced in the `MarkDeleteEntry` constructor by the default at `:260-268`.
4. The read completes → `readOperationCompleted()` (`:3733-3746`) → `flushPendingMarkDeletes()` (`:3738`) → `internalFlushPendingMarkDeletes()` picks `getLast() == A` and sets `A.callbackGroup = [R, A]`.
5. On persist completion, only **A**'s runnable runs (`:2468`). **R**'s reset runnable is dropped. `triggerComplete()` (`:2477`) nevertheless fires **R**'s callback → `finalCallback.operationComplete()` → `callback.resetComplete(newReadPosition)`.

The mirror ordering (ack queued first, reset last) is benign: the reset is then `getLast()`, its runnable wins, and only the ack's default runnable is lost.

### User-visible consequence

`PersistentSubscription`'s `resetComplete` (`PersistentSubscription.java:1026-1036`) unfences the subscription and completes the future, so the admin REST call returns 2xx and the client `seek()` future completes normally, echoing the requested position — while the cursor never moved:

- **Replay from a timestamp never happens.** `reset-cursor --time 24h` or `seek(timestamp)` reports success; `readPosition` is unchanged, so a caught-up consumer receives nothing and the team concludes retention deleted the data or the timestamp was wrong.
- **A forward skip past a poison message does not take.** Consumers reconnect after the disconnect at `PersistentSubscription.java:978-984` and are immediately redelivered the same message, resuming the crash loop, with no error to point at.
- **The resulting state is worse than a no-op — it is ack-advanced and durable.** `asyncDelete` mutates `individualDeletedMessages` and calls `setAcknowledgedPosition` (which assigns `markDeletePosition` at `:2225` and can push `readPosition` forward) *synchronously before queueing*, and `LAST_MARK_DELETE_ENTRY_UPDATER` (`:2447-2454`) keeps the later position, so `lastMarkDeleteEntry` becomes A. `persistPositionToLedger` writes `mdEntry.newPosition` (`:3493`) plus the live, never-cleared `individualDeletedMessages` (`:3505`/`:3515`). A broker restart therefore reproduces the un-reset state rather than healing it.
- **Cursor properties are collaterally wiped.** `asyncDelete` passes `properties == null` (`:2709`), so A inherits the last queued entry's properties (`:2370-2372`) — the reset's `Collections.emptyMap()` (`:1721`) — and that empty map is what gets persisted.
- **Transient ledger/cursor divergence.** `ledger.onCursorReadPositionUpdated(this, newReadPosition)` still fires at `:1688`, so `ManagedLedgerImpl.updateActiveCursor` (`ManagedLedgerImpl.java:2694-2698`) records a read position the cursor never adopted.

### A second, related silent-success path

`internalMarkDelete` has two skip branches that call `mdEntry.triggerComplete()` **without** ever calling `alignAcknowledgeStatus()`: the `persistentMarkDeletePosition` backward guard (`:2412-2421`) and the `INPROGRESS_MARKDELETE_PERSIST_POSITION` guard (`:2432-2440`). `internalResetCursor` nulls both fields at `:1719-1720`, but the default ack runnable re-arms `persistentMarkDeletePosition` (`:267`). So even when R *is* the flushed entry, a concurrent ack completing in between can make a backward reset take the skip branch — `resetComplete(SUCCESS)` with zero state change and nothing persisted. Any fix should cover both paths.

## Error messages

```text
No exception, no failed future, and no ERROR/WARN log is produced: the reset takes the
success branch end to end.

The one usable forensic marker is an asymmetry in the INFO log. internalResetCursor logs
on entry, outside the dropped runnable:

ManagedCursorImpl.java:1601-1604 "Initiate reset readPosition"

while both confirming lines live INSIDE the dropped runnable:

ManagedCursorImpl.java:1668-1671 "Reset readPosition to before current readPosition"
ManagedCursorImpl.java:1673-1676 "Reset readPosition to, skipping from current readPosition"

A dropped reset therefore shows "Initiate reset readPosition" with no matching completion
line for that cursor.
```

## Reproducing the issue

No reproducer has been written or run. The window is timing-dependent, so a natural-traffic reproduction is unlikely; a deterministic unit test in `managed-ledger` is the practical route. Sketch, using hooks that already exist in the tree:

1. `factory.open(...)`, `openCursor("c1")`, `cursor.setInactive()` before adding entries (so the entry cache does not bypass the mock bookie — precedent at `ManagedCursorTest.java:6283`), then add 10 entries `p0..p9` and `markDelete(p4)` so the state is `Open` with an existing cursor ledger.
2. Hold a read open: `bkc.setReadHandleInterceptor(...)` (`PulsarMockReadHandleInterceptor`, used already at `ManagedCursorTest.java:6253-6280`) returning an uncompleted future. Wait until `cursor.getPendingReadOpsCount() == 1` (`ManagedCursorImpl.java:3878`).
3. `cursor.asyncResetCursor(p0, false, cb)` — it hops to `ledger.getExecutor()` (`:1740`), so wait on `cursor.pendingMarkDeleteOps.size() == 1` (the field is `protected`, `:307`, so a same-package test needs no reflection).
4. `cursor.asyncDelete(p5, cb, null)` with `p5` strictly after `markDeletePosition` so it is not short-circuited; wait for `size() == 2` and assert `getLast().newPosition == p5` to pin the precondition.
5. Complete the interceptor future. `readOperationCompleted()` → flush → `getLast() == A`.
6. Assert the reset actually applied: `getReadPosition() == p0`, `getMarkDeletedPosition() == ledger.getPreviousPosition(p0)`, `isMessageDeleted(p5) == false`. These should fail today while `resetComplete` still fires.
7. Control arm: same test with the ack enqueued *first* and the reset second — the same assertions should pass, isolating displacement as the defect.
8. Optional durability arm: close and reopen the ledger/cursor and re-assert, showing a restart does not heal it.

A second, read-free variant exercises the independent flush path: force `NoLedger`/`SwitchingLedger` (`:2384-2390`) via delayed cursor-ledger creation and enqueue R then A — the flush at `:3229`/`:3248` uses the same `getLast()` logic.

## Additional information

### Regression provenance

This is a regression, not long-standing behaviour. `git log -S"alignAcknowledgeStatusAfterPersisted"` yields exactly one commit:

> `81aff30c4b461b2c630f413c407ba014e7a5d571` — *"[fix][broker]Fix incorrect backlog if use multiple acknowledge types on the same subscription (#25047)"*, 2025-12-15.

Its diff moves the reset's state mutation **out of** `finalCallback.operationComplete()` and **into** the per-entry runnable. That matters because `operationComplete()` is reached via `MarkDeleteEntry.triggerComplete()`, which runs for *every* entry in the `callbackGroup` — so before #25047 the same displacement was harmless and the reset still applied to in-memory state. After #25047 only `getLast()`'s runnable executes, so displacement loses the reset.

The `getLast()`-only persistence and the `callbackGroup` fan-out themselves date to the 2016 initial import (`9f849c32496`); they were safe for cursor reset only because the reset rode the fan-out.

Containment: `81aff30c4b4` is in `origin/master`, `origin/branch-4.2`, `origin/branch-5.0-M1`; backported as `08f6c337a7f` (branch-4.0) and `bfa6de4c221` (branch-4.1). `alignAcknowledgeStatusAfterPersisted` does not appear in `branch-3.0` or `branch-3.3`.

### Ruled out — things checked that do **not** protect against this

- **`RESET_CURSOR_IN_PROGRESS`.** Read in exactly one place on the write path, `asyncMarkDelete` (`:2277`); other occurrences are the declaration (`:169`), init (`:383`) and the reset's own CAS (`:1607`, `:1694`, `:1707`). Cumulative acks are correctly rejected; individual acks enter via `asyncDelete` (`:2575`, only `isClosed()` guarded) and never touch the gate.
- **`synchronized (pendingMarkDeleteOps)`.** Makes each `add` atomic; it establishes no exclusion of *later* adds while an earlier entry is queued, and there is no dedup, merge, or "reset wins" branch.
- **Subscription fencing (`IS_FENCED`).** Only read in `addConsumerInternal` (`PersistentSubscription.java:271`); it blocks re-attaching consumers. `acknowledgeMessageAsync` (`:463-516`) has no fence check at all and reaches `cursor.asyncDelete` at `:509`.
- **Consumer disconnect before the reset.** `resetCursorInternal` disconnects consumers (`:978-984`) before `cursor.asyncResetCursor` (`:1024`), but this is not a drain: `Consumer.disconnect` closes synchronously, so the future is already complete when returned.
- **`cancelPendingRead()`.** `ManagedCursorImpl.cancelPendingReadRequest()` (`:1196-1208`) only clears `WAITING_READ_OP` — a read already submitted to bookies cannot be cancelled, and the two states are disjoint. The tree documents exactly this at `PersistentReplicator.java:385-392`.
- **A later flush re-running R.** The queue is cleared at `:3278` (and again at `:1693`); R survives only inside A's `callbackGroup`, and `triggerComplete()` never calls `alignAcknowledgeStatus()`.
- **Recovery from persisted metadata.** `persistPositionToLedger` writes A's position (`:3493`) with the live `individualDeletedMessages`, so recovery reproduces the damage.
- **Ledger switch.** `createNewMetadataLedger` persists `lastMarkDeleteEntry`, which `:2447-2454` set to A, never R.
- **`rewind()` / `onCursorReadPositionUpdated`.** `rewind()` (`:2836-2858`) derives `readPosition` from the stale `markDeletePosition`, re-affirming the wrong state; `onCursorReadPositionUpdated` is cache-eviction bookkeeping only (`ManagedLedgerImpl.java:2694-2698`).
- **Dispatcher re-derivation after reset.** `dispatcher.cursorIsReset()` only nulls `lastIndividualDeletedRangeFromCursorRecovery`; nothing recomputes cursor state.

### Honest assessment of reachability

The mechanism is structurally proven; the *frequency* is not, and I want to be explicit about the narrowing factors:

- `resetCursorInternal` fences and disconnects consumers before the reset, removing the obvious source of concurrent `CommandAck` traffic.
- `throttleMarkDelete` defaults to `1.0`/s (`ServiceConfiguration.java:2634` → `BrokerService.java:2272-2274`), so `markDeleteLimiter` (`ManagedCursorImpl.java:2701`) drops most individual acks before they can enqueue an entry at all.

Ack sources that fencing does **not** cover, and that need no connected consumer:

- Transaction pending-ack commit: `ServerCnx.handleEndTxnOnSubscription` → `PersistentSubscription.endTxn` → `PendingAckHandleImpl.commitTxn` (`:531`, on its own `internalPinnedExecutor`) → `individualAckCommitCommon` (`:795-806`) → `acknowledgeMessageAsync(..., Individual)` → `cursor.asyncDelete`.
- `ManagedCursorImpl.skipNonRecoverableLedger` (`:3132-3155`) / `skipNonRecoverableEntries` (`:3165+`) under `autoSkipNonRecoverableData`, invoked from `OpReadEntry.internalReadEntriesFailed` — i.e. during a read.
- Entry-filter-driven acks in `AbstractBaseDispatcher` (`:300-334`).
- A `CommandAck` already queued on the topic's ordered executor when the disconnect loop runs.

Additionally, the `NoLedger`/`SwitchingLedger` queueing branches mean a **cursor-ledger rollover concurrent with a seek** reproduces the displacement with no read in flight at all.

### Existing test coverage

None covers this. Nearest neighbours, and why they miss it:

- `ManagedCursorTest.testCompactionCursorResetNeverLoseMarkDeleteProperties` (`:6630-6711`) — the only test interleaving a reset with a mark-delete, but it drives the mark-delete *first* (the benign ordering), never establishes `PENDING_READ_OPS > 0`, and uses cumulative `asyncMarkDelete`, so nothing is ever displaced.
- `ManagedCursorTest.testConcurrentResetCursor` (`:1047-1142`) — concurrency across different cursors, no acks.
- `SubscriptionSeekTest.testConcurrentResetCursor` (`:412`) / `testConcurrentResetCursorByTimestamp` (`:462`) — reset-vs-reset, exercising fencing only.
- `HybridTypesAcknowledgeTest` (added by #25047 itself; `:139`, `:266-275`) — ack-then-seek strictly sequentially. It validates the runnable's *contents*, never whether the runnable runs.
- Repo-wide, `**/src/test/**` has zero hits for `callbackGroup`, `alignAcknowledgeStatus`, or `RESET_CURSOR_IN_PROGRESS`.

### Out of scope, noted while reading (pre-dates #25047)

`finalCallback.operationComplete()` does an unconditional `pendingMarkDeleteOps.clear()` at `:1693` without firing callbacks. Any entry queued between the flush and the reset's persist completion is discarded with its `MarkDeleteCallback` never invoked, leaving the corresponding broker-side ack future hung. Probably worth a separate look.

## Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Start with ManagedCursorImpl.java, especially internalResetCursor, internalAsyncMarkDelete, internalFlushPendingMarkDeletes, and the persistence completion path. Read the related ManagedCursorTest.java interceptor examples and build a deterministic test around a held read, queued reset, and queued acknowledgement. Done means the test demonstrates that reset state is applied and remains correct after reopening the cursor, while the control ordering still passes.

Written by the indexing model from the issue text.