Fix sourcedFrom blob metadata divergence - #647
Conversation
Pin connections to every worker on two replicated nodes and race independent sourcedFrom fills through an external barrier. Require the raw record, point reads, metadata, and blob payload to converge on one write. Co-Authored-By: GPT-5 Codex <noreply@openai.com>
Co-Authored-By: GPT-5 Codex <noreply@openai.com>
Co-Authored-By: GPT-5 Codex <noreply@openai.com>
Co-Authored-By: GPT-5 Codex <noreply@openai.com>
There was a problem hiding this comment.
Code Review
This pull request adds a new integration test suite and associated fixtures to verify that a sourced record's metadata and blob converge correctly during competing cache fills across multiple nodes. The feedback recommends replacing CommonJS-specific globals with ESM-safe fallbacks, ensuring parallel processes are tracked for cleanup even if one fails, wrapping test cleanup steps in try-catch blocks to prevent resource leaks, and adding safety checks for potentially null payload references.
| import { sendOperation } from './clusterShared.mjs'; | ||
|
|
||
| process.env.HARPER_INTEGRATION_TEST_INSTALL_SCRIPT = resolve( | ||
| import.meta.dirname ?? module.path, |
There was a problem hiding this comment.
In ES modules (ESM), avoid using CommonJS-specific globals like module (e.g., module.path) as a fallback when import.meta.dirname is undefined. Use an ESM-safe fallback such as new URL('.', import.meta.url).pathname instead.
| import.meta.dirname ?? module.path, | |
| import.meta.dirname ?? new URL('.', import.meta.url).pathname, |
References
- In ES modules (ESM), avoid using CommonJS-specific globals like
module(e.g.,module.path) as a fallback whenimport.meta.dirnameis undefined. Use an ESM-safe fallback such asnew URL('.', import.meta.url).pathnameinstead.
| 'harper.js' | ||
| ); | ||
|
|
||
| const FIXTURE = resolve(import.meta.dirname ?? module.path, 'fixture-sourced-blob-pairing'); |
There was a problem hiding this comment.
In ES modules (ESM), avoid using CommonJS-specific globals like module (e.g., module.path) as a fallback when import.meta.dirname is undefined. Use an ESM-safe fallback such as new URL('.', import.meta.url).pathname instead.
| const FIXTURE = resolve(import.meta.dirname ?? module.path, 'fixture-sourced-blob-pairing'); | |
| const FIXTURE = resolve(import.meta.dirname ?? new URL('.', import.meta.url).pathname, 'fixture-sourced-blob-pairing'); |
References
- In ES modules (ESM), avoid using CommonJS-specific globals like
module(e.g.,module.path) as a fallback whenimport.meta.dirnameis undefined. Use an ESM-safe fallback such asnew URL('.', import.meta.url).pathnameinstead.
| await Promise.all( | ||
| contexts.map((nodeCtx) => | ||
| startHarper(nodeCtx, { | ||
| config: { | ||
| analytics: { aggregatePeriod: -1 }, | ||
| logging: { colors: false, stdStreams: false, console: true }, | ||
| replication: { securePort: `${nodeCtx.harper.hostname}:9933` }, | ||
| threads: { count: WORKERS }, | ||
| }, | ||
| env: { HARPER_NO_FLUSH_ON_EXIT: true, HARPER_TEST_ORIGIN_URL: ctx.origin.url }, | ||
| }) | ||
| ) | ||
| ); | ||
| ctx.nodes = contexts.map((nodeCtx) => nodeCtx.harper); |
There was a problem hiding this comment.
When starting multiple asynchronous processes in parallel using Promise.all, assign the successfully started instances within their respective promise chains (e.g., using .then()) rather than waiting for Promise.all to resolve. This ensures that if one process fails, the already started processes are still recorded and can be properly cleaned up during teardown.
| await Promise.all( | |
| contexts.map((nodeCtx) => | |
| startHarper(nodeCtx, { | |
| config: { | |
| analytics: { aggregatePeriod: -1 }, | |
| logging: { colors: false, stdStreams: false, console: true }, | |
| replication: { securePort: `${nodeCtx.harper.hostname}:9933` }, | |
| threads: { count: WORKERS }, | |
| }, | |
| env: { HARPER_NO_FLUSH_ON_EXIT: true, HARPER_TEST_ORIGIN_URL: ctx.origin.url }, | |
| }) | |
| ) | |
| ); | |
| ctx.nodes = contexts.map((nodeCtx) => nodeCtx.harper); | |
| ctx.nodes = []; | |
| await Promise.all( | |
| contexts.map((nodeCtx) => | |
| startHarper(nodeCtx, { | |
| config: { | |
| analytics: { aggregatePeriod: -1 }, | |
| logging: { colors: false, stdStreams: false, console: true }, | |
| replication: { securePort: ` ${nodeCtx.harper.hostname}:9933` }, | |
| threads: { count: WORKERS }, | |
| }, | |
| env: { HARPER_NO_FLUSH_ON_EXIT: true, HARPER_TEST_ORIGIN_URL: ctx.origin.url }, | |
| }).then(() => { | |
| ctx.nodes.push(nodeCtx.harper); | |
| }) | |
| ) | |
| ); |
References
- When starting multiple asynchronous processes in parallel using Promise.all, assign the successfully started instances within their respective promise chains (e.g., using .then()) rather than waiting for Promise.all to resolve. This ensures that if one process fails, the already started processes are still recorded and can be properly cleaned up during teardown.
| after(async () => { | ||
| for (const agents of ctx.agentsByNode ?? []) for (const agent of agents.values()) agent.destroy(); | ||
| await Promise.all((ctx.nodes ?? []).map((node) => teardownHarper({ harper: node }))); | ||
| await ctx.origin?.close(); | ||
| }); |
There was a problem hiding this comment.
In test cleanup hooks (such as after or afterEach), wrap individual process termination or cleanup steps in try-catch blocks to ensure that a failure in one step does not prevent subsequent critical cleanup steps (e.g., tearing down servers or child processes) from executing, thereby avoiding resource and process leaks.
after(async () => {
for (const agents of ctx.agentsByNode ?? []) {
for (const agent of agents.values()) {
try {
agent.destroy();
} catch (error) {
// Prevent failure from blocking other cleanups
}
}
}
await Promise.all(
(ctx.nodes ?? []).map(async (node) => {
try {
await teardownHarper({ harper: node });
} catch (error) {
// Prevent failure from blocking other cleanups
}
})
);
try {
await ctx.origin?.close();
} catch (error) {
// Prevent failure from blocking other cleanups
}
});References
- In test cleanup hooks (such as after or afterEach), wrap individual process termination or cleanup steps in try-catch blocks to ensure that a failure in one step does not prevent subsequent critical cleanup steps (e.g., tearing down servers or child processes) from executing, thereby avoiding resource and process leaks.
| function describeRecord(record) { | ||
| if (!record) return null; | ||
| return record.payload.bytes().then((bytes) => ({ | ||
| id: record.id, | ||
| token: record.token, | ||
| sourceNode: record.sourceNode, | ||
| sourceThread: record.sourceThread, | ||
| payloadToken: bytes.subarray(0, Buffer.byteLength(record.token)).toString(), | ||
| })); | ||
| } |
There was a problem hiding this comment.
Ensure appropriate null/undefined checks exist before accessing properties or calling methods on potentially nullable references like record.payload to prevent runtime TypeErrors.
| function describeRecord(record) { | |
| if (!record) return null; | |
| return record.payload.bytes().then((bytes) => ({ | |
| id: record.id, | |
| token: record.token, | |
| sourceNode: record.sourceNode, | |
| sourceThread: record.sourceThread, | |
| payloadToken: bytes.subarray(0, Buffer.byteLength(record.token)).toString(), | |
| })); | |
| } | |
| function describeRecord(record) { | |
| if (!record || !record.payload) return null; | |
| return record.payload.bytes().then((bytes) => ({ | |
| id: record.id, | |
| token: record.token, | |
| sourceNode: record.sourceNode, | |
| sourceThread: record.sourceThread, | |
| payloadToken: bytes.subarray(0, Buffer.byteLength(record.token)).toString(), | |
| })); | |
| } |
Co-Authored-By: GPT-5 Codex <noreply@openai.com>
| await Promise.all( | ||
| contexts.map((nodeCtx) => | ||
| startHarper(nodeCtx, { | ||
| config: { | ||
| analytics: { aggregatePeriod: -1 }, | ||
| logging: { colors: false, stdStreams: false, console: true }, | ||
| replication: { securePort: `${nodeCtx.harper.hostname}:9933` }, | ||
| threads: { count: WORKERS }, | ||
| }, | ||
| env: { HARPER_NO_FLUSH_ON_EXIT: true, HARPER_TEST_ORIGIN_URL: ctx.origin.url }, | ||
| }) | ||
| ) | ||
| ); | ||
| ctx.nodes = contexts.map((nodeCtx) => nodeCtx.harper); |
There was a problem hiding this comment.
Blocker: node-orphan on partial startHarper failure
ctx.nodes is assigned on line 274 only after Promise.all resolves. If node A starts successfully but node B throws, Promise.all rejects and ctx.nodes is never set. after() then falls back to ctx.nodes ?? [] → [], leaving node A's process running and untracked for the rest of the CI run. This is the "setup-throw orphaning" pattern flagged and author-fixed in harper-pro#252/#304/#297.
Fix: store the context array on ctx before the await, then filter for nodes that actually started:
| await Promise.all( | |
| contexts.map((nodeCtx) => | |
| startHarper(nodeCtx, { | |
| config: { | |
| analytics: { aggregatePeriod: -1 }, | |
| logging: { colors: false, stdStreams: false, console: true }, | |
| replication: { securePort: `${nodeCtx.harper.hostname}:9933` }, | |
| threads: { count: WORKERS }, | |
| }, | |
| env: { HARPER_NO_FLUSH_ON_EXIT: true, HARPER_TEST_ORIGIN_URL: ctx.origin.url }, | |
| }) | |
| ) | |
| ); | |
| ctx.nodes = contexts.map((nodeCtx) => nodeCtx.harper); | |
| ctx._contexts = contexts; | |
| await Promise.all( | |
| contexts.map((nodeCtx) => | |
| startHarper(nodeCtx, { | |
| config: { | |
| analytics: { aggregatePeriod: -1 }, | |
| logging: { colors: false, stdStreams: false, console: true }, | |
| replication: { securePort: `${nodeCtx.harper.hostname}:9933` }, | |
| threads: { count: WORKERS }, | |
| }, | |
| env: { HARPER_NO_FLUSH_ON_EXIT: true, HARPER_TEST_ORIGIN_URL: ctx.origin.url }, | |
| }) | |
| ) | |
| ); | |
| ctx.nodes = contexts.map((nodeCtx) => nodeCtx.harper); |
And in after(), change the teardown line to:
const nodes = ctx.nodes ?? (ctx._contexts ?? []).filter(c => c.harper).map(c => c.harper);
await Promise.all(nodes.map((node) => teardownHarper({ harper: node })));|
Reviewed; no blockers found. The prior finding (node-orphan on partial |
Co-Authored-By: GPT-5 Codex <noreply@openai.com>
Problem
Concurrent independent
sourcedFromfills can settle on opposite winners across replicated nodes. The original four-node cached-blob test exposed this as metadata from one fill paired with a blob endpoint response from another, with no self-healing.The smaller harness isolates the mechanism:
The records were not torn internally. Each node could retain a different complete winner after both local fills encountered a peer fill. Routing metadata and blob reads across those divergent nodes produced the apparent split.
Change
This PR adds the surgical regression and points core at HarperFast/harper#2065 — Fix sourcedFrom cache-fill conflict convergence.
The test requires one stable
(version, token)across both nodes and every worker, then verifies both the raw record and the materialized blob bytes carry that token. It also makes origin barriers and convergence polling resilient to late calls and transient restart responses.The core PR reloads commit-time state, applies deterministic ordering to competing positive first fills, preserves strict revalidation/deletion safety, and updates indices/created-time metadata against the actual winner.
Evidence
Before the core fix, the stable all-worker assertion failed repeatedly and sometimes showed the nodes retaining swapped winners after 30 seconds. Surgical controls remained clean:
sourcedFromraces: 5/5This excludes a general blob atomicity or cross-thread visibility failure and isolates the source-fill conflict path.
Verification
HARPER_645_TRIALS=10 HARPER_645_WORKERS=2 node --test integrationTests/cluster/sourcedBlobPairing.test.mjs— 10/10 races passedfullyConnectedReplication.test.mjs— 10/10 across RocksDB and LMDB, including “Replicating cached blobs”npm run test:unit— 585 passingnpm run buildand caching suite — clean, 25 passing7c83f72bplus graded delta at04e2aeee— Claude graded review + Harper-domain adjudication9f46d08d— review gate failed because the generated artifact omitted its verdict; locally verified 10/10 in a fresh poolThe Pro build still reports unrelated baseline type errors in
analytics/profile.tsand replication WebSocket typings; this change does not touch those paths.Dependency
Keep this PR draft until core PR #2065 lands, then repoint the submodule to the merged core SHA before marking it ready.
Fixes #645
Authored by GPT-5 Codex.
🤖 Generated with Claude Code