diff --git a/lib/admin/__tests__/shared-store-fake.ts b/lib/admin/__tests__/shared-store-fake.ts index 56e40bcad..d9309a02c 100644 --- a/lib/admin/__tests__/shared-store-fake.ts +++ b/lib/admin/__tests__/shared-store-fake.ts @@ -36,6 +36,8 @@ export function separateFakeStore(root: string): SharedStore & { down: boolean } const log = localLog(name, { ...options, local: { ...options.local, path: () => path.join(root, 'logs', `${name}.log`) } }); return { append: (line: string) => guard(() => log.append(line)), + appendMany: (lines: string[]) => guard(() => log.appendMany(lines)), + clear: () => guard(() => log.clear()), readAll: () => guard(() => log.readAll()), length: () => guard(() => log.length()), }; diff --git a/lib/shared-store/__tests__/contract.ts b/lib/shared-store/__tests__/contract.ts index 055b685e1..2a496bb49 100644 --- a/lib/shared-store/__tests__/contract.ts +++ b/lib/shared-store/__tests__/contract.ts @@ -307,6 +307,33 @@ export function runStoreContract(target: ContractTarget): void { await log.append('one\ntwo'); expect(await log.readAll()).toEqual(['one two']); }); + + it('appends a batch in order, with no other writer\'s line inside it', async () => { + const [a, b] = await open(2); + const batch = Array.from({ length: 200 }, (_, i) => `batch-${i}`); + await Promise.all([ + a.log(`audit-${run}`, logOptions()).appendMany(batch), + ...Array.from({ length: 20 }, (_, i) => b.log(`audit-${run}`, logOptions()).append(`single-${i}`)), + ]); + const lines = await a.log(`audit-${run}`, logOptions()).readAll(); + expect(lines).toHaveLength(220); + const start = lines.indexOf('batch-0'); + expect(lines.slice(start, start + 200)).toEqual(batch); + await a.log(`audit-${run}`, logOptions()).appendMany([]); + expect(await b.log(`audit-${run}`, logOptions()).length()).toBe(220); + }); + + it('clears the log for every replica', async () => { + const [a, b] = await open(2); + await a.log(`audit-${run}`, logOptions()).appendMany(['one', 'two\nthree']); + expect(await b.log(`audit-${run}`, logOptions()).readAll()).toEqual(['one', 'two three']); + await b.log(`audit-${run}`, logOptions()).clear(); + expect(await a.log(`audit-${run}`, logOptions()).readAll()).toEqual([]); + expect(await a.log(`audit-${run}`, logOptions()).length()).toBe(0); + await b.log(`audit-${run}`, logOptions()).clear(); + await a.log(`audit-${run}`, logOptions()).append('again'); + expect(await b.log(`audit-${run}`, logOptions()).readAll()).toEqual(['again']); + }); }); describe('leases', () => { diff --git a/lib/shared-store/local.ts b/lib/shared-store/local.ts index 781174a6e..7f5372f96 100644 --- a/lib/shared-store/local.ts +++ b/lib/shared-store/local.ts @@ -402,6 +402,23 @@ export function localLog(name: string, options: LogOptions): AppendLog { await rotateIfNeeded(file); }); }, + async appendMany(lines) { + if (lines.length === 0) return; + const file = path.resolve(local.path()); + await serializeLocal(`log\0${file}`, async () => { + await mkdir(path.dirname(file), { recursive: true }); + await appendFile(file, lines.map((line) => line.replace(/[\r\n]+/g, ' ') + '\n').join(''), 'utf-8'); + await rotateIfNeeded(file); + }); + }, + async clear() { + const file = path.resolve(local.path()); + await serializeLocal(`log\0${file}`, async () => { + await unlink(file).catch((error) => { + if (!isMissing(error)) throw error; + }); + }); + }, async readAll() { try { const content = await readFile(path.resolve(local.path()), 'utf-8'); diff --git a/lib/shared-store/redis.ts b/lib/shared-store/redis.ts index 5483f4e5d..63b9ac562 100644 --- a/lib/shared-store/redis.ts +++ b/lib/shared-store/redis.ts @@ -353,6 +353,14 @@ export function createRedisStore(options: RedisStoreOptions): SharedStore { const clean = line.replace(/[\r\n]+/g, ' '); await call(() => client.multi().rpush(key, clean).ltrim(key, -opts.maxEntries, -1).exec()); }, + async appendMany(lines) { + if (lines.length === 0) return; + const clean = lines.map((line) => line.replace(/[\r\n]+/g, ' ')); + await call(() => client.multi().rpush(key, ...clean).ltrim(key, -opts.maxEntries, -1).exec()); + }, + async clear() { + await call(() => client.del(key)); + }, async readAll() { return call(() => client.lrange(key, 0, -1)); }, diff --git a/lib/shared-store/types.ts b/lib/shared-store/types.ts index d0fd47eaf..8c77d5502 100644 --- a/lib/shared-store/types.ts +++ b/lib/shared-store/types.ts @@ -205,6 +205,13 @@ export interface LogOptions { export interface AppendLog { /** One line, without a newline. Lines are never interleaved or torn. */ append(line: string): Promise; + /** + * Several lines in one call, in this order; no other writer's line lands + * between them. For bulk copies such as the import. + */ + appendMany(lines: string[]): Promise; + /** Remove every line readAll returns (local: the current file; rotated files stay). */ + clear(): Promise; /** Every line, oldest first (local: the current file only, as today). */ readAll(): Promise; length(): Promise;