[Bug] Make DLQ transition durable before retiring source delivery

[qqeasonchen]

Execution meta — see plan: #5296

• Phase: 1 — single-instance correctness
• Severity: P0
• Depends on: (none)
• Blocks: [Architecture Review][P0] Unified state control-plane stores #5301

Problem

When a delivery exhausts retries, ReliableDispatcher.tick() removes it from the in-memory pending map and invokes DeadLetterSink. The Runtime DLQ sink publishes asynchronously to <topic>_DLQ; a send failure is logged but does not restore the original delivery, persist an outbox record, or retry the terminal transition.

For RocketMQ 5 POP, the current DLQ path intentionally does not invoke the deferred source broker ACK. After POP invisible time expires, the source message can be redelivered and dead-lettered again. If the source ACK is added without a durable DLQ write boundary, a DLQ write failure can lose the source message.

Relevant code:

• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/delivery/ReliableDispatcher.java: tick() terminal DLQ path
• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java: deadLetterSink()
• eventmesh-storage-plugin/eventmesh-storage-rocketmq5/src/main/java/org/apache/eventmesh/storage/rocketmq5/storage/RocketMQ5RemotingStoragePlugin.java: deferred POP ACK handling

Reproduction

1. Configure a low retry budget and a storage double that fails writes to <topic>_DLQ.
2. Deliver a message and allow retry exhaustion.
3. Observe that the original pending delivery is already gone despite failed DLQ persistence.
4. For RocketMQ 5 POP, allow a successful DLQ write but no source ACK, then wait beyond invisible time.
5. Observe source redelivery and duplicate DLQ records.

Expected behavior

DLQ transition must be durable and idempotent:

• A source message is terminal only after DLQ persistence succeeds.
• A failed DLQ write leaves the source retryable/recoverable.
• A POP source is broker-ACKed only after durable DLQ completion.
• Repeated source redelivery does not create duplicate logical DLQ records.

Proposed direction

• Make DeadLetterSink return a completion result rather than void.
• Introduce a durable terminal-transition/outbox record keyed by source message identity.
• Persist DLQ metadata: original topic, source message ID, physical location or POP key, attempts, failure reason, timestamps, and replay count.
• After durable DLQ completion, execute the source broker ACK exactly once where required.

Acceptance criteria

• DLQ send failure cannot silently retire the original delivery.
• The Runtime retries or durably records a failed DLQ transition.
• RocketMQ 5 POP source messages are ACKed only after successful durable DLQ write.
• A POP message that reaches DLQ does not reappear and create duplicate DLQ records after invisible time.
• DLQ replay preserves and exposes source metadata.
• Unit tests and RocketMQ 5 E2E tests cover DLQ write failure, source redelivery, idempotency, and replay.

Suggested labels

bug, reliability, storage, priority:p0

[qqeasonchen]

Fixed in #5307: DeadLetterSink.deadLetter now returns CompletableFuture<Boolean>. ReliableDispatcher.tick() retires a dead-lettered delivery only after storage confirms the DLQ write; otherwise it returns to pending and the dead-letter transition is retried. UniIngressService completes the future from the storage send callback.