[Bug] Geo-replication can silently drop a message when a head-of-queue send failure rewinds the cursor after later pipelined messages already persisted

## Summary

In geo-replication with the V2 source-position deduplication path, a specific interleaving on the single FIFO replication producer (`GeoReplicationProducerImpl`) can cause a message to be **permanently lost on the destination while being mark-deleted on the source as if it had been replicated** — i.e. silent data loss, with no error surfaced.

The trigger is the conjunction of (a) a per-message *head-of-queue* send failure that fails **only** the head pending op and does **not** close the connection, while (b) later, higher-source-position messages already pipelined behind it get persisted at the remote. The replicator then rewinds its cursor and re-sends the head entry; by then the remote dedup high-watermark has advanced past it, so the re-sent entry is classified as a duplicate, dropped, and acked back as success — which advances the source mark-delete past a message that was never persisted at the destination.

This is a pre-existing issue on `master` and is independent of #26106 (see Scope below).

## Interaction timeline

Scenario: the replicator reads a batch and sends `E3`, `E4`, `E5` (source positions `L:3`, `L:4`, `L:5`, same source ledger `L`) in order on the single FIFO producer; `E3` is the head of `pendingMessages`. Time flows top‑to‑bottom; each column is an actor.

| t | Source replicator (cursor / mark-delete) | Source producer `GeoReplicationProducerImpl` (FIFO `pendingMessages`) | Remote broker (publish + dedup watermark) | Remote ledger |
| --- | --- | --- | --- | --- |
| 1 | read batch `E3,E4,E5`; tag `MSG_PROP_REPL_SOURCE_POSITION` | enqueue `E3(head),E4,E5` | | |
| 2 | | send `E3` then `E4,E5` in order | | |
| 3 | | | `E3`: **pre-dedup** reject (`ChecksumError`), connection stays **open** | `E3` *not* persisted |
| 4 | | | `E4,E5`: `NotDup` → persist; watermark → `L:5` | `E4,E5` persisted |
| 5 | | `recoverChecksumError` (local buffer corrupted): remove **only** head `E3`, fail its callback | | |
| 6 | `sendComplete(E3, error)` → `doRewindCursor(false)` | | | |
| 7 | `E4,E5` acked (Case‑3) → `asyncDelete(E4)`, `asyncDelete(E5)`; **mark-delete stays before `E3`** | | | |
| 8 | reads resume; rewind re-reads **only `E3`** (`E4,E5` already deleted) | re-send `E3'` (`L:3`) | | |
| 9 | | | `E3'` `L:3` ≤ watermark `L:5` → **`Dup`** → drop; ack success (target `-1/-1`) | `E3` **still not persisted = GAP** |
| 10 | | `E3'` ack → `ackReceivedReplicatedMsg` Case‑1 (`≤ lastPersisted`) → `sendComplete(null)` = success | | |
| 11 | `asyncDelete(E3)`; **mark-delete advances past `E3` = SILENT DATA LOSS** | | | |

```mermaid
sequenceDiagram
autonumber
participant R as Source replicator
(cursor / mark-delete)
participant P as GeoReplicationProducerImpl
(FIFO pendingMessages)
participant B as Remote broker
(publish plus dedup watermark)
participant L as Remote ledger

R->>P: read and tag E3(L:3), E4(L:4), E5(L:5), send in order
P->>B: send E3 (head)
B-->>P: ChecksumError (pre-dedup, transient), connection stays open
P->>B: send E4, E5
B->>L: persist E4, E5
B-->>P: ack E4, E5 (persisted watermark becomes L:5)
Note over P: recoverChecksumError (local buffer corrupted),
remove ONLY head E3 and fail its callback
P-->>R: sendComplete(E3, error), then doRewindCursor(false)
R->>R: asyncDelete(E4), asyncDelete(E5),
mark-delete stays before E3
R->>P: rewind re-reads E3 only, re-send E3-retry (L:3)
P->>B: send E3-retry (L:3)
Note over B: L:3 at or below persisted watermark L:5, classified Dup
B-->>P: ack success (Dup-drop, target -1/-1),
E3 never persisted = GAP
Note over P: ackReceivedReplicatedMsg Case-1 (pos at or below lastPersisted),
sendComplete(null) means success
P-->>R: success
R->>R: asyncDelete(E3), mark-delete advances past E3
= SILENT DATA LOSS
```

