mirror of
https://github.com/morluto/rea.git
synced 2026-11-10 00:36:35 +00:00
fix: preserve source closure during snapshot replay (#8870)
* fix: preserve source closure during snapshot replay * fix: revalidate restored snapshot sources lazily * fix: bound and share source-closure checks
This commit is contained in:
1 parent
b31fc5d5fb
commit
441ba94086
9 files changed
+661
-63
No files matched your search
@@ -46,6 +46,7 @@ import {
|
||||
snapshotReplayBinding,
|
||||
type SnapshotReplayBinding,
|
||||
} from "./AnalysisSnapshotReplayBinding.js";
|
||||
import { createInvestigationEvidenceClosureChecker } from "../investigation/InvestigationEvidence.js";
|
||||
|
||||
const RELATIONSHIP_OPERATIONS: ReadonlySet<AnalysisOperation> = new Set([
|
||||
"procedure_callers",
|
||||
@@ -162,9 +163,12 @@ export const SNAPSHOT_REPLAY_LIMITATION =
|
||||
/** Bounded in-memory cache for one immutable binary identity. */
|
||||
export class AnalysisSnapshotCache {
|
||||
readonly #entries = new Map<string, AnalysisSnapshotEntry>();
|
||||
// Null means imported provenance cannot be faithfully replayed. It is distinct
|
||||
// from an entry recorded before an Evidence ID was assigned.
|
||||
readonly #replayBindings = new Map<string, SnapshotReplayBinding | null>();
|
||||
// Null is ambiguous provenance; a pending ID is an exact parent whose source
|
||||
// closure was missing and can be checked again against the owning ledger.
|
||||
readonly #replayBindings = new Map<
|
||||
string,
|
||||
SnapshotReplayBinding | { readonly pendingEvidenceId: string } | null
|
||||
>();
|
||||
readonly #workflowEntries = new Map<string, AnalysisSnapshotWorkflowEntry>();
|
||||
#target: AnalysisSnapshotTarget | undefined;
|
||||
#binding: AnalysisSnapshotBinding | undefined;
|
||||
@@ -233,6 +237,10 @@ export class AnalysisSnapshotCache {
|
||||
record,
|
||||
]),
|
||||
);
|
||||
const lookup = (id: string) => records.get(id);
|
||||
const checkClosure = createInvestigationEvidenceClosureChecker({
|
||||
evidenceForAnalysis: lookup,
|
||||
});
|
||||
const ambiguous = new Set<string>();
|
||||
for (const evidence of snapshot.evidence_bundle.records) {
|
||||
const queryId = analysisQueryId(
|
||||
@@ -257,11 +265,15 @@ export class AnalysisSnapshotCache {
|
||||
else evidenceByQuery.set(queryId, evidence);
|
||||
}
|
||||
for (const [queryId, evidence] of evidenceByQuery) {
|
||||
if (ambiguous.has(queryId)) {
|
||||
this.#replayBindings.set(queryId, null);
|
||||
continue;
|
||||
}
|
||||
this.#replayBindings.set(
|
||||
queryId,
|
||||
ambiguous.has(queryId)
|
||||
? null
|
||||
: snapshotReplayBinding(evidence, records),
|
||||
snapshotReplayBinding(evidence, lookup, checkClosure) ?? {
|
||||
pendingEvidenceId: evidence.evidence_id,
|
||||
},
|
||||
);
|
||||
}
|
||||
for (const entry of snapshot.workflow_entries) {
|
||||
@@ -287,7 +299,12 @@ export class AnalysisSnapshotCache {
|
||||
replacements.map(({ current }) => [current.evidence_id, current]),
|
||||
);
|
||||
for (const [queryId, binding] of this.#replayBindings) {
|
||||
if (binding === null || binding.sourceEvidence === undefined) continue;
|
||||
if (
|
||||
binding === null ||
|
||||
"pendingEvidenceId" in binding ||
|
||||
binding.sourceEvidence === undefined
|
||||
)
|
||||
continue;
|
||||
const sources = binding.sourceEvidence;
|
||||
if (!sources.some((source) => currentRecords.has(source.evidence_id)))
|
||||
continue;
|
||||
@@ -490,6 +507,7 @@ export class AnalysisSnapshotCache {
|
||||
profile: AnalysisProfileCommitment,
|
||||
operation: AnalysisOperation,
|
||||
parameters: Readonly<Record<string, JsonValue>>,
|
||||
lookup?: (id: string) => Evidence | undefined,
|
||||
): AnalysisExecution | undefined {
|
||||
const queryId = analysisQueryId(
|
||||
snapshotTarget(target),
|
||||
@@ -499,10 +517,18 @@ export class AnalysisSnapshotCache {
|
||||
);
|
||||
const cached = this.#entries.get(queryId);
|
||||
if (cached === undefined) return undefined;
|
||||
const binding = this.#replayBindings.get(queryId);
|
||||
let binding = this.#replayBindings.get(queryId);
|
||||
// Do not turn a refused provenance selection into a fresh, unlinked record.
|
||||
// The provider must produce a new observation before this query can replay.
|
||||
if (binding === null) return undefined;
|
||||
if (binding !== undefined && "pendingEvidenceId" in binding) {
|
||||
const parent = lookup?.(binding.pendingEvidenceId);
|
||||
if (parent === undefined || lookup === undefined) return undefined;
|
||||
const restored = snapshotReplayBinding(parent, lookup);
|
||||
if (restored === null) return undefined;
|
||||
this.#replayBindings.set(queryId, restored);
|
||||
binding = restored;
|
||||
}
|
||||
const cachedSubject = cached.execution.subject;
|
||||
const subject =
|
||||
cachedSubject == null
|
||||
@@ -550,22 +576,31 @@ export class AnalysisSnapshotCache {
|
||||
}
|
||||
|
||||
/** Record one successful immutable call unless the cache is full. */
|
||||
record(input: {
|
||||
readonly target: BinaryTarget;
|
||||
readonly profile: AnalysisProfileCommitment;
|
||||
readonly operation: AnalysisOperation;
|
||||
readonly parameters: Readonly<Record<string, JsonValue>>;
|
||||
readonly execution: AnalysisExecution;
|
||||
readonly evidence?: Evidence;
|
||||
}): void {
|
||||
record(
|
||||
input: {
|
||||
readonly target: BinaryTarget;
|
||||
readonly profile: AnalysisProfileCommitment;
|
||||
readonly operation: AnalysisOperation;
|
||||
readonly parameters: Readonly<Record<string, JsonValue>>;
|
||||
readonly execution: AnalysisExecution;
|
||||
readonly evidence?: Evidence;
|
||||
},
|
||||
lookup?: (id: string) => Evidence | undefined,
|
||||
): void {
|
||||
const { target, profile, operation, parameters, execution } = input;
|
||||
this.select(target, profile);
|
||||
// The snapshot format does not carry reference truncation; a replay would
|
||||
// present a capped read as complete, so it is not recorded.
|
||||
if (execution.referencesTruncated === true) return;
|
||||
// Provider snapshot entries do not carry execution source records. Replay
|
||||
// must not discard the source closure that supplies the parent's Evidence links.
|
||||
if ((execution.sourceEvidence?.length ?? 0) > 0) return;
|
||||
let sourceBinding: SnapshotReplayBinding | undefined;
|
||||
if ((execution.sourceEvidence?.length ?? 0) > 0) {
|
||||
if (execution.evidenceId === undefined || lookup === undefined) return;
|
||||
const parent = lookup(execution.evidenceId);
|
||||
if (parent === undefined) return;
|
||||
const closed = snapshotReplayBinding(parent, lookup);
|
||||
if (closed === null) return;
|
||||
sourceBinding = closed;
|
||||
}
|
||||
const queryId = analysisQueryId(
|
||||
snapshotTarget(target),
|
||||
snapshotBinding(profile),
|
||||
@@ -597,12 +632,15 @@ export class AnalysisSnapshotCache {
|
||||
if (execution.evidenceId === undefined)
|
||||
this.#replayBindings.delete(entry.query_id);
|
||||
else
|
||||
this.#replayBindings.set(entry.query_id, {
|
||||
evidenceId: execution.evidenceId,
|
||||
...(input.evidence === undefined
|
||||
? {}
|
||||
: { sourceEvidence: Object.freeze([input.evidence]) }),
|
||||
});
|
||||
this.#replayBindings.set(
|
||||
entry.query_id,
|
||||
sourceBinding ?? {
|
||||
evidenceId: execution.evidenceId,
|
||||
...(input.evidence === undefined
|
||||
? {}
|
||||
: { sourceEvidence: Object.freeze([input.evidence]) }),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/** Forget all target-bound entries. */
|
||||
|
||||
Reference in new issue
Block a user