mirror of
https://github.com/MoonshotAI/kimi-code.git
synced 2026-08-22 15:16:07 +00:00
fix(agent-core-v2): pin durable file generations
This commit is contained in:
qer
parent
087ac412e5
commit
7e157286cc
3 changed files with 351 additions and 105 deletions
|
|
@ -5,8 +5,16 @@
|
|||
* deletion, and watching through the local filesystem. Bound at App scope.
|
||||
*/
|
||||
|
||||
import { constants, createReadStream, mkdirSync } from 'node:fs';
|
||||
import { lstat, mkdir, open, readFile, readdir, stat, unlink } from 'node:fs/promises';
|
||||
import {
|
||||
close as closeFd,
|
||||
closeSync,
|
||||
constants,
|
||||
createReadStream,
|
||||
fstat as fstatFd,
|
||||
mkdirSync,
|
||||
open as openFd,
|
||||
} from 'node:fs';
|
||||
import { lstat, mkdir, open, readFile, readdir, unlink } from 'node:fs/promises';
|
||||
import { FSWatcher } from 'chokidar';
|
||||
import { dirname, join, normalize } from 'pathe';
|
||||
|
||||
|
|
@ -24,6 +32,7 @@ import type {
|
|||
import { toStorageIoError } from '#/persistence/interface/storage';
|
||||
|
||||
const WATCH_DEBOUNCE_MS = 150;
|
||||
const MAX_DURABLE_ENTRIES = 64;
|
||||
|
||||
function isEnoent(error: unknown): boolean {
|
||||
return (error as NodeJS.ErrnoException).code === 'ENOENT';
|
||||
|
|
@ -34,41 +43,53 @@ function isEexist(error: unknown): boolean {
|
|||
}
|
||||
|
||||
interface FileIdentity {
|
||||
readonly birthtimeNs: bigint;
|
||||
readonly dev: bigint;
|
||||
readonly ino: bigint;
|
||||
}
|
||||
|
||||
function fileIdentity(stats: {
|
||||
readonly birthtimeNs: bigint;
|
||||
readonly dev: bigint;
|
||||
readonly ino: bigint;
|
||||
}): FileIdentity | undefined {
|
||||
return stats.ino === 0n || stats.birthtimeNs === 0n
|
||||
? undefined
|
||||
: { birthtimeNs: stats.birthtimeNs, dev: stats.dev, ino: stats.ino };
|
||||
return stats.ino === 0n ? undefined : { dev: stats.dev, ino: stats.ino };
|
||||
}
|
||||
|
||||
function sameFile(left: FileIdentity | undefined, right: FileIdentity | undefined): boolean {
|
||||
return (
|
||||
left !== undefined &&
|
||||
right !== undefined &&
|
||||
left.birthtimeNs === right.birthtimeNs &&
|
||||
left.dev === right.dev &&
|
||||
left.ino === right.ino
|
||||
);
|
||||
}
|
||||
|
||||
interface DurableEntry extends FileIdentity {
|
||||
readonly fd: number;
|
||||
}
|
||||
|
||||
const durableEntryFinalizer = new FinalizationRegistry<Map<string, DurableEntry>>((entries) => {
|
||||
for (const entry of entries.values()) {
|
||||
try {
|
||||
closeSync(entry.fd);
|
||||
} catch (error) {
|
||||
onUnexpectedError(error);
|
||||
}
|
||||
}
|
||||
entries.clear();
|
||||
});
|
||||
|
||||
export class FileStorageService implements IFileSystemStorageService {
|
||||
declare readonly _serviceBrand: undefined;
|
||||
|
||||
private readonly durableEntries = new Map<string, FileIdentity>();
|
||||
private readonly durableEntries = new Map<string, DurableEntry>();
|
||||
|
||||
constructor(
|
||||
private readonly baseDir: string,
|
||||
private readonly dirMode?: number,
|
||||
private readonly fileMode?: number,
|
||||
) {}
|
||||
) {
|
||||
durableEntryFinalizer.register(this, this.durableEntries);
|
||||
}
|
||||
|
||||
async read(scope: string, key: string): Promise<Uint8Array | undefined> {
|
||||
const filePath = this.path(scope, key);
|
||||
|
|
@ -109,11 +130,9 @@ export class FileStorageService implements IFileSystemStorageService {
|
|||
const filePath = this.path(scope, key);
|
||||
try {
|
||||
await mkdir(dirname(filePath), { recursive: true, mode: this.dirMode });
|
||||
this.durableEntries.delete(filePath);
|
||||
await this.removeDurableEntry(filePath);
|
||||
await atomicWrite(filePath, data, undefined, this.fileMode);
|
||||
const identity = fileIdentity(await stat(filePath, { bigint: true }));
|
||||
await syncDir(dirname(filePath));
|
||||
this.markDurable(filePath, identity);
|
||||
} catch (error) {
|
||||
throw toStorageIoError(error, { path: filePath, op: 'write' });
|
||||
}
|
||||
|
|
@ -134,7 +153,6 @@ export class FileStorageService implements IFileSystemStorageService {
|
|||
while (true) {
|
||||
try {
|
||||
fh = await open(filePath, 'ax', this.fileMode);
|
||||
this.durableEntries.delete(filePath);
|
||||
break;
|
||||
} catch (error) {
|
||||
if (!isEexist(error)) throw error;
|
||||
|
|
@ -154,7 +172,6 @@ export class FileStorageService implements IFileSystemStorageService {
|
|||
if (entry.isSymbolicLink()) throw error;
|
||||
}
|
||||
}
|
||||
let identity: FileIdentity | undefined;
|
||||
try {
|
||||
if (data.byteLength > 0) {
|
||||
await fh.writeFile(data);
|
||||
|
|
@ -162,14 +179,29 @@ export class FileStorageService implements IFileSystemStorageService {
|
|||
if (options.durable !== false) {
|
||||
await fh.sync();
|
||||
}
|
||||
identity = fileIdentity(await fh.stat({ bigint: true }));
|
||||
const identity = fileIdentity(await fh.stat({ bigint: true }));
|
||||
const durableEntry = this.durableEntries.get(filePath);
|
||||
if (
|
||||
durableEntry !== undefined &&
|
||||
sameFile(durableEntry, identity) &&
|
||||
this.touchDurableEntry(filePath, durableEntry)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
await syncDir(dir);
|
||||
if (identity === undefined) {
|
||||
await this.removeDurableEntry(filePath);
|
||||
} else {
|
||||
const entry = await this.openDurableEntry(filePath, identity);
|
||||
if (entry === undefined) {
|
||||
await this.removeDurableEntry(filePath);
|
||||
} else {
|
||||
await this.installDurableEntry(filePath, entry);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await fh.close();
|
||||
}
|
||||
if (!sameFile(this.durableEntries.get(filePath), identity)) {
|
||||
await syncDir(dir);
|
||||
this.markDurable(filePath, identity);
|
||||
}
|
||||
} catch (error) {
|
||||
throw toStorageIoError(error, { path: filePath, op: 'append' });
|
||||
}
|
||||
|
|
@ -189,13 +221,20 @@ export class FileStorageService implements IFileSystemStorageService {
|
|||
async delete(scope: string, key: string): Promise<void> {
|
||||
const filePath = this.path(scope, key);
|
||||
try {
|
||||
await unlink(filePath);
|
||||
this.durableEntries.delete(filePath);
|
||||
await this.removeDurableEntry(filePath);
|
||||
} catch (error) {
|
||||
if (isEnoent(error)) {
|
||||
this.durableEntries.delete(filePath);
|
||||
return;
|
||||
throw toStorageIoError(error, { path: filePath, op: 'delete' });
|
||||
}
|
||||
try {
|
||||
await unlink(filePath);
|
||||
} catch (error) {
|
||||
if (!isEnoent(error)) {
|
||||
throw toStorageIoError(error, { path: filePath, op: 'delete' });
|
||||
}
|
||||
}
|
||||
try {
|
||||
await this.removeDurableEntry(filePath);
|
||||
} catch (error) {
|
||||
throw toStorageIoError(error, { path: filePath, op: 'delete' });
|
||||
}
|
||||
}
|
||||
|
|
@ -266,7 +305,23 @@ export class FileStorageService implements IFileSystemStorageService {
|
|||
|
||||
async flush(): Promise<void> {}
|
||||
|
||||
async close(): Promise<void> {}
|
||||
dispose(): void {
|
||||
void this.close().catch(onUnexpectedError);
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
const entries = [...this.durableEntries.entries()];
|
||||
this.durableEntries.clear();
|
||||
let firstError: Error | undefined;
|
||||
for (const [filePath, entry] of entries) {
|
||||
try {
|
||||
await closeAnchor(entry.fd);
|
||||
} catch (error) {
|
||||
firstError ??= toStorageIoError(error, { path: filePath, op: 'close' });
|
||||
}
|
||||
}
|
||||
if (firstError !== undefined) throw firstError;
|
||||
}
|
||||
|
||||
private path(scope: string, key: string): string {
|
||||
return join(this.baseDir, scope, key);
|
||||
|
|
@ -276,11 +331,81 @@ export class FileStorageService implements IFileSystemStorageService {
|
|||
return join(this.baseDir, scope);
|
||||
}
|
||||
|
||||
private markDurable(filePath: string, identity: FileIdentity | undefined): void {
|
||||
if (identity === undefined) {
|
||||
private touchDurableEntry(filePath: string, entry: DurableEntry): boolean {
|
||||
if (this.durableEntries.get(filePath) !== entry) return false;
|
||||
this.durableEntries.delete(filePath);
|
||||
this.durableEntries.set(filePath, entry);
|
||||
return true;
|
||||
}
|
||||
|
||||
private async installDurableEntry(filePath: string, entry: DurableEntry): Promise<void> {
|
||||
const entriesToClose: DurableEntry[] = [];
|
||||
const previous = this.durableEntries.get(filePath);
|
||||
if (previous !== undefined) {
|
||||
this.durableEntries.delete(filePath);
|
||||
} else {
|
||||
this.durableEntries.set(filePath, identity);
|
||||
if (previous.fd !== entry.fd) entriesToClose.push(previous);
|
||||
}
|
||||
if (this.durableEntries.size >= MAX_DURABLE_ENTRIES) {
|
||||
const oldestPath = this.durableEntries.keys().next().value;
|
||||
if (oldestPath !== undefined) {
|
||||
const oldest = this.durableEntries.get(oldestPath);
|
||||
this.durableEntries.delete(oldestPath);
|
||||
if (oldest !== undefined && oldest.fd !== entry.fd) entriesToClose.push(oldest);
|
||||
}
|
||||
}
|
||||
this.durableEntries.set(filePath, entry);
|
||||
for (const entryToClose of entriesToClose) {
|
||||
try {
|
||||
await closeAnchor(entryToClose.fd);
|
||||
} catch (error) {
|
||||
onUnexpectedError(error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async removeDurableEntry(filePath: string): Promise<void> {
|
||||
const entry = this.durableEntries.get(filePath);
|
||||
if (entry === undefined) return;
|
||||
this.durableEntries.delete(filePath);
|
||||
await closeAnchor(entry.fd);
|
||||
}
|
||||
|
||||
private openDurableEntry(
|
||||
filePath: string,
|
||||
expectedIdentity: FileIdentity,
|
||||
): Promise<DurableEntry | undefined> {
|
||||
return new Promise((resolve) => {
|
||||
openFd(filePath, constants.O_WRONLY, (openError, fd) => {
|
||||
if (openError !== null) {
|
||||
if (!isEnoent(openError)) onUnexpectedError(openError);
|
||||
resolve(undefined);
|
||||
return;
|
||||
}
|
||||
fstatFd(fd, { bigint: true }, (statError, stats) => {
|
||||
if (statError !== null) {
|
||||
void closeAnchor(fd).catch(onUnexpectedError);
|
||||
if (!isEnoent(statError)) onUnexpectedError(statError);
|
||||
resolve(undefined);
|
||||
return;
|
||||
}
|
||||
const identity = fileIdentity(stats);
|
||||
if (identity === undefined || !sameFile(identity, expectedIdentity)) {
|
||||
void closeAnchor(fd).catch(onUnexpectedError);
|
||||
resolve(undefined);
|
||||
return;
|
||||
}
|
||||
resolve({ fd, ...identity });
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function closeAnchor(fd: number): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
closeFd(fd, (error) => {
|
||||
if (error !== null) reject(error);
|
||||
else resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,8 +3,8 @@
|
|||
* temporary files and a controlled directory-fsync boundary.
|
||||
*/
|
||||
|
||||
import { constants } from 'node:fs';
|
||||
import { mkdtemp, mkdir, open, rm, stat, symlink, writeFile } from 'node:fs/promises';
|
||||
import { constants, type BigIntStats } from 'node:fs';
|
||||
import { mkdtemp, mkdir, open, rm, stat, symlink, unlink, writeFile } from 'node:fs/promises';
|
||||
import { tmpdir } from 'node:os';
|
||||
|
||||
import { join } from 'pathe';
|
||||
|
|
@ -17,6 +17,22 @@ const fsBoundary = vi.hoisted(() => ({
|
|||
syncDir: vi.fn<(dir: string) => Promise<void>>(),
|
||||
}));
|
||||
|
||||
const fsAnchorBoundary = vi.hoisted(() => ({
|
||||
close: vi.fn<typeof import('node:fs').close>(),
|
||||
fstat: vi.fn<typeof import('node:fs').fstat>(),
|
||||
open: vi.fn<typeof import('node:fs').open>(),
|
||||
}));
|
||||
|
||||
vi.mock('node:fs', async (importOriginal) => {
|
||||
const original = await importOriginal<typeof import('node:fs')>();
|
||||
return {
|
||||
...original,
|
||||
close: fsAnchorBoundary.close,
|
||||
fstat: fsAnchorBoundary.fstat,
|
||||
open: fsAnchorBoundary.open,
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock('node:fs/promises', async (importOriginal) => {
|
||||
const original = await importOriginal<typeof import('node:fs/promises')>();
|
||||
return { ...original, open: fsBoundary.open };
|
||||
|
|
@ -31,7 +47,14 @@ const isWin = process.platform === 'win32';
|
|||
const encoder = new TextEncoder();
|
||||
|
||||
beforeEach(async () => {
|
||||
const originalFs = await vi.importActual<typeof import('node:fs')>('node:fs');
|
||||
const original = await vi.importActual<typeof import('node:fs/promises')>('node:fs/promises');
|
||||
fsAnchorBoundary.close.mockReset();
|
||||
fsAnchorBoundary.close.mockImplementation(originalFs.close);
|
||||
fsAnchorBoundary.fstat.mockReset();
|
||||
fsAnchorBoundary.fstat.mockImplementation(originalFs.fstat);
|
||||
fsAnchorBoundary.open.mockReset();
|
||||
fsAnchorBoundary.open.mockImplementation(originalFs.open);
|
||||
fsBoundary.open.mockReset();
|
||||
fsBoundary.open.mockImplementation(original.open);
|
||||
fsBoundary.syncDir.mockReset();
|
||||
|
|
@ -55,17 +78,40 @@ function fsError(code: string): Error & { code: string } {
|
|||
return Object.assign(new Error(code), { code });
|
||||
}
|
||||
|
||||
function servicePool(): {
|
||||
create(baseDir: string, dirMode?: number, fileMode?: number): FileStorageService;
|
||||
close(): Promise<void>;
|
||||
} {
|
||||
const services: FileStorageService[] = [];
|
||||
return {
|
||||
create(baseDir, dirMode, fileMode) {
|
||||
const service = new FileStorageService(baseDir, dirMode, fileMode);
|
||||
services.push(service);
|
||||
return service;
|
||||
},
|
||||
async close() {
|
||||
await Promise.all(services.splice(0).map((service) => service.close()));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe('FileStorageService — durable directory entries', () => {
|
||||
let dir: string;
|
||||
let service: FileStorageService;
|
||||
let services: ReturnType<typeof servicePool>;
|
||||
|
||||
beforeEach(async () => {
|
||||
dir = await mkdtemp(join(tmpdir(), 'fss-durable-'));
|
||||
service = new FileStorageService(dir);
|
||||
services = servicePool();
|
||||
service = services.create(dir);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
try {
|
||||
await services.close();
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('syncs the directory when a second key is atomically created in the same scope', async () => {
|
||||
|
|
@ -89,7 +135,7 @@ describe('FileStorageService — durable directory entries', () => {
|
|||
});
|
||||
|
||||
it('waits for directory durability when two instances first append the same log', async () => {
|
||||
const other = new FileStorageService(dir);
|
||||
const other = services.create(dir);
|
||||
const firstEntered = deferred();
|
||||
const secondEntered = deferred();
|
||||
const release = deferred();
|
||||
|
|
@ -124,73 +170,105 @@ describe('FileStorageService — durable directory entries', () => {
|
|||
expect(fsBoundary.syncDir).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('resyncs the directory when another instance recreates a known log', async () => {
|
||||
const other = new FileStorageService(dir);
|
||||
const probe = await open(join(dir, 'probe'), 'a');
|
||||
const fileHandlePrototype = Object.getPrototypeOf(probe) as {
|
||||
stat(options: { bigint: true }): Promise<{
|
||||
birthtimeNs: bigint;
|
||||
dev: bigint;
|
||||
ino: bigint;
|
||||
}>;
|
||||
};
|
||||
await probe.close();
|
||||
const fileStat = vi
|
||||
.spyOn(fileHandlePrototype, 'stat')
|
||||
.mockResolvedValueOnce({ birthtimeNs: 10n, dev: 1n, ino: 7n })
|
||||
.mockResolvedValueOnce({ birthtimeNs: 20n, dev: 1n, ino: 7n })
|
||||
.mockResolvedValueOnce({ birthtimeNs: 20n, dev: 1n, ino: 7n });
|
||||
const release = deferred();
|
||||
let recreate: Promise<void> | undefined;
|
||||
let appendFromStaleInstance: Promise<void> | undefined;
|
||||
|
||||
try {
|
||||
await service.append('scope', 'wire.jsonl', encoder.encode('old\n'));
|
||||
await other.delete('scope', 'wire.jsonl');
|
||||
fsBoundary.syncDir.mockClear();
|
||||
|
||||
const firstEntered = deferred();
|
||||
const secondEntered = deferred();
|
||||
let entries = 0;
|
||||
fsBoundary.syncDir.mockImplementation(async () => {
|
||||
entries++;
|
||||
if (entries === 1) firstEntered.resolve();
|
||||
if (entries === 2) secondEntered.resolve();
|
||||
await release.promise;
|
||||
it.skipIf(isWin)(
|
||||
'resyncs a recreated log even when the filesystem would recycle its inode',
|
||||
async () => {
|
||||
const other = services.create(dir);
|
||||
const filePath = join(dir, 'scope', 'wire.jsonl');
|
||||
const handles = new WeakMap<object, number>();
|
||||
let openedHandles = 0;
|
||||
const anchorGenerations = new Map<number, bigint>();
|
||||
const originalFs = await vi.importActual<typeof import('node:fs')>('node:fs');
|
||||
const originalPromises = await vi.importActual<typeof import('node:fs/promises')>(
|
||||
'node:fs/promises',
|
||||
);
|
||||
const openAnchorForTest = (
|
||||
path: string,
|
||||
flags: number,
|
||||
callback: (error: NodeJS.ErrnoException | null, fd: number) => void,
|
||||
): void => {
|
||||
originalFs.open(path, flags, (error, fd) => {
|
||||
if (error !== null) {
|
||||
callback(error, fd);
|
||||
return;
|
||||
}
|
||||
anchorGenerations.set(fd, anchorGenerations.size === 0 ? 7n : 8n);
|
||||
callback(null, fd);
|
||||
});
|
||||
};
|
||||
fsAnchorBoundary.open.mockImplementation(openAnchorForTest as typeof originalFs.open);
|
||||
const fstatAnchorForTest = (
|
||||
fd: number,
|
||||
options: { bigint: true },
|
||||
callback: (error: NodeJS.ErrnoException | null, stats: BigIntStats) => void,
|
||||
): void => {
|
||||
originalFs.fstat(fd, options, (error, stats) => {
|
||||
if (error !== null) {
|
||||
callback(error, stats);
|
||||
return;
|
||||
}
|
||||
const generation = anchorGenerations.get(fd) ?? 8n;
|
||||
callback(null, { ...stats, dev: 1n, ino: generation } as BigIntStats);
|
||||
});
|
||||
};
|
||||
fsAnchorBoundary.fstat.mockImplementation(fstatAnchorForTest as typeof originalFs.fstat);
|
||||
const closeAnchorForTest = (
|
||||
fd: number,
|
||||
callback: (error?: NodeJS.ErrnoException | null) => void,
|
||||
): void => {
|
||||
anchorGenerations.delete(fd);
|
||||
originalFs.close(fd, callback);
|
||||
};
|
||||
fsAnchorBoundary.close.mockImplementation(closeAnchorForTest as typeof originalFs.close);
|
||||
fsBoundary.open.mockImplementation(async (...args) => {
|
||||
const handle = await originalPromises.open(...args);
|
||||
if (args[0] === filePath) {
|
||||
const ordinal = ++openedHandles;
|
||||
handles.set(handle, ordinal);
|
||||
}
|
||||
return handle;
|
||||
});
|
||||
recreate = other.append('scope', 'wire.jsonl', encoder.encode('new\n'));
|
||||
await firstEntered.promise;
|
||||
appendFromStaleInstance = service.append(
|
||||
'scope',
|
||||
'wire.jsonl',
|
||||
encoder.encode('later\n'),
|
||||
);
|
||||
|
||||
const beforeDurability = await Promise.race([
|
||||
secondEntered.promise.then(() => 'syncing'),
|
||||
appendFromStaleInstance.then(() => 'succeeded'),
|
||||
]);
|
||||
expect(beforeDurability).toBe('syncing');
|
||||
const probe = await open(join(dir, 'probe'), 'a');
|
||||
const fileHandlePrototype = Object.getPrototypeOf(probe) as {
|
||||
stat(options: { bigint: true }): Promise<{
|
||||
birthtimeNs: bigint;
|
||||
dev: bigint;
|
||||
ino: bigint;
|
||||
}>;
|
||||
};
|
||||
await probe.close();
|
||||
const fileStat = vi.spyOn(fileHandlePrototype, 'stat').mockImplementation(async function (
|
||||
this: object,
|
||||
) {
|
||||
const ordinal = handles.get(this);
|
||||
if (ordinal === undefined) return { birthtimeNs: 1n, dev: 1n, ino: 1n };
|
||||
return {
|
||||
birthtimeNs: 10n,
|
||||
dev: 1n,
|
||||
ino: anchorGenerations.size === 0 ? 7n : 8n,
|
||||
};
|
||||
});
|
||||
|
||||
release.resolve();
|
||||
await Promise.all([recreate, appendFromStaleInstance]);
|
||||
expect(fsBoundary.syncDir).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
release.resolve();
|
||||
await Promise.allSettled(
|
||||
[recreate, appendFromStaleInstance].filter(
|
||||
(operation): operation is Promise<void> => operation !== undefined,
|
||||
),
|
||||
);
|
||||
fileStat.mockRestore();
|
||||
}
|
||||
});
|
||||
try {
|
||||
await service.append('scope', 'wire.jsonl', encoder.encode('old\n'));
|
||||
await unlink(filePath);
|
||||
fsBoundary.syncDir.mockClear();
|
||||
|
||||
await other.append('scope', 'wire.jsonl', encoder.encode('new\n'));
|
||||
await service.append('scope', 'wire.jsonl', encoder.encode('later\n'));
|
||||
|
||||
expect(fsBoundary.syncDir).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
fileStat.mockRestore();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it('resyncs each append when the filesystem exposes no stable file identity', async () => {
|
||||
const probe = await open(join(dir, 'probe'), 'a');
|
||||
const fileHandlePrototype = Object.getPrototypeOf(probe) as {
|
||||
stat(options: { bigint: true }): Promise<{
|
||||
birthtimeNs: bigint;
|
||||
dev: bigint;
|
||||
ino: bigint;
|
||||
}>;
|
||||
|
|
@ -198,7 +276,7 @@ describe('FileStorageService — durable directory entries', () => {
|
|||
await probe.close();
|
||||
const fileStat = vi
|
||||
.spyOn(fileHandlePrototype, 'stat')
|
||||
.mockResolvedValue({ birthtimeNs: 0n, dev: 0n, ino: 0n });
|
||||
.mockResolvedValue({ dev: 0n, ino: 0n });
|
||||
|
||||
try {
|
||||
await service.append('scope', 'wire.jsonl', encoder.encode('first\n'));
|
||||
|
|
@ -209,6 +287,30 @@ describe('FileStorageService — durable directory entries', () => {
|
|||
}
|
||||
});
|
||||
|
||||
it('resyncs an evicted log when the anchor cache reaches its limit', async () => {
|
||||
for (let index = 0; index < 65; index++) {
|
||||
await service.append('scope', `wire-${index}.jsonl`, encoder.encode('first\n'));
|
||||
}
|
||||
fsBoundary.syncDir.mockClear();
|
||||
|
||||
await service.append('scope', 'wire-0.jsonl', encoder.encode('second\n'));
|
||||
|
||||
expect(fsBoundary.syncDir).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it('releases cached append anchors when storage closes', async () => {
|
||||
await service.append('scope', 'wire.jsonl', encoder.encode('first\n'));
|
||||
fsBoundary.syncDir.mockClear();
|
||||
await service.append('scope', 'wire.jsonl', encoder.encode('second\n'));
|
||||
expect(fsBoundary.syncDir).not.toHaveBeenCalled();
|
||||
|
||||
await service.close();
|
||||
fsBoundary.syncDir.mockClear();
|
||||
await service.append('scope', 'wire.jsonl', encoder.encode('third\n'));
|
||||
|
||||
expect(fsBoundary.syncDir).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it('reclaims a log that disappears before the non-creating append open', async () => {
|
||||
fsBoundary.open
|
||||
.mockRejectedValueOnce(fsError('EEXIST'))
|
||||
|
|
@ -346,17 +448,23 @@ describe('FileStorageService — durable directory entries', () => {
|
|||
|
||||
describe('FileStorageService — file permissions', () => {
|
||||
let dir: string;
|
||||
let services: ReturnType<typeof servicePool>;
|
||||
|
||||
beforeEach(async () => {
|
||||
dir = await mkdtemp(join(tmpdir(), 'fss-perm-'));
|
||||
services = servicePool();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
try {
|
||||
await services.close();
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.skipIf(isWin)('creates scope directories with dirMode (0700)', async () => {
|
||||
const svc = new FileStorageService(dir, 0o700, 0o600);
|
||||
const svc = services.create(dir, 0o700, 0o600);
|
||||
await svc.write('cron/ws', 'abc.json', encoder.encode('{}'));
|
||||
|
||||
const dirStat = await stat(join(dir, 'cron/ws'));
|
||||
|
|
@ -364,7 +472,7 @@ describe('FileStorageService — file permissions', () => {
|
|||
});
|
||||
|
||||
it.skipIf(isWin)('writes documents with fileMode (0600)', async () => {
|
||||
const svc = new FileStorageService(dir, 0o700, 0o600);
|
||||
const svc = services.create(dir, 0o700, 0o600);
|
||||
await svc.write('cron/ws', 'abc.json', encoder.encode('{"x":1}'));
|
||||
|
||||
const fileStat = await stat(join(dir, 'cron/ws', 'abc.json'));
|
||||
|
|
@ -374,7 +482,7 @@ describe('FileStorageService — file permissions', () => {
|
|||
it.skipIf(isWin)('defaults to the process umask when modes are omitted', async () => {
|
||||
// Backwards compatibility: an unconfigured FileStorageService must not
|
||||
// start tightening permissions on its own — bootstrap opts into 0700/0600.
|
||||
const svc = new FileStorageService(dir);
|
||||
const svc = services.create(dir);
|
||||
await svc.write('scope', 'k.json', encoder.encode('{}'));
|
||||
const fileStat = await stat(join(dir, 'scope', 'k.json'));
|
||||
// Owner-read/write is always set; we only assert the file is readable by
|
||||
|
|
@ -385,24 +493,30 @@ describe('FileStorageService — file permissions', () => {
|
|||
|
||||
describe('FileStorageService — error translation', () => {
|
||||
let dir: string;
|
||||
let services: ReturnType<typeof servicePool>;
|
||||
|
||||
beforeEach(async () => {
|
||||
dir = await mkdtemp(join(tmpdir(), 'fss-err-'));
|
||||
services = servicePool();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
try {
|
||||
await services.close();
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('keeps ENOENT semantics: read returns undefined, list returns []', async () => {
|
||||
const svc = new FileStorageService(dir);
|
||||
const svc = services.create(dir);
|
||||
expect(await svc.read('scope', 'missing.json')).toBeUndefined();
|
||||
expect(await svc.list('missing-scope')).toEqual([]);
|
||||
await expect(svc.delete('scope', 'missing.json')).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it.skipIf(isWin)('translates non-ENOENT failures into StorageError(io_failed)', async () => {
|
||||
const svc = new FileStorageService(dir);
|
||||
const svc = services.create(dir);
|
||||
// Reading a directory fails with EISDIR — an I/O failure, not a miss.
|
||||
await mkdir(join(dir, 'scope', 'adir'), { recursive: true });
|
||||
await expect(svc.read('scope', 'adir')).rejects.toSatisfy((error: unknown) => {
|
||||
|
|
@ -419,7 +533,7 @@ describe('FileStorageService — error translation', () => {
|
|||
});
|
||||
|
||||
it.skipIf(isWin)('translates write failures into StorageError(io_failed)', async () => {
|
||||
const svc = new FileStorageService(dir);
|
||||
const svc = services.create(dir);
|
||||
// A file blocks the scope directory: mkdir('<dir>/blocked/k') fails
|
||||
// (EEXIST/ENOTDIR depending on platform and fs implementation).
|
||||
await writeFile(join(dir, 'blocked'), 'x');
|
||||
|
|
|
|||
|
|
@ -138,8 +138,15 @@ storageServiceSuite('InMemoryStorageService', async () => ({
|
|||
|
||||
storageServiceSuite('FileStorageService', async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'storage-service-test-'));
|
||||
const service = new FileStorageService(dir);
|
||||
return {
|
||||
service: new FileStorageService(dir),
|
||||
cleanup: () => rm(dir, { recursive: true, force: true }),
|
||||
service,
|
||||
cleanup: async () => {
|
||||
try {
|
||||
await service.close();
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
},
|
||||
};
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue