feat: append lines in batches to the shared log and clear it #1066

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()),
       };

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', () => {

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');

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));
       },

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<void>;
+  /**
+   * 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<void>;
+  /** Remove every line readAll returns (local: the current file; rotated files stay). */
+  clear(): Promise<void>;
   /** Every line, oldest first (local: the current file only, as today). */
   readAll(): Promise<string[]>;
   length(): Promise<number>;