fix(data-plane): accept exact replacement replays

services/external-data-plane/src/data-product-delivery.mjs

@@ -104,8 +104,6 @@ export async function persistDataProductPublish(pool, batch, definition, {
       return { ...normalizeReceipt(existing.rows[0]), idempotent: true };
     }
 
-    if (batch.batch.mode === "replace") await lockReplacementGeneration(client, batch);
-
     const records = batch.facts.map((fact) => ({
       source_id: fact.sourceId,
       semantic_type: fact.semanticType,
@@ -116,6 +114,33 @@ export async function persistDataProductPublish(pool, batch, definition, {
       fingerprint: fingerprint(fact),
     }));
 
+    const replacement = batch.batch.mode === "replace"
+      ? await lockReplacementGeneration(client, batch, records)
+      : { disposition: "advance" };
+    if (replacement.disposition === "replay") {
+      // Transport keys may change across publisher releases. Equal replacement
+      // generations are idempotent only when the complete stored fact set is exact.
+      const cursor = await currentCursor(client, batch);
+      await client.query(
+        `update external_data_plane_batches
+         set delivery_cursor = $2
+         where id = $1`,
+        [batchId, cursor],
+      );
+      await client.query("commit");
+      return normalizeReceipt({
+        batchId,
+        idempotent: true,
+        publishedFactCount: batch.facts.length,
+        currentUpdatedCount: 0,
+        currentRemovedCount: 0,
+        historyInsertedCount: 0,
+        patchOperationCount: 0,
+        cursor,
+        acceptedAt: batch.batch.receivedAt,
+      });
+    }
+
     const updated = await client.query(
       `insert into external_data_plane_current (
         tenant_id, connection_id, provider_id, data_product_id, source_id, semantic_type,
@@ -448,7 +473,7 @@ export async function pruneBatchReceipts(db, { retentionMs, limit = 10_000 }) {
   return result.rowCount;
 }
 
-async function lockReplacementGeneration(client, batch) {
+async function lockReplacementGeneration(client, batch, records) {
   const scope = [batch.source.tenantId, batch.source.connectionId, batch.source.providerId, batch.contract.dataProductId];
   await client.query(
     `insert into external_data_plane_delivery_state (
@@ -468,9 +493,59 @@ async function lockReplacementGeneration(client, batch) {
   const currentGenerationAt = state.rows[0]?.currentGenerationAt
     ? new Date(state.rows[0].currentGenerationAt)
     : null;
-  if (currentGenerationAt && generationAt <= currentGenerationAt) {
+  if (!currentGenerationAt || generationAt > currentGenerationAt) {
+    return { disposition: "advance" };
+  }
+  if (generationAt < currentGenerationAt) {
     throw deliveryError("data_product_generation_not_newer", 409);
   }
+  // The delivery-state row remains locked while the complete current snapshot
+  // is compared, so an equal-generation replay cannot race a newer replacement.
+  const comparison = await client.query(
+    `with incoming as (
+       select source_id, semantic_type, fingerprint
+       from jsonb_to_recordset($5::jsonb) as item(
+         source_id text, semantic_type text, fingerprint text
+       )
+     ), current_scope as (
+       select source_id, semantic_type, fingerprint
+       from external_data_plane_current
+       where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4
+     )
+     select
+       (select count(*)::integer from incoming) as "incomingCount",
+       (select count(*)::integer from current_scope) as "currentCount",
+       (select count(*)::integer
+        from incoming
+        join current_scope using (source_id, semantic_type, fingerprint)) as "matchingCount"`,
+    [...scope, JSON.stringify(records)],
+  );
+  const disposition = classifyReplacementGeneration({
+    currentGenerationAt,
+    generationAt,
+    incomingCount: comparison.rows[0]?.incomingCount,
+    currentCount: comparison.rows[0]?.currentCount,
+    matchingCount: comparison.rows[0]?.matchingCount,
+  });
+  if (disposition !== "replay") throw deliveryError("data_product_generation_not_newer", 409);
+  return { disposition };
+}
+
+export function classifyReplacementGeneration({
+  currentGenerationAt,
+  generationAt,
+  incomingCount = 0,
+  currentCount = 0,
+  matchingCount = 0,
+}) {
+  const incomingTime = new Date(generationAt).getTime();
+  const currentTime = currentGenerationAt ? new Date(currentGenerationAt).getTime() : null;
+  if (currentTime === null || incomingTime > currentTime) return "advance";
+  if (incomingTime < currentTime) return "stale";
+  const incoming = Number(incomingCount);
+  const current = Number(currentCount);
+  const matching = Number(matchingCount);
+  return incoming === current && matching === incoming ? "replay" : "conflict";
 }
 
 async function removeMissingCurrent(client, batch, records) {