fix: make run log creation atomic
This commit is contained in:
1 parent
4cb61399b1
commit
d2e94959e0
3 files changed
+186
-7
No files matched your search
@@ -1,4 +1,4 @@
|
|||||||
import { mkdirSync } from "node:fs";
|
import { mkdirSync, rmSync, writeFileSync } from "node:fs";
|
||||||
import { join } from "node:path";
|
import { join } from "node:path";
|
||||||
import { DatabaseSync, type SQLInputValue } from "node:sqlite";
|
import { DatabaseSync, type SQLInputValue } from "node:sqlite";
|
||||||
import { buildThreadMethods } from "./database/threads";
|
import { buildThreadMethods } from "./database/threads";
|
||||||
@@ -38,6 +38,15 @@ function nowIso() {
|
|||||||
return new Date().toISOString();
|
return new Date().toISOString();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function assertWritableDirectory(dir: string, label: string): void {
|
||||||
|
const probePath = join(dir, `.agentcompany-${label}-write-probe-${process.pid}-${Date.now()}`);
|
||||||
|
try {
|
||||||
|
writeFileSync(probePath, "", { encoding: "utf8", flag: "wx" });
|
||||||
|
} finally {
|
||||||
|
rmSync(probePath, { force: true });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
export interface AppDatabaseCoreModules {
|
export interface AppDatabaseCoreModules {
|
||||||
threads: ReturnType<typeof buildThreadMethods>;
|
threads: ReturnType<typeof buildThreadMethods>;
|
||||||
company: ReturnType<typeof buildCompanyMethods>;
|
company: ReturnType<typeof buildCompanyMethods>;
|
||||||
@@ -101,6 +110,7 @@ export class AppDatabaseBase {
|
|||||||
this.appVersion = options.appVersion ?? "unknown";
|
this.appVersion = options.appVersion ?? "unknown";
|
||||||
mkdirSync(rootDir, { recursive: true });
|
mkdirSync(rootDir, { recursive: true });
|
||||||
mkdirSync(this.logsDir, { recursive: true });
|
mkdirSync(this.logsDir, { recursive: true });
|
||||||
|
assertWritableDirectory(this.logsDir, "logs");
|
||||||
mkdirSync(this.artifactsDir, { recursive: true });
|
mkdirSync(this.artifactsDir, { recursive: true });
|
||||||
mkdirSync(this.backupsDir, { recursive: true });
|
mkdirSync(this.backupsDir, { recursive: true });
|
||||||
this.vault = new SecretVault(rootDir, { isPackaged: options.isPackaged });
|
this.vault = new SecretVault(rootDir, { isPackaged: options.isPackaged });
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
import { randomUUID } from "node:crypto";
|
import { randomUUID } from "node:crypto";
|
||||||
import { writeFileSync } from "node:fs";
|
import { rmSync, writeFileSync } from "node:fs";
|
||||||
import { join } from "node:path";
|
import { join } from "node:path";
|
||||||
import type { DatabaseSync, SQLInputValue } from "node:sqlite";
|
import type { DatabaseSync, SQLInputValue } from "node:sqlite";
|
||||||
import type { ConnectorId, RunRecord, RunUsageStatus } from "@shared/types";
|
import type { ConnectorId, RunRecord, RunUsageStatus } from "@shared/types";
|
||||||
@@ -24,6 +24,14 @@ export function buildRunMethods(
|
|||||||
logsDir: string;
|
logsDir: string;
|
||||||
},
|
},
|
||||||
) {
|
) {
|
||||||
|
function createEmptyRunLog(logPath: string): void {
|
||||||
|
writeFileSync(logPath, "", { encoding: "utf8", flag: "wx" });
|
||||||
|
}
|
||||||
|
|
||||||
|
function cleanupEmptyRunLog(logPath: string): void {
|
||||||
|
rmSync(logPath, { force: true });
|
||||||
|
}
|
||||||
|
|
||||||
function insertRunRow(input: {
|
function insertRunRow(input: {
|
||||||
id: string;
|
id: string;
|
||||||
companyId: string;
|
companyId: string;
|
||||||
@@ -85,10 +93,15 @@ export function buildRunMethods(
|
|||||||
const id = input.id ?? randomUUID();
|
const id = input.id ?? randomUUID();
|
||||||
const ts = nowIso();
|
const ts = nowIso();
|
||||||
const logPath = join(deps.logsDir, `${id}.log`);
|
const logPath = join(deps.logsDir, `${id}.log`);
|
||||||
transaction(() => {
|
createEmptyRunLog(logPath);
|
||||||
insertRunRow({ ...input, id, logPath, ts });
|
try {
|
||||||
});
|
transaction(() => {
|
||||||
writeFileSync(logPath, "", "utf8");
|
insertRunRow({ ...input, id, logPath, ts });
|
||||||
|
});
|
||||||
|
} catch (error) {
|
||||||
|
cleanupEmptyRunLog(logPath);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
return { id, logPath };
|
return { id, logPath };
|
||||||
},
|
},
|
||||||
|
|
||||||
@@ -103,6 +116,7 @@ export function buildRunMethods(
|
|||||||
const id = input.id ?? randomUUID();
|
const id = input.id ?? randomUUID();
|
||||||
const ts = nowIso();
|
const ts = nowIso();
|
||||||
const logPath = join(deps.logsDir, `${id}.log`);
|
const logPath = join(deps.logsDir, `${id}.log`);
|
||||||
|
createEmptyRunLog(logPath);
|
||||||
try {
|
try {
|
||||||
transaction(() => {
|
transaction(() => {
|
||||||
insertRunRow({ ...input, id, logPath, ts });
|
insertRunRow({ ...input, id, logPath, ts });
|
||||||
@@ -112,12 +126,12 @@ export function buildRunMethods(
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
|
cleanupEmptyRunLog(logPath);
|
||||||
if (error instanceof TaskCheckoutConflict) {
|
if (error instanceof TaskCheckoutConflict) {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
throw error;
|
throw error;
|
||||||
}
|
}
|
||||||
writeFileSync(logPath, "", "utf8");
|
|
||||||
return { id, logPath };
|
return { id, logPath };
|
||||||
},
|
},
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,155 @@
|
|||||||
|
import { randomUUID } from "node:crypto";
|
||||||
|
import { existsSync, mkdirSync } from "node:fs";
|
||||||
|
import { mkdtemp, rm } from "node:fs/promises";
|
||||||
|
import { tmpdir } from "node:os";
|
||||||
|
import { join } from "node:path";
|
||||||
|
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||||
|
|
||||||
|
vi.mock("electron", () => ({
|
||||||
|
safeStorage: {
|
||||||
|
isEncryptionAvailable: () => false,
|
||||||
|
getSelectedStorageBackend: () => "basic_text",
|
||||||
|
encryptString: (value: string) => Buffer.from(value, "utf8"),
|
||||||
|
decryptString: (value: Buffer) => value.toString("utf8"),
|
||||||
|
},
|
||||||
|
}));
|
||||||
|
|
||||||
|
import { AppDatabase } from "../../src/main/database";
|
||||||
|
|
||||||
|
type RawDb = {
|
||||||
|
db: {
|
||||||
|
prepare: (sql: string) => {
|
||||||
|
get: (...args: unknown[]) => unknown;
|
||||||
|
};
|
||||||
|
};
|
||||||
|
};
|
||||||
|
|
||||||
|
function createCompany(db: AppDatabase): string {
|
||||||
|
return db.core.company.saveCompany({
|
||||||
|
name: "Run Log Atomicity Co",
|
||||||
|
description: "Run log creation tests",
|
||||||
|
status: "active",
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
function createConnector(db: AppDatabase): void {
|
||||||
|
db.core.connectors.saveConnector({
|
||||||
|
id: "openai-codex",
|
||||||
|
model: "gpt-5",
|
||||||
|
envBindingText: "",
|
||||||
|
notes: "Test connector",
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
function createWorkspace(db: AppDatabase, companyId: string): string {
|
||||||
|
return db.core.workspaces.saveWorkspace({
|
||||||
|
companyId,
|
||||||
|
name: "Workspace",
|
||||||
|
localPath: "/tmp/run-log-atomicity",
|
||||||
|
repoUrl: "",
|
||||||
|
repoRef: "",
|
||||||
|
isPrimary: true,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
function createAgent(db: AppDatabase, companyId: string, workspaceId: string): string {
|
||||||
|
return db.core.agents.saveAgent({
|
||||||
|
companyId,
|
||||||
|
name: "Agent",
|
||||||
|
role: "IC",
|
||||||
|
title: "Agent",
|
||||||
|
status: "active",
|
||||||
|
reportsTo: null,
|
||||||
|
connectorId: "openai-codex",
|
||||||
|
workspaceId,
|
||||||
|
model: null,
|
||||||
|
capabilities: "Executes tasks",
|
||||||
|
budgetMonthlyUsd: 100,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
function createTask(db: AppDatabase, companyId: string, agentId: string | null, workspaceId: string): string {
|
||||||
|
return db.core.tasks.saveTask({
|
||||||
|
companyId,
|
||||||
|
title: "Task",
|
||||||
|
description: "A task",
|
||||||
|
assigneeAgentId: agentId,
|
||||||
|
workspaceId,
|
||||||
|
priority: "medium",
|
||||||
|
status: "todo",
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
describe("run log creation atomicity", () => {
|
||||||
|
let tempDir: string;
|
||||||
|
let db: AppDatabase;
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
tempDir = await mkdtemp(join(tmpdir(), "agentcompany-run-log-"));
|
||||||
|
db = new AppDatabase(tempDir);
|
||||||
|
db.init();
|
||||||
|
});
|
||||||
|
|
||||||
|
afterEach(async () => {
|
||||||
|
db.close();
|
||||||
|
await rm(tempDir, { recursive: true, force: true });
|
||||||
|
});
|
||||||
|
|
||||||
|
it("createRun rejects before inserting the run row when the log path is unavailable", () => {
|
||||||
|
const companyId = createCompany(db);
|
||||||
|
createConnector(db);
|
||||||
|
const workspaceId = createWorkspace(db, companyId);
|
||||||
|
const agentId = createAgent(db, companyId, workspaceId);
|
||||||
|
const taskId = createTask(db, companyId, null, workspaceId);
|
||||||
|
const blockedRunId = randomUUID();
|
||||||
|
mkdirSync(join(db.logsDir, `${blockedRunId}.log`));
|
||||||
|
|
||||||
|
expect(() =>
|
||||||
|
db.core.runs.createRun({
|
||||||
|
id: blockedRunId,
|
||||||
|
companyId,
|
||||||
|
taskId,
|
||||||
|
agentId,
|
||||||
|
workspaceId,
|
||||||
|
connectorId: "openai-codex",
|
||||||
|
}),
|
||||||
|
).toThrow();
|
||||||
|
|
||||||
|
const storedRun = (db as unknown as RawDb).db
|
||||||
|
.prepare("SELECT id FROM runs WHERE id = ?")
|
||||||
|
.get(blockedRunId);
|
||||||
|
expect(storedRun).toBeUndefined();
|
||||||
|
const task = db.core.entity.getTask(taskId);
|
||||||
|
expect(task.activeRunId).toBeNull();
|
||||||
|
expect(task.status).toBe("todo");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("createTaskRun removes the empty log when checkout loses the race", () => {
|
||||||
|
const companyId = createCompany(db);
|
||||||
|
createConnector(db);
|
||||||
|
const workspaceId = createWorkspace(db, companyId);
|
||||||
|
const agentId = createAgent(db, companyId, workspaceId);
|
||||||
|
const taskId = createTask(db, companyId, null, workspaceId);
|
||||||
|
const firstRun = db.core.runs.createTaskRun({
|
||||||
|
companyId,
|
||||||
|
taskId,
|
||||||
|
agentId,
|
||||||
|
workspaceId,
|
||||||
|
connectorId: "openai-codex",
|
||||||
|
});
|
||||||
|
expect(firstRun).not.toBeNull();
|
||||||
|
|
||||||
|
const losingRunId = randomUUID();
|
||||||
|
const secondRun = db.core.runs.createTaskRun({
|
||||||
|
id: losingRunId,
|
||||||
|
companyId,
|
||||||
|
taskId,
|
||||||
|
agentId,
|
||||||
|
workspaceId,
|
||||||
|
connectorId: "openai-codex",
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(secondRun).toBeNull();
|
||||||
|
expect(existsSync(join(db.logsDir, `${losingRunId}.log`))).toBe(false);
|
||||||
|
});
|
||||||
|
});
|
||||||
Reference in new issue
Block a user