fix(n8n): make replacement replay idempotent

2026-07-21 15:06:59 +03:00
parent bc23c550db
commit 358b259aac
10 changed files with 117 additions and 35 deletions
@@ -98,7 +98,11 @@ Each incoming item must be either a canonical fact or `{ "fact": <fact> }`:

 The node emits `nodedc.data-product.publish/v1`, enforces the 5000-fact and
 64-KiB-per-attributes limits, rejects secret-like material in attributes, and
-derives stable run/idempotency identifiers from workflow execution context.
+derives upsert run/idempotency identifiers from workflow execution context.
+Complete-snapshot replacement instead derives one canonical identity from the
+product, generation timestamp, sequence and normalized fact set. Replaying the
+same immutable generation therefore reaches the Data Plane as the same request,
+while changed content under an already-used generation remains a conflict.

 ## Build and verification

@@ -109,10 +109,6 @@ export function buildPublishPayload(
 ): NdcPublishPayload {
  if (!Number.isInteger(sequence) || sequence < 0 || sequence > 2_147_483_647) throw new Error('batch_sequence_invalid');
  requireIdentifier(dataProductId, 'dataProductId');
-  const runId = `run-${sha256(`${workflowId}|${executionId}`).slice(0, 48)}`;
-  const idempotencyKey = `publish-${sha256(
-    `${workflowId}|${executionId}|${nodeId}|${dataProductId}|${sequence}`,
-  )}`;
  const facts = factsFromItems(items, publishMode === 'replace');
  const explicitGenerationAt = items.length === 1 && Array.isArray(items[0]?.json?.facts)
    ? items[0].json.generationAt
@@ -120,6 +116,15 @@ export function buildPublishPayload(
  const generationAt = publishMode === 'replace'
    ? replacementGenerationAt(facts, explicitGenerationAt)
    : undefined;
+  const replacementDigest = generationAt
+    ? replacementReplayDigest(dataProductId, sequence, generationAt, facts)
+    : undefined;
+  const runId = replacementDigest
+    ? `run-${replacementDigest.slice(0, 48)}`
+    : `run-${sha256(`${workflowId}|${executionId}`).slice(0, 48)}`;
+  const idempotencyKey = replacementDigest
+    ? `publish-${replacementDigest}`
+    : `publish-${sha256(`${workflowId}|${executionId}|${nodeId}|${dataProductId}|${sequence}`)}`;
  return {
    schemaVersion: DATA_PRODUCT_PUBLISH_SCHEMA_VERSION,
    batch: {
@@ -332,6 +337,35 @@ function replacementGenerationAt(facts: NdcFact[], explicit: unknown): string {
  return timestamps[0];
 }

+function replacementReplayDigest(
+  dataProductId: string,
+  sequence: number,
+  generationAt: string,
+  facts: NdcFact[],
+): string {
+  const canonicalFacts = [...facts].sort((left, right) => {
+    const leftKey = `${left.sourceId}\u0000${left.semanticType}`;
+    const rightKey = `${right.sourceId}\u0000${right.semanticType}`;
+    return leftKey.localeCompare(rightKey);
+  });
+  return sha256(stableJson({
+    schemaVersion: DATA_PRODUCT_PUBLISH_SCHEMA_VERSION,
+    dataProductId,
+    mode: 'replace',
+    generationAt,
+    sequence,
+    facts: canonicalFacts,
+  }));
+}
+
+function stableJson(value: unknown): string {
+  if (Array.isArray(value)) return `[${value.map(stableJson).join(',')}]`;
+  if (isObject(value)) {
+    return `{${Object.keys(value).sort().map((key) => `${JSON.stringify(key)}:${stableJson(value[key])}`).join(',')}}`;
+  }
+  return JSON.stringify(value);
+}
+
 function requireIsoTimestamp(value: unknown, field: string): string {
  const normalized = String(value ?? '').trim();
  if (!normalized || Number.isNaN(Date.parse(normalized))) throw new Error(`${field}_invalid`);
@@ -1,12 +1,12 @@
 {
  "name": "n8n-nodes-ndc",
-  "version": "0.1.5",
+  "version": "0.1.6",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "n8n-nodes-ndc",
-      "version": "0.1.5",
+      "version": "0.1.6",
      "license": "UNLICENSED",
      "devDependencies": {
        "@n8n/node-cli": "0.39.3",
@@ -1,6 +1,6 @@
 {
  "name": "n8n-nodes-ndc",
-  "version": "0.1.5",
+  "version": "0.1.6",
  "description": "Private NODE.DC nodes for scoped data products and Foundry bindings.",
  "private": true,
  "license": "UNLICENSED",
@@ -170,6 +170,39 @@ async function main() {
  assert.equal(externalContract.validateDataProductPublish(replacement).ok, true);
  assert.equal(replacement.batch.mode, 'replace');
  assert.equal(replacement.batch.generationAt, zoneItems[0].json.observedAt);
+  const replacementReplay = contracts.buildPublishPayload(
+    zoneItems,
+    'execution-replay',
+    'workflow-other',
+    'node-other',
+    'map.zones.current.v1',
+    0,
+    'replace',
+  );
+  assert.equal(
+    replacementReplay.batch.runId,
+    replacement.batch.runId,
+    'the same complete generation must keep one run identity across executions',
+  );
+  assert.equal(
+    replacementReplay.batch.idempotencyKey,
+    replacement.batch.idempotencyKey,
+    'the same complete generation must replay through the data plane idempotently',
+  );
+  const changedReplacement = contracts.buildPublishPayload(
+    [{ json: { ...zoneItems[0].json, attributes: { display_name: 'Zone 42 changed' } } }],
+    'execution-replay',
+    'workflow-other',
+    'node-other',
+    'map.zones.current.v1',
+    0,
+    'replace',
+  );
+  assert.notEqual(
+    changedReplacement.batch.idempotencyKey,
+    replacement.batch.idempotencyKey,
+    'different content at one declared generation must not alias an accepted replay',
+  );
  const replacementBatch = contracts.buildPublishPayload(
    [{ json: { generationAt: zoneItems[0].json.observedAt, facts: zoneItems.map((item) => item.json) } }],
    'execution-44',
@@ -181,6 +214,7 @@ async function main() {
  );
  assert.deepEqual(replacementBatch.facts, replacement.facts);
  assert.equal(replacementBatch.batch.generationAt, zoneItems[0].json.observedAt);
+  assert.equal(replacementBatch.batch.idempotencyKey, replacement.batch.idempotencyKey);
  const emptyReplacement = contracts.buildPublishPayload(
    [{ json: { generationAt: '2026-07-15T13:00:00.000Z', facts: [] } }],
    'execution-45',