From a9b8d7196822d54ad58d83fb04d026b570e47390 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 21 Jul 2026 23:11:35 +0300 Subject: [PATCH] fix(data-plane): accept exact replacement replays --- ...data-plane-replacement-replay-artifact.mjs | 74 +++++++++++++++++ ..._data_plane_replacement_replay_artifact.py | 64 ++++++++++++++ .../src/data-product-delivery.mjs | 83 ++++++++++++++++++- ...data-product-delivery.integration.test.mjs | 20 +++++ .../test/data-product-delivery.test.mjs | 43 +++++++++- 5 files changed, 279 insertions(+), 5 deletions(-) create mode 100755 infra/deploy-runner/build-external-data-plane-replacement-replay-artifact.mjs create mode 100755 infra/deploy-runner/test_external_data_plane_replacement_replay_artifact.py diff --git a/infra/deploy-runner/build-external-data-plane-replacement-replay-artifact.mjs b/infra/deploy-runner/build-external-data-plane-replacement-replay-artifact.mjs new file mode 100755 index 0000000..b2a3049 --- /dev/null +++ b/infra/deploy-runner/build-external-data-plane-replacement-replay-artifact.mjs @@ -0,0 +1,74 @@ +#!/usr/bin/env node +import { createHash } from "node:crypto"; +import { spawnSync } from "node:child_process"; +import { cp, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +const scriptDir = dirname(fileURLToPath(import.meta.url)); +const platformRoot = resolve(scriptDir, "../.."); +const artifactDir = resolve(process.env.NODEDC_DEPLOY_ARTIFACT_DIR || join(scriptDir, "../deploy-artifacts")); +const [patchId = "external-data-plane-replacement-replay-v1-20260721-014", ...extra] = process.argv.slice(2); +const sourceRelative = "services/external-data-plane/src/data-product-delivery.mjs"; +const destinationRelative = `platform/${sourceRelative}`; + +if (extra.length || !/^[A-Za-z0-9._-]{1,96}$/.test(patchId)) { + throw new Error("usage: build-external-data-plane-replacement-replay-artifact.mjs [patch-id]"); +} + +const source = resolve(platformRoot, sourceRelative); +const sourceText = await readFile(source, "utf8"); +for (const marker of [ + 'replacement.disposition === "replay"', + "classifyReplacementGeneration", + 'throw deliveryError("data_product_generation_not_newer", 409)', + 'join current_scope using (source_id, semantic_type, fingerprint)', +]) { + if (!sourceText.includes(marker)) throw new Error(`replacement_replay_boundary_missing:${marker}`); +} + +const stage = await mkdtemp(join(tmpdir(), "nodedc-edp-replacement-replay-")); +const payloadPath = join(stage, "payload", destinationRelative); +const target = join(artifactDir, `nodedc-platform-${patchId}.tgz`); + +try { + await mkdir(dirname(payloadPath), { recursive: true }); + await cp(source, payloadPath, { force: false }); + await writeFile(join(stage, "manifest.env"), `id=${patchId}\ncomponent=platform\ntype=app-overlay\n`, "utf8"); + await writeFile(join(stage, "files.txt"), `${destinationRelative}\n`, "utf8"); + await mkdir(artifactDir, { recursive: true }); + const tar = spawnSync("python3", ["-c", canonicalTarScript(), target, stage], { + encoding: "utf8", + maxBuffer: 32 * 1024 * 1024, + }); + if (tar.status !== 0) throw new Error(`tar_failed:${tar.stderr || tar.stdout}`); + const sha256 = createHash("sha256").update(await readFile(target)).digest("hex"); + console.log(JSON.stringify({ + ok: true, + patchId, + artifact: target, + sha256, + entries: [destinationRelative], + service: "external-data-plane", + database: "preserved", + }, null, 2)); +} finally { + await rm(stage, { recursive: true, force: true }); +} + +function canonicalTarScript() { + return [ + "import gzip, io, pathlib, sys, tarfile", + "root=pathlib.Path(sys.argv[2])", + "with open(sys.argv[1], 'wb') as out:", + " with gzip.GzipFile(filename='', mode='wb', fileobj=out, compresslevel=9, mtime=0) as gz:", + " with tarfile.open(fileobj=gz, mode='w', format=tarfile.PAX_FORMAT) as tar:", + " for top in ('manifest.env','files.txt','payload'):", + " p=root/top; paths=[p] + (sorted(p.rglob('*')) if p.is_dir() else [])", + " for x in paths:", + " info=tar.gettarinfo(str(x), arcname=x.relative_to(root).as_posix())", + " info.uid=info.gid=0; info.uname=info.gname='root'; info.mtime=0; info.mode=0o755 if info.isdir() else 0o644", + " with (open(x,'rb') if info.isfile() else io.BytesIO()) as src: tar.addfile(info, src if info.isfile() else None)", + ].join("\n"); +} diff --git a/infra/deploy-runner/test_external_data_plane_replacement_replay_artifact.py b/infra/deploy-runner/test_external_data_plane_replacement_replay_artifact.py new file mode 100755 index 0000000..94d3d5c --- /dev/null +++ b/infra/deploy-runner/test_external_data_plane_replacement_replay_artifact.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +import hashlib +import json +import os +import subprocess +import tarfile +import tempfile +import unittest +from pathlib import Path + + +SCRIPT_DIR = Path(__file__).resolve().parent +PLATFORM_ROOT = SCRIPT_DIR.parent.parent +BUILDER = SCRIPT_DIR / "build-external-data-plane-replacement-replay-artifact.mjs" +PATCH_ID = "external-data-plane-replacement-replay-v1-20260721-014" +ENTRY = "platform/services/external-data-plane/src/data-product-delivery.mjs" + + +class ExternalDataPlaneReplacementReplayArtifactTest(unittest.TestCase): + def build(self, artifact_dir): + environment = os.environ.copy() + environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir) + result = subprocess.run( + ["node", str(BUILDER), PATCH_ID], + check=True, + capture_output=True, + text=True, + env=environment, + ) + return json.loads(result.stdout) + + def test_exact_runtime_slice_is_deterministic(self): + with tempfile.TemporaryDirectory(prefix="nodedc-edp-replay-artifact-") as directory: + artifact_dir = Path(directory) + first = self.build(artifact_dir) + first_bytes = Path(first["artifact"]).read_bytes() + second = self.build(artifact_dir) + second_bytes = Path(second["artifact"]).read_bytes() + + self.assertEqual(first["entries"], [ENTRY]) + self.assertEqual(first["service"], "external-data-plane") + self.assertEqual(first["database"], "preserved") + self.assertEqual(first["sha256"], hashlib.sha256(first_bytes).hexdigest()) + self.assertEqual(first["sha256"], second["sha256"]) + self.assertEqual(first_bytes, second_bytes) + + with tarfile.open(first["artifact"], "r:gz") as archive: + names = archive.getnames() + manifest = archive.extractfile("manifest.env").read().decode("utf-8") + files = archive.extractfile("files.txt").read().decode("utf-8") + payload = archive.extractfile(f"payload/{ENTRY}").read() + + self.assertEqual(manifest, f"id={PATCH_ID}\ncomponent=platform\ntype=app-overlay\n") + self.assertEqual(files, f"{ENTRY}\n") + self.assertIn(f"payload/{ENTRY}", names) + self.assertEqual( + payload, + (PLATFORM_ROOT / "services/external-data-plane/src/data-product-delivery.mjs").read_bytes(), + ) + self.assertFalse(any("/test/" in name or "/node_modules/" in name for name in names)) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/services/external-data-plane/src/data-product-delivery.mjs b/services/external-data-plane/src/data-product-delivery.mjs index 1f749d5..10908ff 100644 --- a/services/external-data-plane/src/data-product-delivery.mjs +++ b/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) { diff --git a/services/external-data-plane/test/data-product-delivery.integration.test.mjs b/services/external-data-plane/test/data-product-delivery.integration.test.mjs index 09df3c8..f48b12f 100644 --- a/services/external-data-plane/test/data-product-delivery.integration.test.mjs +++ b/services/external-data-plane/test/data-product-delivery.integration.test.mjs @@ -232,6 +232,26 @@ try { ); assert.equal(firstZoneGeneration.currentUpdatedCount, 2); assert.equal(firstZoneGeneration.currentRemovedCount, 0); + const replayedFirstZoneGeneration = await persistDataProductPublish( + pool, + zoneBatch("zones-01-stable-replay-key", firstZoneGenerationAt, [ + zoneFact("zone-01", firstZoneGenerationAt), + zoneFact("zone-02", firstZoneGenerationAt), + ]), + storedZoneDefinition, + ); + assert.equal(replayedFirstZoneGeneration.idempotent, true); + assert.equal(replayedFirstZoneGeneration.publishedFactCount, 2); + assert.equal(replayedFirstZoneGeneration.currentUpdatedCount, 0); + assert.equal(replayedFirstZoneGeneration.currentRemovedCount, 0); + assert.equal(replayedFirstZoneGeneration.patchOperationCount, 0); + assert.equal(replayedFirstZoneGeneration.cursor, firstZoneGeneration.cursor); + await assert.rejects( + persistDataProductPublish(pool, zoneBatch("zones-01-conflicting-replay", firstZoneGenerationAt, [ + zoneFact("zone-01", firstZoneGenerationAt), + ]), storedZoneDefinition), + (error) => error?.status === 409 && error?.code === "data_product_generation_not_newer", + ); const secondZoneGenerationAt = "2026-07-15T10:11:00.000Z"; const secondZoneGeneration = await persistDataProductPublish( pool, diff --git a/services/external-data-plane/test/data-product-delivery.test.mjs b/services/external-data-plane/test/data-product-delivery.test.mjs index 2c8b47e..e84dad1 100644 --- a/services/external-data-plane/test/data-product-delivery.test.mjs +++ b/services/external-data-plane/test/data-product-delivery.test.mjs @@ -1,6 +1,9 @@ import assert from "node:assert/strict"; -import { assertPublishMatchesDefinition } from "../src/data-product-delivery.mjs"; +import { + assertPublishMatchesDefinition, + classifyReplacementGeneration, +} from "../src/data-product-delivery.mjs"; import { normalizeDataProductDefinition } from "../src/data-product-policy.mjs"; const definition = normalizeDataProductDefinition({ @@ -92,4 +95,42 @@ assert.doesNotThrow(() => assertPublishMatchesDefinition({ }], }, legacyDefinition)); +const acceptedGeneration = "2026-07-15T10:10:00.000Z"; +assert.equal(classifyReplacementGeneration({ + currentGenerationAt: null, + generationAt: acceptedGeneration, +}), "advance"); +assert.equal(classifyReplacementGeneration({ + currentGenerationAt: acceptedGeneration, + generationAt: "2026-07-15T10:11:00.000Z", +}), "advance"); +assert.equal(classifyReplacementGeneration({ + currentGenerationAt: acceptedGeneration, + generationAt: acceptedGeneration, + incomingCount: 903, + currentCount: 903, + matchingCount: 903, +}), "replay"); +assert.equal(classifyReplacementGeneration({ + currentGenerationAt: acceptedGeneration, + generationAt: acceptedGeneration, + incomingCount: 903, + currentCount: 903, + matchingCount: 902, +}), "conflict"); +assert.equal(classifyReplacementGeneration({ + currentGenerationAt: acceptedGeneration, + generationAt: acceptedGeneration, + incomingCount: 902, + currentCount: 903, + matchingCount: 902, +}), "conflict"); +assert.equal(classifyReplacementGeneration({ + currentGenerationAt: acceptedGeneration, + generationAt: "2026-07-15T10:09:00.000Z", + incomingCount: 903, + currentCount: 903, + matchingCount: 903, +}), "stale"); + console.log("external-data-plane data product delivery policy: ok");