Write persisted files atomically
This commit is contained in:
+2
-2
@@ -30,7 +30,7 @@ $PASEO_HOME/
|
|||||||
└── push-tokens.json # Expo push notification tokens
|
└── push-tokens.json # Expo push notification tokens
|
||||||
```
|
```
|
||||||
|
|
||||||
The `agents/{sanitized-cwd}/` directory name is derived from the agent's `cwd` by stripping the filesystem root and replacing path separators with `-` (Windows drive letters become a `C-` style prefix). Atomic writes (temp file + rename): agent records, chat, project/workspace registries, push tokens. Non-atomic (plain `writeFile`): `config.json`, `schedules/*.json`, `loops/loops.json`, `server-id`, `daemon-keypair.json`.
|
The `agents/{sanitized-cwd}/` directory name is derived from the agent's `cwd` by stripping the filesystem root and replacing path separators with `-` (Windows drive letters become a `C-` style prefix). Persistent server stores write atomically by writing a temp file in the target directory and then renaming it into place.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
@@ -187,7 +187,7 @@ Local speech model ids are intentionally narrow: STT uses `parakeet-tdt-0.6b-v2-
|
|||||||
|
|
||||||
**Path:** `$PASEO_HOME/schedules/{id}.json`
|
**Path:** `$PASEO_HOME/schedules/{id}.json`
|
||||||
|
|
||||||
One file per schedule. ID is 8 hex characters. Writes are direct (not atomic).
|
One file per schedule. ID is 8 hex characters.
|
||||||
|
|
||||||
| Field | Type | Description |
|
| Field | Type | Description |
|
||||||
| ----------- | ------------------------------------- | -------------------------------- |
|
| ----------- | ------------------------------------- | -------------------------------- |
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
import { randomUUID } from "node:crypto";
|
|
||||||
import { promises as fs } from "node:fs";
|
import { promises as fs } from "node:fs";
|
||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
import { z } from "zod";
|
import { z } from "zod";
|
||||||
import type { Logger } from "pino";
|
import type { Logger } from "pino";
|
||||||
|
|
||||||
|
import { writeJsonFileAtomic } from "../atomic-file.js";
|
||||||
import { AgentFeatureSchema, AgentStatusSchema } from "../messages.js";
|
import { AgentFeatureSchema, AgentStatusSchema } from "../messages.js";
|
||||||
import { toStoredAgentRecord } from "./agent-projections.js";
|
import { toStoredAgentRecord } from "./agent-projections.js";
|
||||||
import type { ManagedAgent } from "./agent-manager.js";
|
import type { ManagedAgent } from "./agent-manager.js";
|
||||||
@@ -143,8 +143,7 @@ export class AgentStorage {
|
|||||||
const nextPath = this.buildRecordPath(record);
|
const nextPath = this.buildRecordPath(record);
|
||||||
const previousPath = this.pathById.get(agentId);
|
const previousPath = this.pathById.get(agentId);
|
||||||
|
|
||||||
await fs.mkdir(path.dirname(nextPath), { recursive: true });
|
await writeJsonFileAtomic(nextPath, record);
|
||||||
await writeFileAtomically(nextPath, JSON.stringify(record, null, 2));
|
|
||||||
this.addIndexedPath(agentId, nextPath);
|
this.addIndexedPath(agentId, nextPath);
|
||||||
|
|
||||||
if (previousPath && previousPath !== nextPath) {
|
if (previousPath && previousPath !== nextPath) {
|
||||||
@@ -387,10 +386,3 @@ function projectDirNameFromCwd(cwd: string): string {
|
|||||||
}
|
}
|
||||||
return prefix + withoutRoot.replace(/[\\/]+/g, "-");
|
return prefix + withoutRoot.replace(/[\\/]+/g, "-");
|
||||||
}
|
}
|
||||||
|
|
||||||
async function writeFileAtomically(targetPath: string, payload: string) {
|
|
||||||
const directory = path.dirname(targetPath);
|
|
||||||
const tempPath = path.join(directory, `.agent.tmp-${process.pid}-${Date.now()}-${randomUUID()}`);
|
|
||||||
await fs.writeFile(tempPath, payload, "utf8");
|
|
||||||
await fs.rename(tempPath, targetPath);
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -0,0 +1,25 @@
|
|||||||
|
import { randomUUID } from "node:crypto";
|
||||||
|
import { promises as fs } from "node:fs";
|
||||||
|
import path from "node:path";
|
||||||
|
|
||||||
|
export async function writeFileAtomic(
|
||||||
|
filePath: string,
|
||||||
|
data: string | NodeJS.ArrayBufferView,
|
||||||
|
): Promise<void> {
|
||||||
|
await fs.mkdir(path.dirname(filePath), { recursive: true });
|
||||||
|
const tempPath = path.join(
|
||||||
|
path.dirname(filePath),
|
||||||
|
`.${path.basename(filePath)}.${process.pid}.${Date.now()}.${randomUUID()}.tmp`,
|
||||||
|
);
|
||||||
|
try {
|
||||||
|
await fs.writeFile(tempPath, data, "utf8");
|
||||||
|
await fs.rename(tempPath, filePath);
|
||||||
|
} catch (error) {
|
||||||
|
await fs.rm(tempPath, { force: true });
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function writeJsonFileAtomic(filePath: string, value: unknown): Promise<void> {
|
||||||
|
await writeFileAtomic(filePath, JSON.stringify(value, null, 2));
|
||||||
|
}
|
||||||
@@ -3,6 +3,7 @@ import { promises as fs } from "node:fs";
|
|||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
import type pino from "pino";
|
import type pino from "pino";
|
||||||
import { z } from "zod";
|
import { z } from "zod";
|
||||||
|
import { writeJsonFileAtomic } from "../atomic-file.js";
|
||||||
import {
|
import {
|
||||||
ChatMessageSchema,
|
ChatMessageSchema,
|
||||||
ChatRoomDetailSchema,
|
ChatRoomDetailSchema,
|
||||||
@@ -364,10 +365,7 @@ export class FileBackedChatService {
|
|||||||
.flat()
|
.flat()
|
||||||
.sort((left, right) => left.createdAt.localeCompare(right.createdAt)),
|
.sort((left, right) => left.createdAt.localeCompare(right.createdAt)),
|
||||||
};
|
};
|
||||||
await fs.mkdir(path.dirname(this.filePath), { recursive: true });
|
await writeJsonFileAtomic(this.filePath, payload);
|
||||||
const tempPath = `${this.filePath}.${process.pid}.${Date.now()}.${randomUUID()}.tmp`;
|
|
||||||
await fs.writeFile(tempPath, JSON.stringify(payload, null, 2), "utf8");
|
|
||||||
await fs.rename(tempPath, this.filePath);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private findRoomByName(name: string): ChatRoom | null {
|
private findRoomByName(name: string): ChatRoom | null {
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ import {
|
|||||||
importSecretKey,
|
importSecretKey,
|
||||||
type KeyPair,
|
type KeyPair,
|
||||||
} from "@getpaseo/relay/e2ee";
|
} from "@getpaseo/relay/e2ee";
|
||||||
import { ensurePrivateFile, writePrivateFileSync } from "./private-files.js";
|
import { ensurePrivateFile, writePrivateFileAtomicSync } from "./private-files.js";
|
||||||
|
|
||||||
const KeyPairSchema = z.object({
|
const KeyPairSchema = z.object({
|
||||||
v: z.literal(2),
|
v: z.literal(2),
|
||||||
@@ -62,7 +62,7 @@ export async function loadOrCreateDaemonKeyPair(
|
|||||||
secretKeyB64,
|
secretKeyB64,
|
||||||
};
|
};
|
||||||
|
|
||||||
writePrivateFileSync(filePath, JSON.stringify(payload, null, 2) + "\n");
|
writePrivateFileAtomicSync(filePath, JSON.stringify(payload, null, 2) + "\n");
|
||||||
log?.info({ filePath }, "Saved daemon keypair");
|
log?.info({ filePath }, "Saved daemon keypair");
|
||||||
|
|
||||||
return { keyPair, publicKeyB64 };
|
return { keyPair, publicKeyB64 };
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ import { promises as fs } from "node:fs";
|
|||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
import { z } from "zod";
|
import { z } from "zod";
|
||||||
import type { Logger } from "pino";
|
import type { Logger } from "pino";
|
||||||
|
import { writeJsonFileAtomic } from "./atomic-file.js";
|
||||||
import { curateAgentActivity } from "./agent/activity-curator.js";
|
import { curateAgentActivity } from "./agent/activity-curator.js";
|
||||||
import type { AgentManager } from "./agent/agent-manager.js";
|
import type { AgentManager } from "./agent/agent-manager.js";
|
||||||
import { getStructuredAgentResponse } from "./agent/agent-response-loop.js";
|
import { getStructuredAgentResponse } from "./agent/agent-response-loop.js";
|
||||||
@@ -891,11 +892,10 @@ export class LoopService {
|
|||||||
|
|
||||||
private async persist(): Promise<void> {
|
private async persist(): Promise<void> {
|
||||||
const nextPersist = this.persistQueue.then(async () => {
|
const nextPersist = this.persistQueue.then(async () => {
|
||||||
await fs.mkdir(path.dirname(this.storePath), { recursive: true });
|
|
||||||
const records = Array.from(this.loops.values()).sort((left, right) =>
|
const records = Array.from(this.loops.values()).sort((left, right) =>
|
||||||
left.createdAt.localeCompare(right.createdAt),
|
left.createdAt.localeCompare(right.createdAt),
|
||||||
);
|
);
|
||||||
await fs.writeFile(this.storePath, JSON.stringify(records, null, 2), "utf8");
|
await writeJsonFileAtomic(this.storePath, records);
|
||||||
return;
|
return;
|
||||||
});
|
});
|
||||||
this.persistQueue = nextPersist.catch(() => {});
|
this.persistQueue = nextPersist.catch(() => {});
|
||||||
|
|||||||
@@ -8,7 +8,7 @@ import {
|
|||||||
ProviderOverridesSchema,
|
ProviderOverridesSchema,
|
||||||
} from "./agent/provider-launch-config.js";
|
} from "./agent/provider-launch-config.js";
|
||||||
import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js";
|
import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js";
|
||||||
import { ensurePrivateFile, writePrivateFileSync } from "./private-files.js";
|
import { ensurePrivateFile, writePrivateFileAtomicSync } from "./private-files.js";
|
||||||
|
|
||||||
export const LogLevelSchema = z.enum(["trace", "debug", "info", "warn", "error", "fatal"]);
|
export const LogLevelSchema = z.enum(["trace", "debug", "info", "warn", "error", "fatal"]);
|
||||||
export const LogFormatSchema = z.enum(["pretty", "json"]);
|
export const LogFormatSchema = z.enum(["pretty", "json"]);
|
||||||
@@ -357,7 +357,10 @@ export function loadPersistedConfig(paseoHome: string, logger?: LoggerLike): Per
|
|||||||
|
|
||||||
if (!existsSync(configPath)) {
|
if (!existsSync(configPath)) {
|
||||||
try {
|
try {
|
||||||
writePrivateFileSync(configPath, JSON.stringify(DEFAULT_PERSISTED_CONFIG, null, 2) + "\n");
|
writePrivateFileAtomicSync(
|
||||||
|
configPath,
|
||||||
|
JSON.stringify(DEFAULT_PERSISTED_CONFIG, null, 2) + "\n",
|
||||||
|
);
|
||||||
log?.info(`Initialized config file at ${configPath}`);
|
log?.info(`Initialized config file at ${configPath}`);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
const message = err instanceof Error ? err.message : String(err);
|
const message = err instanceof Error ? err.message : String(err);
|
||||||
@@ -416,7 +419,7 @@ export function savePersistedConfig(
|
|||||||
}
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
writePrivateFileSync(configPath, JSON.stringify(result.data, null, 2) + "\n");
|
writePrivateFileAtomicSync(configPath, JSON.stringify(result.data, null, 2) + "\n");
|
||||||
log?.info(`Saved to ${configPath}`);
|
log?.info(`Saved to ${configPath}`);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
const message = err instanceof Error ? err.message : String(err);
|
const message = err instanceof Error ? err.message : String(err);
|
||||||
|
|||||||
@@ -26,15 +26,6 @@ export function ensurePrivateFile(filePath: string): void {
|
|||||||
chmodBestEffort(filePath, PRIVATE_FILE_MODE);
|
chmodBestEffort(filePath, PRIVATE_FILE_MODE);
|
||||||
}
|
}
|
||||||
|
|
||||||
export function writePrivateFileSync(
|
|
||||||
filePath: string,
|
|
||||||
data: string | NodeJS.ArrayBufferView,
|
|
||||||
): void {
|
|
||||||
ensurePrivateDirectory(path.dirname(filePath));
|
|
||||||
writeFileSync(filePath, data, { mode: PRIVATE_FILE_MODE });
|
|
||||||
ensurePrivateFile(filePath);
|
|
||||||
}
|
|
||||||
|
|
||||||
export function writePrivateFileAtomicSync(
|
export function writePrivateFileAtomicSync(
|
||||||
filePath: string,
|
filePath: string,
|
||||||
data: string | NodeJS.ArrayBufferView,
|
data: string | NodeJS.ArrayBufferView,
|
||||||
|
|||||||
@@ -1,7 +1,8 @@
|
|||||||
import { randomBytes } from "node:crypto";
|
import { randomBytes } from "node:crypto";
|
||||||
import { mkdir, readFile, readdir, rm, writeFile } from "node:fs/promises";
|
import { mkdir, readFile, readdir, rm } from "node:fs/promises";
|
||||||
import { join } from "node:path";
|
import { join } from "node:path";
|
||||||
import { StoredScheduleSchema, type StoredSchedule } from "@getpaseo/protocol/schedule/types";
|
import { StoredScheduleSchema, type StoredSchedule } from "@getpaseo/protocol/schedule/types";
|
||||||
|
import { writeJsonFileAtomic } from "../atomic-file.js";
|
||||||
|
|
||||||
function generateScheduleId(): string {
|
function generateScheduleId(): string {
|
||||||
return randomBytes(4).toString("hex");
|
return randomBytes(4).toString("hex");
|
||||||
@@ -53,7 +54,7 @@ export class ScheduleStore {
|
|||||||
|
|
||||||
async put(schedule: StoredSchedule): Promise<void> {
|
async put(schedule: StoredSchedule): Promise<void> {
|
||||||
await this.ensureDir();
|
await this.ensureDir();
|
||||||
await writeFile(this.filePath(schedule.id), JSON.stringify(schedule, null, 2), "utf-8");
|
await writeJsonFileAtomic(this.filePath(schedule.id), schedule);
|
||||||
}
|
}
|
||||||
|
|
||||||
async delete(id: string): Promise<void> {
|
async delete(id: string): Promise<void> {
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ import path from "node:path";
|
|||||||
import { existsSync, readFileSync } from "node:fs";
|
import { existsSync, readFileSync } from "node:fs";
|
||||||
import { randomBytes } from "node:crypto";
|
import { randomBytes } from "node:crypto";
|
||||||
|
|
||||||
import { ensurePrivateFile, writePrivateFileSync } from "./private-files.js";
|
import { ensurePrivateFile, writePrivateFileAtomicSync } from "./private-files.js";
|
||||||
|
|
||||||
interface LoggerLike {
|
interface LoggerLike {
|
||||||
child(bindings: Record<string, unknown>): LoggerLike;
|
child(bindings: Record<string, unknown>): LoggerLike;
|
||||||
@@ -49,7 +49,7 @@ export function getOrCreateServerId(
|
|||||||
// Persist the override for consistent identity across restarts.
|
// Persist the override for consistent identity across restarts.
|
||||||
if (!existsSync(serverIdPath)) {
|
if (!existsSync(serverIdPath)) {
|
||||||
try {
|
try {
|
||||||
writePrivateFileSync(serverIdPath, `${envOverride}\n`);
|
writePrivateFileAtomicSync(serverIdPath, `${envOverride}\n`);
|
||||||
log?.info({ serverId: envOverride }, "Persisted PASEO_SERVER_ID override");
|
log?.info({ serverId: envOverride }, "Persisted PASEO_SERVER_ID override");
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
log?.warn({ error }, "Failed to persist PASEO_SERVER_ID override");
|
log?.warn({ error }, "Failed to persist PASEO_SERVER_ID override");
|
||||||
@@ -75,7 +75,7 @@ export function getOrCreateServerId(
|
|||||||
|
|
||||||
const created = generateServerId();
|
const created = generateServerId();
|
||||||
try {
|
try {
|
||||||
writePrivateFileSync(serverIdPath, `${created}\n`);
|
writePrivateFileAtomicSync(serverIdPath, `${created}\n`);
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
log?.warn({ error }, "Failed to persist serverId (continuing with in-memory id)");
|
log?.warn({ error }, "Failed to persist serverId (continuing with in-memory id)");
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,10 +1,9 @@
|
|||||||
import { randomUUID } from "node:crypto";
|
|
||||||
import { promises as fs } from "node:fs";
|
import { promises as fs } from "node:fs";
|
||||||
import path from "node:path";
|
|
||||||
|
|
||||||
import type { Logger } from "pino";
|
import type { Logger } from "pino";
|
||||||
import { z } from "zod";
|
import { z } from "zod";
|
||||||
|
|
||||||
|
import { writeJsonFileAtomic } from "./atomic-file.js";
|
||||||
import type { PersistedProjectKind, PersistedWorkspaceKind } from "./workspace-registry-model.js";
|
import type { PersistedProjectKind, PersistedWorkspaceKind } from "./workspace-registry-model.js";
|
||||||
|
|
||||||
const PersistedProjectRecordSchema = z.object({
|
const PersistedProjectRecordSchema = z.object({
|
||||||
@@ -161,10 +160,7 @@ class FileBackedRegistry<TRecord extends RegistryRecord> {
|
|||||||
|
|
||||||
private async persist(): Promise<void> {
|
private async persist(): Promise<void> {
|
||||||
const records = Array.from(this.cache.values());
|
const records = Array.from(this.cache.values());
|
||||||
await fs.mkdir(path.dirname(this.filePath), { recursive: true });
|
await writeJsonFileAtomic(this.filePath, records);
|
||||||
const tempPath = `${this.filePath}.${process.pid}.${Date.now()}.${randomUUID()}.tmp`;
|
|
||||||
await fs.writeFile(tempPath, JSON.stringify(records, null, 2), "utf8");
|
|
||||||
await fs.rename(tempPath, this.filePath);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private async enqueuePersist(): Promise<void> {
|
private async enqueuePersist(): Promise<void> {
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import { readFile, writeFile } from "node:fs/promises";
|
import { readFile } from "node:fs/promises";
|
||||||
|
import { writeFileAtomic } from "../server/atomic-file.js";
|
||||||
import type { Task, TaskStatus } from "./types.js";
|
import type { Task, TaskStatus } from "./types.js";
|
||||||
|
|
||||||
function serializeTask(task: Task): string {
|
function serializeTask(task: Task): string {
|
||||||
@@ -130,5 +131,5 @@ export async function readTaskDocument(filePath: string): Promise<Task> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export async function writeTaskDocument(filePath: string, task: Task): Promise<void> {
|
export async function writeTaskDocument(filePath: string, task: Task): Promise<void> {
|
||||||
await writeFile(filePath, serializeTask(task), "utf-8");
|
await writeFileAtomic(filePath, serializeTask(task));
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user