## Impact / Severity

- **Impact: high** — silent, permanent, unrecoverable data loss in geo-replication for the affected message(s). The source considers the message replicated; nothing logs an error.
- **Probability: low** — the only path that completes the full silent-loss chain is rare (see Trigger conditions). The far more likely observable symptom of these head-only failure paths is a *recoverable* replication stall with a destination gap, not silent loss.
- **Net operational risk: low-to-medium.** Worth a defensive fix because the failure mode, when it does occur, is silent and unrecoverable.

This is a normal reliability/correctness bug, not a security vulnerability.

## Root cause: step-by-step chain (verified against `master`)

Scenario: replicator reads a batch and sends entries E3, E4, E5 (source positions `L:3`, `L:4`, `L:5`, same source ledger `L`) in order on the single producer. E3 is the head of `pendingMessages`.

**1. FIFO send + source-position tagging.**
`GeoPersistentReplicator.replicateEntries` tags each message with `MSG_PROP_REPL_SOURCE_POSITION = ":"` and sends them in order, one `PersistentReplicator.ProducerSendCallback` per entry.
- `pulsar-broker/.../persistent/GeoPersistentReplicator.java:293-303`

Batching is disabled on the replicator producer, so each entry is its own `OpSendMsg` with an independent source position and checksum:
- `pulsar-broker/.../persistent/AbstractReplicator.java:156` (`enableBatching(false)`)

**2. Head-only failure that leaves the connection OPEN.**
`ClientCnx.handleSendError` routes exactly two server errors to per-message recovery without closing the connection; every other error hits the `default` branch which closes the connection (note the comment at line 880):
- `pulsar-client/.../impl/ClientCnx.java:869-882` — `ChecksumError -> recoverChecksumError`, `NotAllowedError -> recoverNotAllowedError`, `default -> connectionClosed(...)`.

`recoverNotAllowedError` removes **only** the head op and fails **only** its callback (no resend, no connection close), leaving E4/E5 in flight:
- `pulsar-client/.../impl/ProducerImpl.java:1510-1527`

`recoverChecksumError`, **when the local buffer is corrupted** (`verifyLocalBufferIsNotCorrupted(op) == false`), removes **only** the head op and fails its callback, then returns (no resend, no close):
- `pulsar-client/.../impl/ProducerImpl.java:1465-1493`
- Note the non-corrupted branch instead resends **all** pending via `resendMessages(...)` (`ProducerImpl.java:1507`), which produces no gap.

On the broker, these per-message errors are emitted **before** the dedup check, so the head entry never advances the dedup watermark while E4/E5 do:
- `NotAllowedError`: `PersistentTopic.publishMessage` rejects max-message-size at `pulsar-broker/.../persistent/PersistentTopic.java:659-663` and max-delivery-delay at `:664-671`, both **before** `messageDeduplication.isDuplicate(...)` at `:673-674`.
- `ChecksumError`: `Producer.checkAndStartPublish` fails checksum verification and sends `ServerError.ChecksumError` pre-publish/pre-dedup (`pulsar-broker/.../service/Producer.java`, ~`:259-266`).

So E4/E5 proceed to dedup as `NotDup` (`isDuplicateReplV2`) and persist; the persisted watermark advances to `(L,5)`.

