From bd8ef16d964a0ba33a9e9c61c7a69b1cf971cfb6 Mon Sep 17 00:00:00 2001 From: Bojan Date: Fri, 10 Jul 2026 15:17:49 +0200 Subject: [PATCH] =?UTF-8?q?fix(publish):=20bounded=20grace=20for=20the=20p?= =?UTF-8?q?ost-confirm=20tail=20=E2=80=94=20stop=20gating=20the=20publish?= =?UTF-8?q?=20response=20on=20store-queue=20congestion=20(#1572)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After a V10 publish confirms on-chain, the caller's result (ual/kaId/ txHash/status) is complete, but the sync response still awaited a serial tail of store mutations with no deadline: the SWM cleanup sweep in publisher.publishFromSharedMemory, the durable keep-root stamps after the finalization broadcast, and the _meta lifecycle tail (VM-pointer stamps, memoryLayer/state flips, publishedUal, clearSwmShareComplete) in publishFromFinalizedAssertion. On a node whose store queue is congested (large accumulated graphs + sustained inbound ACK/SWM traffic) that tail takes minutes: measured on mainnet cores as confirm at +4s but the response at +9m11s / +14m42s / +17m48s (growing per KA), with the TCP connection dying at ~12min — every client times out on a publish that actually succeeded, and peers don't learn about the KA until the equally delayed finalization broadcast. Reproduced causally on an idle edge: saturating the store queue turned a 14s response into no-response-> transport-death while the publish still confirmed on-chain; releasing the load restored seconds. See #1572 for the full evidence. The fix keeps every tail step intact — same order, same code, exactly once, error propagation preserved — but awaits each segment through awaitTailWithGrace (new, dkg-core): if the segment settles within DKG_PUBLISH_TAIL_GRACE_MS (default 5000, 0 = fully detached) behavior is byte-identical to before, including rethrow of in-grace failures; if the grace elapses the response proceeds and the segment finishes detached, its eventual settlement logged (failure at WARN) instead of becoming an unhandled rejection. On a healthy node the grace never elapses. The finalization broadcast now also goes out within seconds of the confirm instead of after the cleanup sweep, so cross-node visibility of fresh KAs is no longer delayed by minutes. Verified: new unit tests for the grace helper (completion, in-grace rethrow, detach timing, post-grace failure routing, grace=0); core 1185/1185, publisher 268/268, agent 432/432 green; live A/B on a Base-Sepolia edge under artificial store-queue saturation — stock build: confirmed on-chain, response never delivered (>300s); this fix: response delivered with all three grace detaches logged and the tail completing in the background. Fixes #1572 Co-Authored-By: Claude Fable 5 --- packages/agent/src/dkg-agent-publish.ts | 43 +++++++++- packages/core/src/index.ts | 1 + packages/core/src/publish-tail-grace.ts | 81 +++++++++++++++++++ packages/core/test/publish-tail-grace.test.ts | 74 +++++++++++++++++ packages/publisher/src/dkg-publisher.ts | 31 +++++-- 5 files changed, 222 insertions(+), 8 deletions(-) create mode 100644 packages/core/src/publish-tail-grace.ts create mode 100644 packages/core/test/publish-tail-grace.test.ts diff --git a/packages/agent/src/dkg-agent-publish.ts b/packages/agent/src/dkg-agent-publish.ts index 1ccb0ece7..cf49ac9d0 100644 --- a/packages/agent/src/dkg-agent-publish.ts +++ b/packages/agent/src/dkg-agent-publish.ts @@ -99,6 +99,8 @@ import { withSpan, getMetrics, assertQuadLiteralsMutf8Safe, + awaitTailWithGrace, + resolvePublishTailGraceMs, } from '@origintrail-official/dkg-core'; import { SpanStatusCode } from '@opentelemetry/api'; import { GraphManager, PrivateContentStore, createTripleStore, loadSelectedSharedMemoryQuads, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig } from '@origintrail-official/dkg-storage'; @@ -4453,6 +4455,14 @@ export class PublishMethods extends DKGAgentBase { } } + // GH #1572: everything from here to the return is eventual `_meta` + // bookkeeping (VM-pointer stamps, memoryLayer/state flips, publishedUal, + // assertion-graph re-stamps, swmShareComplete clearing) — the caller's + // result is already complete. Run it with a bounded grace so a congested + // store queue cannot hold the publish response hostage; on a healthy node + // it finishes inside the grace and behavior is unchanged, including error + // propagation. + const metaTail = (async () => { // OT-RFC-43 A2 (decision 2) — stamp the VM pointer on the lifecycle URN // whenever the publish/update is confirmed. (For the mint path this is the // first VM pointer; for the update path the DELETE/INSERT above already set @@ -4595,6 +4605,21 @@ export class PublishMethods extends DKGAgentBase { ); } } + })(); + { + const tailCtx = opts?.operationCtx ?? createOperationContext('publishFromSWM'); + const graceMs = resolvePublishTailGraceMs(); + const outcome = await awaitTailWithGrace(graceMs, metaTail, (err) => { + if (err !== undefined) { + this.log.warn(tailCtx, `Detached publish metadata tail failed for <${lifecycleUri}> (GH #1572): ${err instanceof Error ? err.message : String(err)}`); + } else { + this.log.info(tailCtx, `Detached publish metadata tail completed for <${lifecycleUri}> (GH #1572)`); + } + }); + if (outcome === 'detached') { + this.log.info(tailCtx, `Publish metadata tail still running after ${graceMs}ms grace for <${lifecycleUri}> — responding now, tail continues detached (GH #1572)`); + } + } return { ...result, assertionUri, seal }; } @@ -5095,7 +5120,10 @@ export class PublishMethods extends DKGAgentBase { // reconcile path can mirror the gossip dual-write decision. Read back by // `FinalizationHandler.getKeepRootCopySignal`. Updates reuse a root // entity, so replace any prior value rather than accumulate. - try { + // GH #1572: this durable stamp is a reconcile backstop for subscribers + // that missed the (already-sent) gossip broadcast — eventual by design, + // so it must not gate the publish response on a congested store queue. + const keepRootStamp = (async () => { const gm = new GraphManager(this.store); const wsMetaGraph = options?.subGraphName ? gm.sharedMemoryMetaUri(contextGraphId, options.subGraphName) @@ -5114,8 +5142,19 @@ export class PublishMethods extends DKGAgentBase { graph: wsMetaGraph, }]); } + })(); + const keepRootOnSettled = (err?: unknown) => { + if (err !== undefined) { + this.log.warn(ctx, `Failed to persist keepRootCopyOnLabel signal for ${result.ual}: ${err instanceof Error ? err.message : String(err)}`); + } + }; + try { + const graceMs = resolvePublishTailGraceMs(); + if (await awaitTailWithGrace(graceMs, keepRootStamp, keepRootOnSettled) === 'detached') { + this.log.info(ctx, `keepRootCopyOnLabel stamp still running after ${graceMs}ms grace for ${result.ual} — continuing detached (GH #1572)`); + } } catch (err) { - this.log.warn(ctx, `Failed to persist keepRootCopyOnLabel signal for ${result.ual}: ${err instanceof Error ? err.message : String(err)}`); + keepRootOnSettled(err); } } diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 3c8a6f0f0..e6ceb75d7 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -2,6 +2,7 @@ export * from './types.js'; export * from './constants.js'; export * from './assertion-scoped-graphs.js'; export * from './protocol-limits.js'; +export * from './publish-tail-grace.js'; export * from './catalog.js'; export { parseDotenvValue } from './dotenv.js'; export * from './memory-model.js'; diff --git a/packages/core/src/publish-tail-grace.ts b/packages/core/src/publish-tail-grace.ts new file mode 100644 index 000000000..584f0b0f2 --- /dev/null +++ b/packages/core/src/publish-tail-grace.ts @@ -0,0 +1,81 @@ +/** + * Bounded grace for post-confirm publish "tails" (GH #1572). + * + * After a V10 publish confirms on-chain, the caller-facing result + * (ual / kaId / txHash / status) is complete — but the code still runs a + * serial tail of store mutations (SWM cleanup sweeps, `_meta` lifecycle + * stamps). Every one of those mutations is submitted to the store queue + * with no deadline, so on a node whose store queue is congested (large + * accumulated graphs + sustained inbound ACK/SWM traffic) the tail — and + * therefore the HTTP response that awaits it — can take many minutes. + * Measured on mainnet cores: confirm at +4s, response at +9..18 min and + * growing, with the TCP connection dying at ~12 min. The same publish on + * an idle node completes the tail in well under a second. + * + * `awaitTailWithGrace` keeps the tail's semantics intact while unhooking + * it from the response path: + * - the tail always runs to completion, in order, exactly once; + * - if it settles within `graceMs`, behavior is byte-identical to a plain + * `await` (including error propagation to the caller); + * - if it is still running after `graceMs`, the caller proceeds (the + * response goes out) and the tail continues detached; its eventual + * settlement is reported through `onDetachedSettled` so a failure is + * logged rather than becoming an unhandled rejection. + * + * On a healthy node the grace never elapses, so this changes nothing. + */ + +export const DEFAULT_PUBLISH_TAIL_GRACE_MS = 5_000; + +/** + * Resolve the grace budget from `DKG_PUBLISH_TAIL_GRACE_MS`. Invalid or + * missing values fall back to {@link DEFAULT_PUBLISH_TAIL_GRACE_MS}; `0` + * is honored (respond immediately, tail fully detached). + */ +export function resolvePublishTailGraceMs(env: NodeJS.ProcessEnv = process.env): number { + const raw = env.DKG_PUBLISH_TAIL_GRACE_MS; + if (raw === undefined || raw === '') return DEFAULT_PUBLISH_TAIL_GRACE_MS; + const n = Number(raw); + if (!Number.isFinite(n) || n < 0) return DEFAULT_PUBLISH_TAIL_GRACE_MS; + return Math.floor(n); +} + +const TAIL_GRACE_TIMEOUT: unique symbol = Symbol('publish-tail-grace-timeout'); + +/** + * Await `tail` for up to `graceMs`. + * + * Returns `'completed'` when the tail settled in time — a rejection inside + * the grace window is rethrown to the caller, exactly like a plain `await`. + * Returns `'detached'` when the grace elapsed first; the still-running tail + * keeps executing and `onDetachedSettled` fires exactly once when it + * eventually settles (with the error when it failed, `undefined` on + * success), guaranteeing the rejection is consumed. + */ +export async function awaitTailWithGrace( + graceMs: number, + tail: Promise, + onDetachedSettled: (error?: unknown) => void, +): Promise<'completed' | 'detached'> { + let timer: ReturnType | undefined; + try { + const winner = await Promise.race([ + tail, + new Promise((resolve) => { + timer = setTimeout(() => resolve(TAIL_GRACE_TIMEOUT), Math.max(0, graceMs)); + // Never keep the process alive just to time out a grace window. + timer.unref?.(); + }), + ]); + if (winner === TAIL_GRACE_TIMEOUT) { + tail.then( + () => onDetachedSettled(), + (err) => onDetachedSettled(err), + ); + return 'detached'; + } + return 'completed'; + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} diff --git a/packages/core/test/publish-tail-grace.test.ts b/packages/core/test/publish-tail-grace.test.ts new file mode 100644 index 000000000..88feb2a30 --- /dev/null +++ b/packages/core/test/publish-tail-grace.test.ts @@ -0,0 +1,74 @@ +import { describe, it, expect } from 'vitest'; +import { + awaitTailWithGrace, + resolvePublishTailGraceMs, + DEFAULT_PUBLISH_TAIL_GRACE_MS, +} from '../src/publish-tail-grace.js'; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +describe('awaitTailWithGrace (GH #1572)', () => { + it('returns "completed" when the tail settles inside the grace, without firing the detach callback', async () => { + let detached = 0; + const outcome = await awaitTailWithGrace(1_000, sleep(20), () => { detached += 1; }); + expect(outcome).toBe('completed'); + // give any stray callback a chance to fire before asserting + await sleep(50); + expect(detached).toBe(0); + }); + + it('rethrows a tail failure that happens inside the grace (plain-await parity)', async () => { + let detached = 0; + const failing = (async () => { await sleep(10); throw new Error('store exploded'); })(); + await expect(awaitTailWithGrace(1_000, failing, () => { detached += 1; })).rejects.toThrow('store exploded'); + await sleep(50); + expect(detached).toBe(0); + }); + + it('returns "detached" promptly when the tail outlives the grace, and the tail still completes', async () => { + let tailDone = false; + let detachedError: unknown = 'not-called'; + const tail = (async () => { await sleep(300); tailDone = true; })(); + const started = Date.now(); + const outcome = await awaitTailWithGrace(50, tail, (err) => { detachedError = err; }); + const waited = Date.now() - started; + expect(outcome).toBe('detached'); + expect(waited).toBeLessThan(250); // responded at ~graceMs, not at tail completion + expect(tailDone).toBe(false); // tail genuinely still running at detach time + await sleep(400); + expect(tailDone).toBe(true); // ...and it ran to completion afterwards + expect(detachedError).toBeUndefined(); // settled-ok callback fired with no error + }); + + it('routes a post-grace tail failure into the callback instead of an unhandled rejection', async () => { + let detachedError: unknown; + const tail = (async () => { await sleep(150); throw new Error('late failure'); })(); + const outcome = await awaitTailWithGrace(20, tail, (err) => { detachedError = err; }); + expect(outcome).toBe('detached'); + await sleep(300); + expect(detachedError).toBeInstanceOf(Error); + expect((detachedError as Error).message).toBe('late failure'); + }); + + it('graceMs=0 detaches immediately for a pending tail', async () => { + let tailDone = false; + const tail = (async () => { await sleep(100); tailDone = true; })(); + const outcome = await awaitTailWithGrace(0, tail, () => {}); + expect(outcome).toBe('detached'); + expect(tailDone).toBe(false); + await sleep(200); + expect(tailDone).toBe(true); + }); +}); + +describe('resolvePublishTailGraceMs', () => { + it('defaults when unset, empty, or invalid; honors valid values including 0', () => { + expect(resolvePublishTailGraceMs({} as NodeJS.ProcessEnv)).toBe(DEFAULT_PUBLISH_TAIL_GRACE_MS); + expect(resolvePublishTailGraceMs({ DKG_PUBLISH_TAIL_GRACE_MS: '' } as NodeJS.ProcessEnv)).toBe(DEFAULT_PUBLISH_TAIL_GRACE_MS); + expect(resolvePublishTailGraceMs({ DKG_PUBLISH_TAIL_GRACE_MS: 'soon' } as NodeJS.ProcessEnv)).toBe(DEFAULT_PUBLISH_TAIL_GRACE_MS); + expect(resolvePublishTailGraceMs({ DKG_PUBLISH_TAIL_GRACE_MS: '-5' } as NodeJS.ProcessEnv)).toBe(DEFAULT_PUBLISH_TAIL_GRACE_MS); + expect(resolvePublishTailGraceMs({ DKG_PUBLISH_TAIL_GRACE_MS: '0' } as NodeJS.ProcessEnv)).toBe(0); + expect(resolvePublishTailGraceMs({ DKG_PUBLISH_TAIL_GRACE_MS: '12000' } as NodeJS.ProcessEnv)).toBe(12_000); + expect(resolvePublishTailGraceMs({ DKG_PUBLISH_TAIL_GRACE_MS: '2500.9' } as NodeJS.ProcessEnv)).toBe(2_500); + }); +}); diff --git a/packages/publisher/src/dkg-publisher.ts b/packages/publisher/src/dkg-publisher.ts index d37945619..7a3b35b18 100644 --- a/packages/publisher/src/dkg-publisher.ts +++ b/packages/publisher/src/dkg-publisher.ts @@ -2,7 +2,7 @@ import type { Quad, TripleStore } from '@origintrail-official/dkg-storage'; import type { ChainAdapter, OnChainPublishResult, AddBatchToContextGraphParams } from '@origintrail-official/dkg-chain'; import { enrichEvmError } from '@origintrail-official/dkg-chain'; import type { EventBus, OperationContext } from '@origintrail-official/dkg-core'; -import { DKGEvent, Logger, createOperationContext, sha256, encodeWorkspacePublishRequest, encodeEncryptedWorkspacePayload, encryptWorkspacePayload, contextGraphDataUri, contextGraphDataGraphUri, contextGraphMetaUri, contextGraphAssertionUri, contextGraphLayerUri, MemoryLayer, assertionLifecycleUri, contextGraphSubGraphUri, contextGraphSubGraphMetaUri, SYSTEM_CONTEXT_GRAPHS, validateSubGraphName, isSafeIri, assertSafeIri, assertSafeRdfTerm, assertQuadLiteralsMutf8Safe, DKG_GOSSIP_MAX_MESSAGE_BYTES, SwmGossipPayloadTooLargeError, STORAGE_ACK_MAX_STAGING_BYTES, type Ed25519Keypair, buildAuthorAttestationTypedData, buildUpdateAuthorAttestationTypedData, AUTHOR_SCHEME_VERSION_V1, TrustLevel, TRUST_LEVEL_PREDICATE, assertNoUserAuthoredTrustLevelQuads, buildTrustLevelQuads, isTrustLevelQuad, isSwmMerkleExcludedQuad, WORKSPACE_OWNER_PREDICATE, DKG_ENTITY, DKG_ROOT_ENTITY_LEGACY, ENTITY_PRED_ALT, parseAssertionSealQuads, ASSERTION_SEAL_PREDICATES, sharedMemoryReadBothFilter, DKG_ONTOLOGY } from '@origintrail-official/dkg-core'; +import { DKGEvent, Logger, createOperationContext, sha256, encodeWorkspacePublishRequest, encodeEncryptedWorkspacePayload, encryptWorkspacePayload, contextGraphDataUri, contextGraphDataGraphUri, contextGraphMetaUri, contextGraphAssertionUri, contextGraphLayerUri, MemoryLayer, assertionLifecycleUri, contextGraphSubGraphUri, contextGraphSubGraphMetaUri, SYSTEM_CONTEXT_GRAPHS, validateSubGraphName, isSafeIri, assertSafeIri, assertSafeRdfTerm, assertQuadLiteralsMutf8Safe, DKG_GOSSIP_MAX_MESSAGE_BYTES, SwmGossipPayloadTooLargeError, STORAGE_ACK_MAX_STAGING_BYTES, type Ed25519Keypair, buildAuthorAttestationTypedData, buildUpdateAuthorAttestationTypedData, AUTHOR_SCHEME_VERSION_V1, TrustLevel, TRUST_LEVEL_PREDICATE, assertNoUserAuthoredTrustLevelQuads, buildTrustLevelQuads, isTrustLevelQuad, isSwmMerkleExcludedQuad, WORKSPACE_OWNER_PREDICATE, DKG_ENTITY, DKG_ROOT_ENTITY_LEGACY, ENTITY_PRED_ALT, parseAssertionSealQuads, ASSERTION_SEAL_PREDICATES, sharedMemoryReadBothFilter, DKG_ONTOLOGY, awaitTailWithGrace, resolvePublishTailGraceMs } from '@origintrail-official/dkg-core'; import { GraphManager, PrivateContentStore, loadSelectedSharedMemoryQuads } from '@origintrail-official/dkg-storage'; import { DEFAULT_PUBLISH_EPOCHS, MAX_PUBLISH_EPOCHS, type Publisher, type PublishOptions, type PublishResult, type KAManifestEntry, type PhaseCallback, type V10CoreNodeACK, type V10ACKProviderParams, type V10ACKProviderObject, type LegacyV10ACKProvider } from './publisher.js'; import { skolemizeByEntity } from './auto-partition.js'; @@ -1891,11 +1891,30 @@ export class DKGPublisher implements Publisher { // clearSharedMemoryAfter controls only whether the REMAINING unpublished triples are also cleared. if (publishResult.status === 'confirmed') { const kaMap = skolemizeByEntity(quads); - await this.clearPublishedSwmRoots(contextGraphId, [...kaMap.keys()], options?.subGraphName, ctx); - // If clearSharedMemoryAfter is explicitly true, also clear any remaining unpublished content. - // Default is false: unpublished entities stay in SWM for future publishes. - if (options?.clearSharedMemoryAfter === true) { - await this.clearRemainingSharedMemory(contextGraphId, options?.subGraphName, ctx); + // GH #1572: the cleanup sweep scales with the accumulated SWM graph and + // runs on the store queue with no deadline — on a congested node it can + // take minutes and must not gate the caller's publish response. Give it + // a bounded grace (a healthy node finishes well inside it, keeping + // today's behavior — including error propagation — byte-identical), + // then let it finish detached with its settlement logged. + const swmCleanup = (async () => { + await this.clearPublishedSwmRoots(contextGraphId, [...kaMap.keys()], options?.subGraphName, ctx); + // If clearSharedMemoryAfter is explicitly true, also clear any remaining unpublished content. + // Default is false: unpublished entities stay in SWM for future publishes. + if (options?.clearSharedMemoryAfter === true) { + await this.clearRemainingSharedMemory(contextGraphId, options?.subGraphName, ctx); + } + })(); + const graceMs = resolvePublishTailGraceMs(); + const outcome = await awaitTailWithGrace(graceMs, swmCleanup, (err) => { + if (err !== undefined) { + this.log.warn(ctx, `Detached SWM cleanup failed after confirmed publish (GH #1572): ${err instanceof Error ? err.message : String(err)}`); + } else { + this.log.info(ctx, 'Detached SWM cleanup completed after confirmed publish (GH #1572)'); + } + }); + if (outcome === 'detached') { + this.log.info(ctx, `SWM cleanup still running after ${graceMs}ms grace — continuing detached so the publish response is not delayed (GH #1572)`); } }