optimize delete for streams by batching; use this in conat router

- after several 4 days in production with significant load, we hit a
  scalability issue with trimming the interest stream.  This should
  fully address that by making it 200K times faster...

Author: williamstein

src/packages/backend/conat/test/sync/dstream-stress.test.ts

@@ -0,0 +1,51 @@
+import { dstream } from "@cocalc/backend/conat/sync";
+import { before, after, client, wait } from "@cocalc/backend/conat/test/setup";
+
+const log = process.env.VERBOSE ? console.log : (..._args) => {};
+
+beforeAll(before);
+
+jest.setTimeout(10000);
+
+describe("a stress test", () => {
+  const name = `test-${Math.random()}`;
+  const pushCount = 2000;
+  let s;
+  it(`creates an ephemeral stream and pushes ${pushCount} messages`, async () => {
+    const start = Date.now();
+    s = await dstream({
+      client,
+      name,
+      noAutosave: true,
+      ephemeral: true,
+    });
+    for (let i = 0; i < pushCount; i++) {
+      s.push({ i });
+    }
+    expect(s.length).toBe(pushCount);
+    // NOTE: warning -- this is **MUCH SLOWER**, e.g., 10x slower,
+    // running under jest, hence why count is small.
+    await s.save();
+    expect(s.length).toBe(pushCount);
+    log(
+      "write",
+      Math.round((1000 * pushCount) / (Date.now() - start)),
+      "messages per second",
+    );
+  });
+
+  it("deletes all of the messages we just wrote", async () => {
+    const start = Date.now();
+    await s.delete({ seqs: s.seqs() });
+    await s.save();
+    await wait({ until: () => s.length == 0 });
+    expect(s.length).toBe(0);
+    log(
+      "delete",
+      Math.round((1000 * pushCount) / (Date.now() - start)),
+      "messages per second",
+    );
+  });
+});
+
+afterAll(after);

src/packages/backend/conat/test/sync/dstream.test.ts

@@ -110,7 +110,7 @@ describe("create two dstreams and observe sync between them", () => {
   });
 });
 
-describe.only("get sequence number and time of message", () => {
+describe("get sequence number and time of message", () => {
   let s;
 
   it("creates stream and write message", async () => {
@@ -413,6 +413,17 @@ describe("test delete of messages from stream", () => {
     await wait({ until: () => s1.length == 2 });
     expect(s1.get()).toEqual(["x", "y"]);
   });
+
+  it("delete an array of sequence numbers", async () => {
+    s1.push("x", "y", "z");
+    await s1.save();
+    const v = await s1.getAll();
+    const seqs0 = [s1.seq(0), s1.seq(2)];
+    const { seqs } = await s1.delete({ seqs: seqs0 });
+    expect(seqs).toEqual(seqs0);
+    await wait({ until: () => s1.length == v.length - 2 });
+    expect(s1.getAll()).toEqual([v[1]].concat(v.slice(3)));
+  });
 });
 
 afterAll(after);

src/packages/conat/core/cluster.ts

@@ -296,9 +296,8 @@ export async function trimClusterStreams(
   if (seqs.length > 0) {
     // [ ] todo -- add to interest.delete a version where it takes an array of sequence numbers
     logger.debug("trimClusterStream: trimming interest", { seqs });
-    for (const seq of seqs) {
-      await interest.delete({ seq });
-    }
+    await interest.delete({ seqs });
+    logger.debug("trimClusterStream: successfully trimmed interest", { seqs });
   }
 
   // Next deal with sticky -- trim ones where the pattern is no longer of interest.
@@ -335,9 +334,8 @@ export async function trimClusterStreams(
   if (seqs2.length > 0) {
     // [ ] todo -- add to interest.delete a version where it takes an array of sequence numbers
     logger.debug("trimClusterStream: trimming sticky", { seqs2 });
-    for (const seq of seqs2) {
-      await sticky.delete({ seq });
-    }
+    await sticky.delete({ seqs: seqs2 });
+    logger.debug("trimClusterStream: successfully trimmed sticky", { seqs2 });
   }
 
   return { seqsInterest: seqs, seqsSticky: seqs2 };

src/packages/conat/persist/client.ts

@@ -278,11 +278,13 @@ class PersistStreamClient extends EventEmitter {
   delete = async ({
     timeout,
     seq,
+    seqs,
     last_seq,
     all,
   }: {
     timeout?: number;
     seq?: number;
+    seqs?: number[];
     last_seq?: number;
     all?: boolean;
   }): Promise<{ seqs: number[] }> => {
@@ -291,6 +293,7 @@ class PersistStreamClient extends EventEmitter {
         headers: {
           cmd: "delete",
           seq,
+          seqs,
           last_seq,
           all,
           timeout,

src/packages/conat/persist/storage.ts

@@ -367,7 +367,9 @@ export class PersistentStream extends EventEmitter {
     try {
       await this.db.backup(path);
     } catch (err) {
-      console.log(err);
+      if (!process.env.COCALC_TEST_MODE) {
+        console.log(err);
+      }
       logger.debug("WARNING: error creating a backup", path, err);
     }
   });
@@ -537,10 +539,12 @@ export class PersistentStream extends EventEmitter {
 
   delete = ({
     seq,
+    seqs: seqs0,
     last_seq,
     all,
   }: {
     seq?: number;
+    seqs?: number[];
     last_seq?: number;
     all?: boolean;
   }): { seqs: number[] } => {
@@ -565,6 +569,15 @@ export class PersistentStream extends EventEmitter {
         .all(seq)
         .map((row: any) => row.seq);
       this.db.prepare("DELETE FROM messages WHERE seq=?").run(seq);
+    } else if (seqs0) {
+      const statement = this.db.prepare("DELETE FROM messages WHERE seq=?");
+      const transaction = this.db.transaction((seqs) => {
+        for (const s of seqs) {
+          statement.run(s);
+        }
+      });
+      transaction(seqs0);
+      seqs = seqs0;
     }
     this.emit("change", { op: "delete", seqs });
     this.throttledBackup();

src/packages/conat/sync/core-stream.ts

@@ -936,6 +936,7 @@ export class CoreStream<T = any> extends EventEmitter {
     seq,
     last_seq,
     key,
+    seqs,
   }: {
     // give exactly ONE parameter -- by default nothing happens with no params
     // all: delete everything
@@ -944,6 +945,8 @@ export class CoreStream<T = any> extends EventEmitter {
     last_index?: number;
     // seq: delete message with this sequence number
     seq?: number;
+    // seqs: delete the messages in this array of sequence numbers
+    seqs?: number[];
     // last_seq: delete everything up to and including this sequence number
     last_seq?: number;
     // key: delete the message with this key
@@ -966,6 +969,8 @@ export class CoreStream<T = any> extends EventEmitter {
       }
     } else if (seq != null) {
       opts = { seq };
+    } else if (seqs != null) {
+      opts = { seqs };
     } else if (last_seq != null) {
       opts = { last_seq };
     } else if (key != null) {