Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 17 additions & 2 deletions packages/control-core/src/service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -136,9 +136,13 @@ function sameRequest(left: RunRecordV1, right: RunRecordV1): boolean {
);
}

async function readIfPresent(store: ObjectStore, key: string): Promise<unknown | null> {
async function readIfPresent(
store: ObjectStore,
key: string,
options?: { fresh?: boolean },
): Promise<unknown | null> {
try {
return await readJson(store, key);
return await readJson(store, key, options);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
throw error;
Expand Down Expand Up @@ -908,6 +912,17 @@ export class ControlService {
Date.now() - Date.parse(latest.started_at) < this.options.restartDelayMs
)
return;
// A completed native result can become readable after the projection's
// Bucket listings began. Do not start another parent from that stale view.
const nativeResult = asRecord(
await readIfPresent(
this.store,
`runs/${projected.record.run_id}/job/result.json`,
{ fresh: true },
),
);
if (typeof nativeResult?.finished_at === "string" && nativeResult.finished_at)
return;
let parent: JobObservation;
try {
parent = await this.sequenceInference(async () => {
Expand Down
10 changes: 8 additions & 2 deletions packages/control-core/src/store.ts
Original file line number Diff line number Diff line change
Expand Up @@ -155,8 +155,14 @@ export class FilesystemObjectStore implements ObjectStore {
}
}

export async function readJson(store: ObjectStore, key: string): Promise<unknown> {
return JSON.parse(new TextDecoder().decode(await store.read(key))) as unknown;
export async function readJson(
store: ObjectStore,
key: string,
options?: { fresh?: boolean },
): Promise<unknown> {
return JSON.parse(
new TextDecoder().decode(await store.read(key, options)),
) as unknown;
}

export async function createJson(
Expand Down
40 changes: 40 additions & 0 deletions packages/control-core/test/reconciliation-performance.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -323,6 +323,46 @@ it.each([4, 16, 64])(
},
);

it("does not restart a parent when a finished native result appears after stale listings", async () => {
const id = await seed(0, "run");
const resultKey = `runs/${id}/job/result.json`;
await putJson(store, resultKey, { finished_at: now, n_total_trials: 1 });
const list = vi.mocked(store.list).getMockImplementation();
if (!list) throw new Error("missing list fixture");
let calls = 0;
vi.spyOn(store, "list").mockImplementation(async (prefix) => {
const entries = await list(prefix);
// The direct object is already readable, but both reconciliation listings
// started before its upload became visible in the Bucket listing.
return ++calls <= 2 ? entries.filter((entry) => entry.key !== resultKey) : entries;
});

await service.reconcile();

expect(port.startParent).not.toHaveBeenCalled();
expect(store.read).toHaveBeenCalledWith(resultKey, { fresh: true });
expect(projection.run(id)?.status).toBe("finished");
});

it("bypasses cached unfinished results before starting a replacement parent", async () => {
const id = await seed(0, "run");
const resultKey = `runs/${id}/job/result.json`;
await putJson(store, resultKey, { finished_at: now, n_total_trials: 1 });
const read = vi.mocked(store.read).getMockImplementation();
if (!read) throw new Error("missing read fixture");
vi.spyOn(store, "read").mockImplementation((key, options) =>
key === resultKey && !options?.fresh
? Promise.resolve(new TextEncoder().encode(JSON.stringify({ finished_at: null })))
: read(key, options),
);

await service.reconcile();

expect(port.startParent).not.toHaveBeenCalled();
expect(store.read).toHaveBeenCalledWith(resultKey, { fresh: true });
expect(projection.run(id)?.status).toBe("queued");
});

it("does not skip terminal live Jobs present in the initial complete snapshot", async () => {
const terminal = await seed(0);
jobs.push(observation(terminal), observation(terminal, "child", "trial"));
Expand Down
Loading