feat(state): DeadLetterStore + TaskStore Meta backends (issue #5301 Sub-PR C, fixes #5292 fully, unblocks #5302)

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/delivery/ReliableDispatcher.java

@@ -20,6 +20,7 @@
 import org.apache.eventmesh.common.wire.EventMeshFrame;
 import org.apache.eventmesh.runtime.metrics.UniMetrics;
 import org.apache.eventmesh.runtime.offset.OffsetStore;
+import org.apache.eventmesh.runtime.state.DeadLetterStore;
 import org.apache.eventmesh.runtime.state.DeliveryStateStore;
 import org.apache.eventmesh.runtime.state.DeliveryStateStore.Record;
 import org.apache.eventmesh.runtime.state.InMemoryDeliveryStateStore;
@@ -83,6 +84,14 @@ public class ReliableDispatcher {
     // boots with an empty live map, so recover() retires store records WITHOUT re-invoking the
     // channel (issue #5291 idempotency).
     private final DeliveryStateStore stateStore;
+    /** Sub-PR C: durable DLQ ledger. When non-null, every confirmed DLQ transition is
+     *  recorded via {@link DeadLetterStore#recordDeadLetter} before the delivery is
+     *  retired. Null = legacy behaviour (Sub-PR A/B), the sink is the only durable
+     *  confirmation. */
+    /** Effectively final after construction. Non-final only because the 9-arg ctor chains
+     *  to the 8-arg ctor (which assigns null) and then overwrites with the supplied ledger;
+     *  after the ctor returns this field is never reassigned by the runtime. */
+    private DeadLetterStore deadLetterStore;
     private final Map<String, Delivery> liveDeliveries = new ConcurrentHashMap<>();
     private final AtomicLong deliverySeq = new AtomicLong();
     /** Process boot epoch + per-process random salt: delivery ids stay unique across restarts and
@@ -222,6 +231,20 @@ public ReliableDispatcher(long ackTimeoutMs, int maxAttempts, LongSupplier clock
         this.metrics = metrics;
         this.jitterRatio = Math.max(0.0d, jitterRatio);
         this.stateStore = stateStore;
+        this.deadLetterStore = null;
+    }
+
+    /**
+     * Sub-PR C constructor: same as the 8-arg ctor plus a {@link DeadLetterStore}
+     * that is invoked on every confirmed DLQ transition (issue #5301, fixes #5292
+     * fully). When {@code deadLetterStore} is null, the legacy Sub-PR A/B behaviour
+     * is preserved: the downstream DLQ sink is the only durable confirmation.
+     */
+    public ReliableDispatcher(long ackTimeoutMs, int maxAttempts, LongSupplier clock,
+        OffsetStore offsetStore, DeadLetterSink dlqSink, UniMetrics metrics, double jitterRatio,
+        DeliveryStateStore stateStore, DeadLetterStore deadLetterStore) {
+        this(ackTimeoutMs, maxAttempts, clock, offsetStore, dlqSink, metrics, jitterRatio, stateStore);
+        this.deadLetterStore = deadLetterStore;
     }
 
     public UniMetrics metrics() {
@@ -363,6 +386,22 @@ public int tick() {
                 dlqPersisted.whenComplete((ok, err) -> {
                     org.apache.eventmesh.runtime.metrics.UniTrace.end(dlqSpan);
                     if (Boolean.TRUE.equals(ok)) {
+                        // Sub-PR C: record the durable ledger entry (idempotent --
+                        // putIfAbsent CAS, so a peer that already recorded wins). We
+                        // do NOT block retirement on the ledger result: the downstream
+                        // DLQ topic write has already succeeded, so the message body is
+                        // safe; the ledger is the cluster-visible record for restart.
+                        if (deadLetterStore != null) {
+                            boolean ledgerOk = deadLetterStore.recordDeadLetter(
+                                rec.deliveryId, rec.topic + "_DLQ", -1L);
+                            if (!ledgerOk) {
+                                log.warn(
+                                    "DLQ ledger write failed for delivery {}; sink-side DLQ is confirmed"
+                                        + " but the cluster-wide record is not. Delivery will still be retired; a subsequent"
+                                        + " recover() will see no ledger record and retire via the offset advance (Sub-PR B).",
+                                    rec.deliveryId);
+                            }
+                        }
                         // DLQ confirmed: remove from the state store so a subsequent recover()
                         // does not retire a delivery that is already dead-lettered.
                         stateStore.remove(rec.deliveryId);

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/state/MetaBackedDeadLetterStore.java

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.state;
+
+import org.apache.eventmesh.runtime.cluster.MetaStore;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Production {@link DeadLetterStore} backed by a cluster-shared {@link MetaStore}
+ * (issue #5301 Sub-PR C, durable-egress tier, fixes #5292 fully).
+ *
+ * <p>One record per {@code deliveryId} at key {@code /em/dlq/<deliveryId>} with value
+ * {@code "<dlqTopic>:<dlqOffset>"}. The write uses {@link MetaStore#putIfAbsent(String, String)}
+ * as a CAS so the first recorder wins and a concurrent retry cannot double-record
+ * (idempotency contract inherited from {@link DeadLetterStore#recordDeadLetter}).</p>
+ *
+ * <p>The {@link #close()} method does not close the underlying {@link MetaStore}; the
+ * MetaStore is owned by the Runtime and outlives any individual store wrapper.</p>
+ */
+@Slf4j
+public class MetaBackedDeadLetterStore implements DeadLetterStore {
+
+    /** Meta key prefix for the DLQ ledger. Cluster-shared, namespace-stable. */
+    public static final String PREFIX = "/em/dlq/";
+
+    private final MetaStore meta;
+
+    public MetaBackedDeadLetterStore(MetaStore meta) {
+        this.meta = meta;
+    }
+
+    private static String key(String deliveryId) {
+        return PREFIX + deliveryId;
+    }
+
+    @Override
+    public boolean recordDeadLetter(String deliveryId, String dlqTopic, long dlqOffset) {
+        if (deliveryId == null || dlqTopic == null) {
+            return false;
+        }
+        String value = dlqTopic + ":" + dlqOffset;
+        // First-write-wins. Already-present key => idempotent success; absent key => CAS write.
+        if (meta.get(key(deliveryId)) != null) {
+            return true;
+        }
+        boolean wrote = meta.putIfAbsent(key(deliveryId), value);
+        if (!wrote) {
+            // Lost the race to a peer; the record exists now regardless — treat as success
+            // so the dispatcher proceeds to retire. Returning false would block retirement
+            // and produce a duplicate retry on the next tick.
+            log.debug("DLQ record CAS lost for deliveryId={} (peer wrote first); treating as success", deliveryId);
+            return true;
+        }
+        return true;
+    }
+
+    @Override
+    public boolean isDeadLettered(String deliveryId) {
+        if (deliveryId == null) {
+            return false;
+        }
+        return meta.get(key(deliveryId)) != null;
+    }
+
+    @Override
+    public void flush() {
+        // Meta is the source of truth; no buffered writes here.
+    }
+
+    @Override
+    public void close() {
+        // Do NOT close the underlying MetaStore — it is owned by the Runtime.
+    }
+}