**3. Failure rewinds the cursor; success deletes the entry.**
`PersistentReplicator.ProducerSendCallback.sendComplete` rewinds the cursor on any non-`InvalidMessageException` error, and on success deletes the entry:
- `pulsar-broker/.../persistent/PersistentReplicator.java:447-494` — error branch: `beforeTerminateOrCursorRewinding(Failed_Publishing)` + `doRewindCursor(false)` (lines 462-463); success branch: `cursor.asyncDelete(entry.getPosition(), ...)` (line 472).
- `NotAllowedException` / `ChecksumException` are siblings of (not subclasses of) `InvalidMessageException`, so the rewind fires for both.

E4/E5 were individually deleted, but mark-delete stays before E3; the rewind re-reads only E3 (E4/E5 skipped as already-deleted) and re-sends it as E3'.

**4. Remote drops the re-sent lower position as a duplicate.**
`MessageDeduplication.isDuplicateReplV2` keeps a **single per-producer high-watermark** (keys `producerName + "_LID"` / `"_EID"`), not a per-entry/gap record:
- `pulsar-broker/.../persistent/MessageDeduplication.java:389-457`. The Dup classification is a two-level check: the message must first fall at-or-below `highestSequencedPushed` (lines 411-414) **and then** at-or-below `highestSequencedPersisted` (lines 439-443) to return `MessageDupStatus.Dup`. E3' (`L:3`) satisfies both against the `(L,5)` watermark → `Dup`.
- The persisted watermark is set with an **unconditional `put`** (not a max), carrying no record of skipped/never-persisted lower positions: `MessageDeduplication.java:531-553` (`recordMessagePersistedRepl`, lines 550-551).

`PersistentTopic` acks a `Dup` as success: `publishContext.completed(null, -1, -1)`:
- `pulsar-broker/.../persistent/PersistentTopic.java:679-683`.

E3 is therefore **never persisted at the destination = permanent gap**. The dedup cannot distinguish "never-persisted lower position" from "already-persisted replay".

**5. Source mark-deletes the lost message.**
The success `SendReceipt` hits `GeoReplicationProducerImpl.ackReceivedReplicatedMsg` Case-1 ("pending send is repeated due to repl cursor rewind"), because the pending op's source position (`L:3`) is `<= lastPersistedSource` (`L:5`):
- `pulsar-client/.../impl/GeoReplicationProducerImpl.java:114-141` (condition at 118-120; `removeAndApplyCallback(op, ..., false)` at 138).
- `removeAndApplyCallback` calls `op.sendComplete(null)` (success) regardless of whether the remote actually persisted: `GeoReplicationProducerImpl.java:260-283`.
- `lastPersistedSource` reached `(L,5)` via the expected Case-3 path from E4/E5: `GeoReplicationProducerImpl.java:159-172` (lines 168-169).

`op.sendComplete(null)` flows through to the broker-side `ProducerSendCallback.sendComplete(exception == null)` → `cursor.asyncDelete(E3)` (step 3 success branch). The source mark-delete advances past the never-persisted E3 = **silent data loss**.

There is no guard that (a) checks whether later-pipelined messages already persisted before rewinding on a head failure, or (b) distinguishes a real persist from a dedup-drop on the success ack (the `SendReceipt` carries target ledgerId/entryId `-1/-1`, forwarded verbatim, so the source has no distinguishing signal either).

## Trigger conditions / how rare

All of the following must hold:

1. Geo-replication with V2 replicated dedup active on the destination (`publishContext.supportsReplDedupByLidAndEid()`, dedup enabled).
2. A pipelined batch where the **head** entry fails with an error routed to head-only recovery without closing the connection — only `ChecksumError -> recoverChecksumError` or `NotAllowedError -> recoverNotAllowedError`.
3. Later entries (higher source positions, same source ledger) get persisted at the remote, advancing the persisted watermark past the head position.
4. The head failure is **pre-dedup** (so the head never records a watermark) **and transient** (so the re-read/re-sent head passes the same broker-side check on retry and reaches dedup).

**Reachability of the two candidate errors differs sharply — this is the weakest link, flagged honestly:**

