mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-07 06:16:06 -10:00
Preserve settled native log files during online migration (#19719)
## Thinking Path > - Paperclip manages work for AI agents. > - The web server must retain run history across replacements. > - Older native runs can have local log files with no final size or digest. > - The normal importer correctly refuses to call those files complete. > - The existing raw archive mode requires the entire instance to stop work. > - This change gives settled sources a separate, guarded online archive path. > - Active work can continue while old evidence moves to object storage. ## Linked Issues or Issue Description Refs #18864, #19678, #19688. The offline historical log archive path cannot migrate a busy instance. This change preserves exact surviving bytes without claiming complete native-event coverage or changing the original run outcome. ## What Changed - Add an explicit online option for settled native log files. Keep the default importer and offline option unchanged. - Require a released, closed native finalizer and no pending attempt, allocation, physical owner, or task binding. A stale lease or terminal run status alone is insufficient. - Use the canonical physical finalization lock and row locks. Check the source generation before claims and publication. - Read only the canonical, private local file. Reject aliases, symlinks, hard links, changed files, and changed owners. - Verify remote bytes before publishing a read-only archive and activity receipt in one transaction. Keep original files, errors, run status, event rows, and completeness metadata. - Document the command and its authority limits. Add real PostgreSQL and loopback S3 transport tests. ## Verification - The original implementation failed the new archive regression. It returned blocked instead of migrating the settled source. - Final source integration suite: 15 tests pass on disposable PostgreSQL and the bounded loopback S3 fixture. Coverage includes active unrelated work, failed and successful sources, a 4 MiB segment boundary, owner changes, file changes, unresolved owners, task bindings, allocations, and filesystem aliases. - The qualified source TypeScript check and module boundary check pass. - All 73 tests in five affected existing log-store, object-journal, cancellation, and migration-worker suites pass. The first borrowed dependency setup loaded work contexts from another worktree and failed four tests. Explicit database subpath aliases resolved that setup fault; no worker code changed. - Earlier test fault injection used the wrong storage method, and the first segment test reused a closed fixture endpoint. Those fixture defects were corrected before the final 15-test pass. - Local tests use Node 26. CI supplies the standard full typecheck, build, and test gates. Production object-storage credentials, live Railway replacement, and provider continuation are separate deployment acceptance steps. - Final exact-head CI: 59 successful contexts and 3 skipped, with all typecheck, build, test, E2E, security, and review gates complete. [Workflow evidence](https://github.com/paperclipai/paperclip/actions/runs/42332349644). - The first CI attempt had 12 jobs interrupted by runner shutdown. Their logs end in cancellation, with no test assertion found. The failed/cancelled-only retry passed all 12 jobs and recomputed aggregates. The 38 original successful job records kept their original execution times and were not rerun. Original interruption logs remain retained. ## Risks - This path preserves raw history. It cannot recover missing source bytes or prove a complete native transcript. - Uploads can finish before a later source check fails. The run reference stays local in that case; a retry must still satisfy all authority checks. - The new option is explicit. It adds no startup worker, recurring poll, or background scan. - Source transactions hold scoped locks only for ownership checks and publication. Streaming and remote verification run outside those transactions. ## Model Used OpenAI GPT-6 family through Codex, with reasoning, tool use, and code execution. The host does not expose the exact model ID or context window to this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes: #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip <[email protected]>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
da850990f8
commit
1a60d47804
6 files changed
+341
-12
No files matched your search
@@ -254,6 +254,25 @@ switching the run's reference. It supports terminal `interrupted` runs as well.
|
||||
It does not treat an unproven mirror as an original file or ignore a recorded
|
||||
size/digest mismatch.
|
||||
|
||||
For online preservation of settled native files, use
|
||||
`server/scripts/migrate-run-evidence.ts --preserve-settled-native-files` with
|
||||
the existing company, cursor, limit and apply options. This separate mode does
|
||||
not require global idle. It requires a native run with a finished status and
|
||||
an explicitly released finalizer in `committed` or `terminal_failure`, no pending
|
||||
attempt, no physical finalization owner, no retained allocation for that run,
|
||||
and no task execution or checkout bound to it. A stale controller deadline or
|
||||
historical PID does not establish this authority. Unknown recovery states stay
|
||||
blocked; a terminal failure's existing blocked warning remains unchanged.
|
||||
|
||||
The online path locks the native finalization scope and rechecks the source
|
||||
row, owner generation and reference before publication. The source must use its
|
||||
canonical company/agent/run filename, with no aliases, symlinks or hard links.
|
||||
It streams and verifies the exact local bytes, then publishes the read-only
|
||||
archive and an activity receipt in one guarded transaction. Originals remain
|
||||
on disk. A changed source or owner leaves the original reference unchanged.
|
||||
This mode keeps raw history separate from canonical native-event coverage,
|
||||
including when the original source has an integrity warning.
|
||||
|
||||
An archive preserves historical evidence; it does not certify a complete native
|
||||
transcript. The run's status, canonical events, and original completion metadata
|
||||
remain unchanged. Archive readers do not append missing native events, and
|
||||
|
||||
@@ -1,14 +1,18 @@
|
||||
import { createDb } from "@paperclipai/db";
|
||||
import { migrateRunEvidence } from "../src/services/run-evidence-migration.js";
|
||||
import { isCloudManagedInstance } from "../src/services/cloud-instance.js";
|
||||
import { cloudRuntimeConfigService } from "../src/services/cloud-runtime-config.js";
|
||||
import { setCloudRuntimeConfigSnapshot } from "../src/services/cloud-runtime-config-snapshot.js";
|
||||
|
||||
if (!process.env.DATABASE_URL)
|
||||
throw new Error("Set DATABASE_URL to the intended instance; run with legacy writers drained");
|
||||
const args = process.argv.slice(2);
|
||||
const flags = new Set(["--apply", "--company", "--after-run", "--after-trace", "--limit"]);
|
||||
const booleans = new Set(["--apply", "--preserve-settled-native-files"]);
|
||||
const flags = new Set([...booleans, "--company", "--after-run", "--after-trace", "--limit"]);
|
||||
for (let index = 0; index < args.length; index++) {
|
||||
const flag = args[index]!;
|
||||
if (!flags.has(flag)) throw new Error(`Unknown migration option: ${flag}`);
|
||||
if (flag !== "--apply" && (!args[++index] || args[index]!.startsWith("--"))) {
|
||||
if (!booleans.has(flag) && (!args[++index] || args[index]!.startsWith("--"))) {
|
||||
throw new Error(`Missing value for ${flag}`);
|
||||
}
|
||||
}
|
||||
@@ -18,12 +22,18 @@ const value = (flag: string) => {
|
||||
};
|
||||
const db = createDb(process.env.DATABASE_URL);
|
||||
try {
|
||||
if (isCloudManagedInstance()) {
|
||||
const stored = await cloudRuntimeConfigService(db).inspectStored();
|
||||
if (!stored) throw new Error("cloud_runtime_configuration_missing");
|
||||
setCloudRuntimeConfigSnapshot(stored);
|
||||
}
|
||||
const result = await migrateRunEvidence(db, {
|
||||
apply: args.includes("--apply"),
|
||||
companyId: value("--company"),
|
||||
afterRunId: value("--after-run"),
|
||||
afterTraceId: value("--after-trace"),
|
||||
limit: Number(value("--limit") || 50),
|
||||
preserveSettledNativeFiles: args.includes("--preserve-settled-native-files"),
|
||||
});
|
||||
console.log(JSON.stringify(result, null, 2));
|
||||
if (result.items.some((item) => item.state === "blocked")) process.exitCode = 1;
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { appendFile, link, mkdir, readFile, rm, symlink, writeFile } from "node:fs/promises";
|
||||
import { dirname, join } from "node:path";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "../__tests__/helpers/native-restart.js";
|
||||
import { createMaintenanceS3Boundary } from "../__tests__/helpers/maintenance/s3-boundary.js";
|
||||
import { withMaintenanceProvider } from "../__tests__/helpers/maintenance/server-harness.js";
|
||||
import { migrateRunEvidence } from "./run-evidence-migration.js";
|
||||
import { getRunLogStore } from "./run-log-store.js";
|
||||
import { createObjectByteJournal } from "./object-byte-journal.js";
|
||||
import { readRecoverableRunLog } from "./native-runtime/native-run-log-recovery.js";
|
||||
import { tryLockNativeWorkspaceFinalization } from "./native-runtime/native-workspace-finalization-ownership.js";
|
||||
import * as storage from "./run-evidence-storage.js";
|
||||
|
||||
function beforeUuid(id: string) {
|
||||
const hex = (BigInt(`0x${id.replaceAll("-", "")}`) - 1n).toString(16).padStart(32, "0");
|
||||
return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`;
|
||||
}
|
||||
|
||||
describeEmbeddedPostgres("online native historical log preservation", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
afterEach(() => { vi.unstubAllEnvs(); vi.restoreAllMocks(); });
|
||||
async function fixture() {
|
||||
const f = await environment.seedRun(randomUUID());
|
||||
const { companyId, agentId, paperclipHome } = environment.current;
|
||||
const ref = `${companyId}/${agentId}/${f.runId}.ndjson`, base = join(paperclipHome, `archive-test-${f.runId}`);
|
||||
const file = join(base, ref), bytes = Buffer.from('{"seq":1,"chunk":"surviving 再開 output"}\n{"partial":');
|
||||
await mkdir(dirname(file), { recursive: true }); await writeFile(file, bytes);
|
||||
await f.db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), logStore: "local_file", logRef: ref,
|
||||
logBytes: null, logSha256: null, controllerLeaseExpiresAt: new Date(0), processPid: 123456,
|
||||
processStartedAt: new Date(0), errorCode: "native_run_log_source_integrity_error", error: "original warning" })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null,
|
||||
nextAttemptAt: null, recoveryState: "blocked" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId));
|
||||
const options = { apply: true, companyId, afterRunId: beforeUuid(f.runId), limit: 1,
|
||||
kinds: ["log"] as const, preserveSettledNativeFiles: true };
|
||||
const s3 = createMaintenanceS3Boundary();
|
||||
return { ...f, companyId, agentId, ref, base, file, bytes, options, s3 };
|
||||
}
|
||||
async function withSource(action: (f: Awaited<ReturnType<typeof fixture>>) => Promise<void>) {
|
||||
const f = await fixture();
|
||||
await withMaintenanceProvider(f.s3.server, async endpoint => {
|
||||
for (const [key, value] of Object.entries({ PAPERCLIP_STORAGE_PROVIDER: "s3", PAPERCLIP_STORAGE_S3_ENDPOINT: endpoint,
|
||||
PAPERCLIP_STORAGE_S3_BUCKET: "maintenance-test", PAPERCLIP_STORAGE_S3_REGION: "us-east-1",
|
||||
PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE: "true", PAPERCLIP_STORAGE_S3_PREFIX: "",
|
||||
AWS_ACCESS_KEY_ID: "fixture-only", AWS_SECRET_ACCESS_KEY: "fixture-only",
|
||||
RUN_LOG_S3_ENDPOINT: endpoint, RUN_LOG_S3_BUCKET: "maintenance-test", RUN_LOG_S3_REGION: "us-east-1",
|
||||
RUN_LOG_S3_FORCE_PATH_STYLE: "true", RUN_LOG_S3_PREFIX: "run-logs", RUN_LOG_BASE_PATH: f.base,
|
||||
PAPERCLIP_EPHEMERAL_MIGRATION: "false" })) vi.stubEnv(key, value);
|
||||
await action(f);
|
||||
expect(f.s3.snapshot().errors).toEqual([]);
|
||||
});
|
||||
}
|
||||
|
||||
it("archives a settled failed source while unrelated work runs and preserves its original failure", async () => {
|
||||
await withSource(async f => {
|
||||
const unrelated = await environment.seedRun(randomUUID());
|
||||
const [before] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [ownerBefore] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const result = await migrateRunEvidence(f.db, f.options);
|
||||
expect(result.items).toMatchObject([{ id: f.runId, state: "migrated", historicalArchive: true,
|
||||
bytes: f.bytes.length, sha256: createHash("sha256").update(f.bytes).digest("hex") }]);
|
||||
const [after] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(after).toEqual({ ...before, logStore: "s3_archive", logRef: `${f.companyId}/${f.agentId}/${f.runId}.archive` });
|
||||
expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(ownerBefore);
|
||||
expect((await unrelated.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.runId)))[0].status).toBe("running");
|
||||
expect(await readFile(f.file)).toEqual(f.bytes);
|
||||
await rm(f.file);
|
||||
const store = getRunLogStore(), handle = { store: "s3_archive" as const, logRef: after.logRef! };
|
||||
expect((await readRecoverableRunLog(f.db, store, handle)).content).toBe(f.bytes.toString());
|
||||
expect(() => store.begin({ companyId: f.companyId, agentId: f.agentId, runId: f.runId, previousStore: "s3_archive" })).toThrow("read-only");
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves a committed successful run without changing its result or completeness metadata", async () => {
|
||||
await withSource(async f => {
|
||||
const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [result] = await f.db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: f.runId,
|
||||
completionContractId: run.completionContractId!, serverFingerprint: f.runId, schemaStatus: "accepted",
|
||||
resultJson: {}, canonicalSha256: f.runId }).returning();
|
||||
await f.db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, recoveryState: null })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "migrated", historicalArchive: true }]);
|
||||
expect((await f.db.select().from(nativeRunResults).where(eq(nativeRunResults.id, result.id)))[0]).toEqual(result);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0])
|
||||
.toMatchObject({ status: "succeeded", logBytes: null, logSha256: null, error: "original warning" });
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves exact bytes across the bounded archive segment boundary", async () => {
|
||||
await withSource(async f => {
|
||||
const bytes = Buffer.alloc(4 * 1024 * 1024 + 7, 97);
|
||||
await writeFile(f.file, bytes);
|
||||
expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "migrated", bytes: bytes.length,
|
||||
sha256: createHash("sha256").update(bytes).digest("hex") }]);
|
||||
await rm(f.file);
|
||||
const archive = createObjectByteJournal({ provider: storage.resolveRunEvidenceStorage()!.provider,
|
||||
prefix: `run-logs/segments-v1/${f.companyId}/${f.agentId}/${f.runId}.archive` });
|
||||
const head = (await archive.head())!.head;
|
||||
expect(head).toMatchObject({ closed: true, sourceCursor: 0 });
|
||||
expect(Buffer.concat([await archive.readRange(head, 0, 4 * 1024 * 1024),
|
||||
await archive.readRange(head, 4 * 1024 * 1024, 7)])).toEqual(bytes);
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["expired_owner", "pending_attempt", "pending_recovery", "allocation", "process_group", "physical_owner", "task_bound"])("refuses unresolved source authority: %s", async kind => {
|
||||
await withSource(async f => {
|
||||
if (kind === "expired_owner") await f.db.update(nativeRunFinalizations).set({ leaseOwner: "not-a-stop-proof", leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (kind === "pending_attempt") await f.db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (kind === "pending_recovery") await f.db.update(nativeRunFinalizations).set({ recoveryState: "awaiting_runner_reattach" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (kind === "allocation") await f.db.insert(environmentLeases).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
heartbeatRunId: f.runId, providerLeaseId: f.runId, status: "active" });
|
||||
if (kind === "process_group") await f.db.update(heartbeatRuns).set({ processGroupId: 999999 }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (kind === "physical_owner") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: { token: "unknown-owner" } } })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
if (kind === "task_bound") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
|
||||
expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "blocked", reason: "native_log_archive_source_unsettled" }]);
|
||||
expect(f.s3.snapshot().requests).toEqual([]);
|
||||
expect(await readFile(f.file)).toEqual(f.bytes);
|
||||
});
|
||||
});
|
||||
|
||||
it("honors the physical native finalizer lock and permits a read-only dry run", async () => {
|
||||
await withSource(async f => {
|
||||
let release!: () => void, ready!: () => void;
|
||||
const acquired = new Promise<void>(resolve => { ready = resolve; }), held = new Promise<void>(resolve => { release = resolve; });
|
||||
const owner = f.db.transaction(async tx => {
|
||||
expect(await tryLockNativeWorkspaceFinalization(tx as unknown as Db, { companyId: f.companyId, runId: f.runId })).toBe(true);
|
||||
ready(); await held;
|
||||
});
|
||||
try {
|
||||
await acquired;
|
||||
expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "blocked", reason: "native_log_archive_source_busy" }]);
|
||||
} finally { release(); await owner; }
|
||||
const result = await f.db.transaction(tx => migrateRunEvidence(tx as unknown as Db, { ...f.options, apply: false }), { accessMode: "read only" });
|
||||
expect(result.items).toMatchObject([{ state: "inventoried", historicalArchive: true }]);
|
||||
expect(f.s3.snapshot().requests).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["owner", "file"])("rechecks %s after the remote archive closes", async kind => {
|
||||
await withSource(async f => {
|
||||
const config = storage.resolveRunEvidenceStorage()!;
|
||||
vi.spyOn(storage, "resolveRunEvidenceStorage").mockReturnValue(config);
|
||||
const original = config.provider.putObjectConditional!.bind(config.provider); let changed = false;
|
||||
vi.spyOn(config.provider, "putObjectConditional").mockImplementation(async input => {
|
||||
const result = await original(input);
|
||||
if (!changed && input.objectKey.endsWith("/head.json") && Buffer.isBuffer(input.body) && JSON.parse(input.body.toString()).closed) {
|
||||
changed = true;
|
||||
if (kind === "file") await appendFile(f.file, "late source bytes");
|
||||
else await f.db.update(nativeRunFinalizations).set({ controllerGeneration: sql`${nativeRunFinalizations.controllerGeneration} + 1` })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
}
|
||||
return result;
|
||||
});
|
||||
expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "blocked" }]);
|
||||
expect(changed).toBe(true);
|
||||
expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].logStore).toBe("local_file");
|
||||
expect(f.s3.snapshot().objects.length).toBeGreaterThan(0);
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["symlink", "hardlink"])("refuses aliased source files: %s", async kind => {
|
||||
await withSource(async f => {
|
||||
const other = join(f.base, "other");
|
||||
if (kind === "hardlink") await link(f.file, other);
|
||||
else { await writeFile(other, f.bytes); await rm(f.file); await symlink(other, f.file); }
|
||||
expect((await migrateRunEvidence(f.db, f.options)).items).toMatchObject([{ state: "blocked" }]);
|
||||
expect(f.s3.snapshot().requests).toEqual([]);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,69 @@
|
||||
import { and, eq, ne, sql } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js";
|
||||
|
||||
export type NativeLogArchiveSource = {
|
||||
id: string; companyId: string; agentId: string; status: string;
|
||||
logRef: string | null; bytes: number | null; sha256: string | null;
|
||||
};
|
||||
|
||||
async function readSource(db: Db, source: NativeLogArchiveSource, lock: boolean, expected?: string) {
|
||||
function fail(): never { throw new Error("native_log_archive_source_unsettled"); }
|
||||
if (source.logRef !== `${source.companyId}/${source.agentId}/${source.id}.ndjson`) fail();
|
||||
if (lock && !await tryLockNativeWorkspaceFinalization(db, { companyId: source.companyId, runId: source.id }))
|
||||
throw new Error("native_log_archive_source_busy");
|
||||
const scope = and(eq(heartbeatRuns.id, source.id), eq(heartbeatRuns.companyId, source.companyId));
|
||||
const owners = db.select({ issueId: nativeRunFinalizations.issueId, phase: nativeRunFinalizations.phase,
|
||||
resultId: nativeRunFinalizations.resultId, recoveryState: nativeRunFinalizations.recoveryState,
|
||||
leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt,
|
||||
nextAttemptAt: nativeRunFinalizations.nextAttemptAt, controllerGeneration: nativeRunFinalizations.controllerGeneration,
|
||||
attempt: nativeRunFinalizations.attempt, controllerBootId: nativeRunFinalizations.controllerBootId,
|
||||
controllerPid: nativeRunFinalizations.controllerPid, controllerProcessStartedAt: nativeRunFinalizations.controllerProcessStartedAt,
|
||||
}).from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.runId, source.id), eq(nativeRunFinalizations.companyId, source.companyId)));
|
||||
const [owner] = await (lock ? owners.for("update") : owners);
|
||||
const rows = db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt, runtimeMode: heartbeatRuns.runtimeMode,
|
||||
issueId: heartbeatRuns.nativeIssueId, sessionId: heartbeatRuns.nativeSessionId,
|
||||
logStore: heartbeatRuns.logStore, logRef: heartbeatRuns.logRef, bytes: heartbeatRuns.logBytes, sha256: heartbeatRuns.logSha256,
|
||||
processGroupId: heartbeatRuns.processGroupId,
|
||||
physicalOwner: sql<boolean>`coalesce(${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceFinalizationOwner' <> 'null'::jsonb, false)`,
|
||||
}).from(heartbeatRuns).where(scope);
|
||||
const [run] = await (lock ? rows.for("update") : rows);
|
||||
if (!run || !owner || !run.issueId || !run.sessionId || run.runtimeMode !== "native" || !run.finishedAt
|
||||
|| run.agentId !== source.agentId || run.status !== source.status
|
||||
|| !["succeeded", "failed", "cancelled", "timed_out", "interrupted"].includes(run.status)
|
||||
|| run.logStore !== "local_file" || run.logRef !== source.logRef || run.bytes !== source.bytes || run.sha256 !== source.sha256
|
||||
|| run.processGroupId !== null || run.physicalOwner || owner.issueId !== run.issueId
|
||||
|| owner.leaseOwner !== null || owner.leaseExpiresAt !== null || owner.nextAttemptAt !== null
|
||||
|| !(owner.phase === "committed" && owner.resultId !== null && owner.recoveryState === null
|
||||
|| owner.phase === "terminal_failure" && run.status !== "succeeded"
|
||||
&& (owner.recoveryState === null || owner.recoveryState === "blocked"))) fail();
|
||||
const [task] = await db.select({ executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId })
|
||||
.from(issues).where(and(eq(issues.companyId, run.companyId), eq(issues.id, run.issueId)));
|
||||
const [allocation] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
sql`${environmentLeases.releasedAt} is null`)).limit(1);
|
||||
const [alias] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
ne(heartbeatRuns.id, run.id), eq(heartbeatRuns.logStore, "local_file"), eq(heartbeatRuns.logRef, run.logRef!))).limit(1);
|
||||
if (!task || task.executionRunId === run.id || task.checkoutRunId === run.id || allocation || alias) fail();
|
||||
const fence = nativeSha256({ ...run, finishedAt: run.finishedAt.toISOString(), owner: { phase: owner.phase, generation: owner.controllerGeneration,
|
||||
attempt: owner.attempt, bootId: owner.controllerBootId, pid: owner.controllerPid,
|
||||
startedAt: owner.controllerProcessStartedAt?.toISOString() ?? null, resultId: owner.resultId, recoveryState: owner.recoveryState } });
|
||||
if (expected && fence !== expected) throw new Error("native_log_archive_source_changed");
|
||||
return fence;
|
||||
}
|
||||
|
||||
export async function inspectNativeLogArchiveSource(db: Db, source: NativeLogArchiveSource) {
|
||||
return readSource(db, source, false);
|
||||
}
|
||||
|
||||
export async function withNativeLogArchiveSource<T>(db: Db, source: NativeLogArchiveSource, expected: string | undefined,
|
||||
action: (tx: Db, fence: string) => Promise<T>): Promise<T> {
|
||||
return db.transaction(async tx => {
|
||||
await tx.execute(sql`set local lock_timeout='1s'`);
|
||||
const fence = await readSource(tx as unknown as Db, source, true, expected);
|
||||
return action(tx as unknown as Db, fence);
|
||||
});
|
||||
}
|
||||
@@ -23,6 +23,9 @@ import {
|
||||
createNativeRunLogCoverageVerifier,
|
||||
nativeRunLogSourceWhere,
|
||||
} from "./native-runtime/native-run-log-export.js";
|
||||
import { inspectNativeLogArchiveSource, withNativeLogArchiveSource } from "./native-runtime/native-log-archive-source.js";
|
||||
import { logActivity, publishActivity, type ActivityPublication } from "./activity-log.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
|
||||
const SETTLED = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"];
|
||||
export interface RunEvidenceMigrationItem {
|
||||
@@ -48,6 +51,7 @@ export async function migrateRunEvidence(
|
||||
/** Offline enrollment only: preserve unfinalized local bytes without
|
||||
* claiming canonical native-event coverage or repairing the old run. */
|
||||
preserveHistoricalFiles?: boolean;
|
||||
preserveSettledNativeFiles?: boolean;
|
||||
},
|
||||
): Promise<{
|
||||
items: RunEvidenceMigrationItem[];
|
||||
@@ -61,6 +65,9 @@ export async function migrateRunEvidence(
|
||||
const config = resolveRunEvidenceStorage();
|
||||
if (!config) throw new Error("Run evidence migration requires instance S3 storage");
|
||||
const preserveHistoricalFiles = options.preserveHistoricalFiles === true;
|
||||
const preserveSettledNativeFiles = options.preserveSettledNativeFiles === true;
|
||||
if (preserveHistoricalFiles && preserveSettledNativeFiles)
|
||||
throw new Error("historical_log_archive_authority_conflict");
|
||||
async function assertNoActiveRuns() {
|
||||
options.signal?.throwIfAborted();
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
@@ -133,9 +140,14 @@ export async function migrateRunEvidence(
|
||||
try {
|
||||
if (!SETTLED.includes(run.status)) throw new Error("legacy_writer_not_settled");
|
||||
if (!run.logRef) throw new Error("legacy_reference_missing");
|
||||
const originalLogRef = run.logRef;
|
||||
await assertDrained(run.id, run.companyId);
|
||||
const historicalArchive = preserveHistoricalFiles && run.runtimeMode === "native" &&
|
||||
const historicalArchive = (preserveHistoricalFiles || preserveSettledNativeFiles) && run.runtimeMode === "native" &&
|
||||
(run.bytes === null || run.sha256 === null);
|
||||
const scopedArchive = historicalArchive && preserveSettledNativeFiles;
|
||||
const sourceFence = scopedArchive ? options.apply
|
||||
? await withNativeLogArchiveSource(db, run, undefined, async (_tx, fence) => fence)
|
||||
: await inspectNativeLogArchiveSource(db, run) : undefined;
|
||||
// Keep raw history separate from the canonical log namespace. A later
|
||||
// writer must never acquire this archive as a native source prefix.
|
||||
const ref = `${run.companyId}/${run.agentId}/${run.id}${historicalArchive ? ".archive" : ""}`;
|
||||
@@ -143,11 +155,16 @@ export async function migrateRunEvidence(
|
||||
provider: config.provider,
|
||||
prefix: `${config.keyPrefix || "run-logs"}/segments-v1/${ref}`,
|
||||
});
|
||||
let writer = options.apply
|
||||
const archiveSource = scopedArchive
|
||||
? await openLegacyRunLogForMigration(run.logRef, { requireCanonicalLocalFile: true }) : undefined;
|
||||
let writer: Awaited<ReturnType<typeof journal.claim>> | null = null;
|
||||
try { writer = options.apply
|
||||
? await journal.claim({
|
||||
expectedOwner: (await journal.head(options.signal))?.head.owner ?? null,
|
||||
signal: options.signal,
|
||||
assertAuthority: async () => {
|
||||
if (scopedArchive)
|
||||
await withNativeLogArchiveSource(db, run, sourceFence, async () => undefined);
|
||||
const [current] = await db
|
||||
.select({ id: heartbeatRuns.id })
|
||||
.from(heartbeatRuns)
|
||||
@@ -164,7 +181,8 @@ export async function migrateRunEvidence(
|
||||
},
|
||||
})
|
||||
: null;
|
||||
const source = await openLegacyRunLogForMigration(run.logRef);
|
||||
} catch (error) { archiveSource?.stream.destroy(); throw error; }
|
||||
const source = archiveSource ?? await openLegacyRunLogForMigration(run.logRef);
|
||||
if (source.source === "mirror" && (run.bytes === null || !run.sha256)) {
|
||||
source.stream.destroy();
|
||||
throw new Error("mirror_completeness_unproven");
|
||||
@@ -233,9 +251,10 @@ export async function migrateRunEvidence(
|
||||
if (writer.head.end !== offset) throw new Error("migration_prefix_exceeds_source");
|
||||
|
||||
const remoteHash = createHash("sha256");
|
||||
for (let start = 0; start < offset; start += 1024 * 1024)
|
||||
const verificationBytes = scopedArchive ? 4 * 1024 * 1024 : 1024 * 1024;
|
||||
for (let start = 0; start < offset; start += verificationBytes)
|
||||
remoteHash.update(
|
||||
await journal.readRange(writer.head, start, 1024 * 1024, options.signal),
|
||||
await journal.readRange(writer.head, start, verificationBytes, options.signal),
|
||||
);
|
||||
if (remoteHash.digest("hex") !== digest)
|
||||
throw new Error("migration_remote_digest_mismatch");
|
||||
@@ -243,7 +262,7 @@ export async function migrateRunEvidence(
|
||||
await journal.close(writer, options.signal);
|
||||
options.signal?.throwIfAborted();
|
||||
if (source.source === "local") await source.assertUnchanged();
|
||||
const [updated] = await db
|
||||
const publish = (database: Db) => database
|
||||
.update(heartbeatRuns)
|
||||
.set({ logStore: historicalArchive ? "s3_archive" : "s3_segments", logRef: ref })
|
||||
.where(
|
||||
@@ -251,7 +270,7 @@ export async function migrateRunEvidence(
|
||||
eq(heartbeatRuns.id, run.id),
|
||||
eq(heartbeatRuns.companyId, run.companyId),
|
||||
eq(heartbeatRuns.logStore, "local_file"),
|
||||
eq(heartbeatRuns.logRef, run.logRef),
|
||||
eq(heartbeatRuns.logRef, originalLogRef),
|
||||
run.bytes === null
|
||||
? isNull(heartbeatRuns.logBytes)
|
||||
: eq(heartbeatRuns.logBytes, run.bytes),
|
||||
@@ -268,8 +287,25 @@ export async function migrateRunEvidence(
|
||||
),
|
||||
)
|
||||
.returning({ id: heartbeatRuns.id });
|
||||
const publications: ActivityPublication[] = [];
|
||||
const [updated] = scopedArchive
|
||||
? await withNativeLogArchiveSource(db, run, sourceFence, async tx => {
|
||||
if (source.source !== "local") throw new Error("native_log_archive_local_source_required");
|
||||
await source.assertUnchanged();
|
||||
const result = await publish(tx);
|
||||
if (!result.length) throw new Error("migration_publication_conflict");
|
||||
await logActivity(tx, { companyId: run.companyId, actorType: "system", actorId: "run-evidence-migration",
|
||||
action: "run.historical_log_archived", entityType: "heartbeat_run", entityId: run.id,
|
||||
agentId: run.agentId, runId: run.id, details: { bytes: offset, sha256: digest, sourceFence } }, publications);
|
||||
return result;
|
||||
})
|
||||
: await publish(db);
|
||||
if (!updated) throw new Error("migration_publication_conflict");
|
||||
item.state = "migrated";
|
||||
for (const publication of publications) {
|
||||
try { publishActivity(publication); }
|
||||
catch { logger.warn({ runId: run.id }, "Historical log archive notification failed after publication"); }
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
options.signal?.throwIfAborted();
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { createReadStream, createWriteStream, promises as fs } from "node:fs";
|
||||
import { constants, createReadStream, createWriteStream, promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
@@ -617,9 +617,28 @@ export async function flushInFlightRunLogMirrors(): Promise<void> {
|
||||
}
|
||||
|
||||
|
||||
export async function openLegacyRunLogForMigration(logRef: string) {
|
||||
export async function openLegacyRunLogForMigration(logRef: string, options: { requireCanonicalLocalFile?: boolean } = {}) {
|
||||
const basePath = process.env.RUN_LOG_BASE_PATH ?? path.resolve(resolvePaperclipInstanceRoot(), "data", "run-logs");
|
||||
const file = resolveWithin(basePath, logRef);
|
||||
const base = options.requireCanonicalLocalFile ? await fs.realpath(basePath) : basePath;
|
||||
const file = resolveWithin(base, logRef);
|
||||
if (options.requireCanonicalLocalFile) {
|
||||
if (await fs.realpath(file) !== file) throw new Error("native_log_archive_source_alias");
|
||||
const descriptor = await fs.open(file, constants.O_RDONLY | constants.O_NOFOLLOW);
|
||||
try {
|
||||
const stat = await descriptor.stat();
|
||||
if (!stat.isFile() || stat.nlink !== 1) throw new Error("native_log_archive_source_not_private");
|
||||
return {
|
||||
stream: descriptor.createReadStream({ highWaterMark: 4 * 1024 * 1024 }), bytes: stat.size, source: "local" as const,
|
||||
async assertUnchanged() {
|
||||
const current = await fs.lstat(file);
|
||||
if (await fs.realpath(file) !== file || !current.isFile() || current.nlink !== 1
|
||||
|| current.dev !== stat.dev || current.ino !== stat.ino || current.size !== stat.size
|
||||
|| current.mtimeMs !== stat.mtimeMs || current.ctimeMs !== stat.ctimeMs)
|
||||
throw new Error("legacy_log_source_changed");
|
||||
},
|
||||
};
|
||||
} catch (error) { await descriptor.close(); throw error; }
|
||||
}
|
||||
try {
|
||||
const stat = await fs.stat(file);
|
||||
if (!stat.isFile()) throw new Error("Legacy run log is not a regular file");
|
||||
|
||||
Reference in new issue
Block a user