- **`NotAllowedError` does NOT produce silent loss.** Its geo-reachable triggers (max message size at `PersistentTopic.java:659-663`, max delivery delay at `:664-671`) are evaluated before dedup **and are deterministic** message properties. The rewound/re-read entry re-fails the identical pre-dedup check, never reaches dedup, and is never classified `Dup`. The outcome is a replication **stall/loop** with a destination gap (recoverable; backlog grows; visible). Actual data loss there requires an operator to force-skip the stuck entry. (The other `NotAllowedError` cases in `Producer.checkAndStartPublish` require a shadow topic / message position and are not reachable for a normal geo-replicator.)
- **`ChecksumError` is the only path that completes the full silent-loss chain**, and it additionally requires the client's **local in-memory buffer** of the head op to be genuinely corrupted (`recoverChecksumError` corrupted branch, `ProducerImpl.java:1474-1493`). Wire-corruption checksum errors with an intact local buffer take the non-corrupted branch and resend all pending (`resendMessages`, `ProducerImpl.java:1507`), producing no gap. Because the corruption is transient, the re-read entry carries a fresh valid buffer, passes broker checksum verification, reaches dedup, and is dropped as `Dup`. This demands genuine memory/bit corruption hitting exactly the head op — extremely rare.

Net: the chain is code-valid end-to-end, but the only fully-completing trigger in practice is `ChecksumError` under local client-buffer corruption (or an operator force-skipping a stalled `NotAllowedError` entry). This analysis is static; it has not been reproduced at runtime.

## Scope: rewind paths and relation to #26106

- The bug lives on `master`'s existing rewind paths: the failed-publish rewind (`PersistentReplicator.ProducerSendCallback.sendComplete` error branch → `doRewindCursor`) and, more generally, any path that rewinds the cursor while later pipelined messages have already persisted.
- **Independent of #26106.** #26106 fixes a replication *stall* when a cursor rewind races an already-dispatched in-flight read; its "Note on semantics" only claims at-least-once delivery with bounded duplicates and does not address this head-only-failure/dedup-drop gap, nor does it introduce it. However, #26106 may **extend the reach**: by letting the previously-stalling in-flight-read sub-case resume reads instead of stalling, it adds another route into the same unconditional rewind-on-head-failure logic, which lacks a guard against later-already-persisted entries.
- The V2 source-position dedup mechanism was introduced in #23697.
- Closest existing reports cover *different* failures: #25861 / PR #25860 describe the opposite symptom (V2 watermark loss after target topic reload / producer-inactivity cleanup causing **duplicates**); #26106 fixes a stall. No existing issue covers this head-only-failure silent-gap scenario.

## Suggested fix direction

The core defect is that a cursor rewind on a head-only per-message failure can advance the source mark-delete past an entry that was never persisted at the destination, because (a) the source cannot distinguish a dedup-drop ack from a real persist, and (b) the remote dedup keeps only a monotonic high-watermark with no gap record. Possible directions (to be discussed on dev@):

- Avoid the unconditional rewind-on-head-failure when later entries in the same in-flight batch have already been persisted (i.e. do not re-send a position that would now be dropped as a duplicate by the remote). At minimum, surface/fail the replication rather than silently advancing the cursor.
- On the source side, distinguish a genuine persist from a dedup-drop ack before deleting the source entry. The `SendReceipt` for a dedup-dropped message carries target ledgerId/entryId `-1/-1`; the source could treat a Case-1 ("repeated due to cursor rewind") ack with a sentinel target position as "not confirmed persisted" and refuse to mark-delete past it, instead retrying/stalling visibly.
- For the head-only checksum recovery path, reconsider dropping the head when the local buffer is corrupted (it produces a silent gap); failing the producer/replicator visibly would be safer than a silent drop.

A deterministic test (hold E4/E5 persistence while forcing a transient head-only failure on E3, then assert E3 is present on the destination and not mark-deleted on the source) would convert this from static-confirmed to runtime-proven before/with a fix.

## Affected branch

`master` (static code analysis; all `file:line` references above verified against current `master`). Not runtime-reproduced.