conat: update dstream-ephemeral tests (and fix a bug)

Author: williamstein

src/packages/backend/conat/test/core/core-stream-ephemeral.test.ts

@@ -249,6 +249,7 @@ describe("test creating a non-leader first, then the leader", () => {
   });
 });
 
+// There a lot more similar tests of dstream ephemeral in backend/conat/test/sync/dstream-ephemeral.test.ts
 describe("test using ephemeral dstream", () => {
   let client;
   let stream;

src/packages/backend/conat/test/sync/estream.test.ts renamed to src/packages/backend/conat/test/sync/dstream-ephemeral.test.ts

@@ -6,16 +6,25 @@ The first tests are initially similar to those for dstream.test.ts, but with
 
 DEVELOPMENT:
 
-pnpm test estream.test.ts
+pnpm test ./dstream-estream.test.ts
 
 */
 
+import { connect, before, after } from "@cocalc/backend/conat/test/setup";
 import { createDstreamEphemeral as create } from "./util";
 import { dstream as createDstream0 } from "@cocalc/backend/conat/sync";
 import { once } from "@cocalc/util/async-utils";
 
+beforeAll(before);
+
 async function createDstream<T>(opts) {
-  return await createDstream0<T>({ ephemeral: true, leader: true, ...opts });
+  return await createDstream0<T>({
+    noCache: true,
+    noAutosave: true,
+    ephemeral: true,
+    leader: true,
+    ...opts,
+  });
 }
 
 describe("create a dstream and do some basic operations", () => {
@@ -65,14 +74,11 @@ describe("create a dstream and do some basic operations", () => {
 describe("create two dstreams and observe sync between them", () => {
   const name = `test-${Math.random()}`;
   let s1, s2;
+  let client2;
   it("creates two distinct dstream objects s1 and s2 with the same name", async () => {
-    s1 = await createDstream({ name, noAutosave: true, noCache: true });
-    s2 = await createDstream({
-      name,
-      noAutosave: true,
-      noCache: true,
-      leader: false,
-    });
+    client2 = connect();
+    s1 = await createDstream({ name });
+    s2 = await createDstream({ client: client2, name, leader: false });
     // definitely distinct
     expect(s1 === s2).toBe(false);
   });
@@ -108,21 +114,50 @@ describe("create two dstreams and observe sync between them", () => {
     expect(s2.getAll()).toEqual(["hello", "hi from s2"]);
   });
 
-  it("write to s1 and s2 and save at the same time and see some 'random choice' of order gets imposed by the server", async () => {
+  it("cleans up", () => {
+    s1.close();
+    s2.close();
+    client2.close();
+  });
+});
+
+describe("create two dstreams and test sync with parallel save", () => {
+  const name = `test-${Math.random()}`;
+  let s1, s2;
+  let client2;
+  it("creates two distinct dstream objects s1 and s2 with the same name", async () => {
+    client2 = connect();
+    s1 = await createDstream({ name });
+    s2 = await createDstream({ client: client2, name, leader: false });
+    // definitely distinct
+    expect(s1 === s2).toBe(false);
+  });
+
+  it("write to s1 and s2 and save at the same time", async () => {
     s1.push("s1");
     s2.push("s2");
     // our changes are reflected locally
-    expect(s1.getAll()).toEqual(["hello", "hi from s2", "s1"]);
-    expect(s2.getAll()).toEqual(["hello", "hi from s2", "s2"]);
+    expect(s1.getAll()).toEqual(["s1"]);
+    expect(s2.getAll()).toEqual(["s2"]);
     // now kick off the two saves *in parallel*
     s1.save();
     s2.save();
     await once(s1, "change");
-    while (s2.length != s1.length) {
-      await once(s2, "change");
+    while (s2.length != 2 || s1.length != 2) {
+      if (s1.length > 2 || s2.length > 2) {
+        throw Error("bug");
+      }
+      if (s2.length < 2) {
+        await once(s2, "change");
+      } else if (s1.length < 2) {
+        await once(s1, "change");
+      }
     }
     expect(s1.getAll()).toEqual(s2.getAll());
-    expect(s1.getAll()).toEqual(["hello", "hi from s2", "s1", "s2"]);
+  });
+
+  it("cleans up", () => {
+    client2.close();
   });
 });
 
@@ -195,7 +230,9 @@ describe("testing start_seq", () => {
 
   let t;
   it("it opens another copy of the stream, but starting with the last sequence number, so only one message", async () => {
+    const client = connect();
     t = await createDstream({
+      client,
       name,
       noAutosave: true,
       leader: false,
@@ -227,8 +264,9 @@ describe("a little bit of a stress test", () => {
       s.push({ i });
     }
     expect(s.length).toBe(count);
-    // NOTE: warning -- this is **MUCH SLOWER**, e.g., 10x slower,
-    // running under jest, hence why count is small.
+    // [ ] TODO rewrite this save to send everything in a single message
+    // which gets chunked, will we be much faster, then change the count
+    // above to 1000 or 10000.
     await s.save();
     expect(s.length).toBe(count);
   });
@@ -249,50 +287,52 @@ describe("dstream typescript test", () => {
   });
 });
 
-import { numSubscriptions } from "@cocalc/conat/client";
+describe("ensure there isn't a really obvious subscription leak", () => {
+  let client;
 
-describe("ensure there are no NATS subscription leaks", () => {
-  // There is some slight slack at some point due to the clock stuff,
-  // inventory, etc.  It is constant and small, whereas we allocate
-  // a large number of kv's in the test.
-  const SLACK = 4;
+  it("create a client, which initially has only one subscription (the inbox)", async () => {
+    client = connect();
+    expect(client.numSubscriptions()).toBe(1);
+  });
 
-  it("creates and closes many kv's and checks there is no leak", async () => {
-    const before = numSubscriptions();
-    const COUNT = 20;
+  const count = 100;
+  it(`creates and closes ${count} streams and checks there is no leak`, async () => {
+    const before = client.numSubscriptions();
     // create
     const a: any = [];
-    for (let i = 0; i < COUNT; i++) {
+    for (let i = 0; i < count; i++) {
       a[i] = await createDstream({
         name: `${Math.random()}`,
-        noAutosave: true,
       });
     }
-    for (let i = 0; i < COUNT; i++) {
+    for (let i = 0; i < count; i++) {
       await a[i].close();
     }
-    const after = numSubscriptions();
-    expect(Math.abs(after - before)).toBeLessThan(SLACK);
+    const after = client.numSubscriptions();
+    expect(after).toBe(before);
+
+    // also check count on server went down.
+    expect((await client.getSubscriptions()).size).toBe(before);
   });
 
-  it("does another leak test, but with a set operation each time", async () => {
-    const before = numSubscriptions();
-    const COUNT = 20;
+  it("does another leak test, but with a publish operation each time", async () => {
+    const before = client.numSubscriptions();
     // create
     const a: any = [];
-    for (let i = 0; i < COUNT; i++) {
+    for (let i = 0; i < count; i++) {
       a[i] = await createDstream({
         name: `${Math.random()}`,
         noAutosave: true,
       });
       a[i].publish(i);
       await a[i].save();
     }
-    for (let i = 0; i < COUNT; i++) {
-      await a[i].purge();
+    for (let i = 0; i < count; i++) {
       await a[i].close();
     }
-    const after = numSubscriptions();
-    expect(Math.abs(after - before)).toBeLessThan(SLACK);
+    const after = client.numSubscriptions();
+    expect(after).toBe(before);
   });
 });
+
+afterAll(after);

src/packages/conat/core/client.ts

@@ -395,6 +395,8 @@ export class Client {
     }
   };
 
+  numSubscriptions = () => Object.keys(this.queueGroups).length;
+
   private getSubscriptions = async (): Promise<Set<string>> => {
     const f = (cb) =>
       this.conn.emit("subscriptions", null, (subs) => cb(undefined, subs));

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

@@ -67,7 +67,6 @@ export const ENFORCE_LIMITS_THROTTLE_MS = process.env.COCALC_TEST_MODE
 const HEADER_PREFIX = "CoCalc-";
 
 export const COCALC_MESSAGE_ID_HEADER = `${HEADER_PREFIX}Msg-Id`;
-
 export const COCALC_STREAM_HEADER = `${HEADER_PREFIX}Stream`;
 export const COCALC_OPTIONS_HEADER = `${HEADER_PREFIX}Options`;
 export const COCALC_HEARTBEAT_HEADER = `${HEADER_PREFIX}Heartbeat`;
@@ -618,8 +617,8 @@ export class CoreStream<T = any> extends EventEmitter {
           seq,
           timestamp,
           sessionId: this.sessionId,
-          msgID: options?.msgID,
         },
+        [COCALC_MESSAGE_ID_HEADER]: options?.msgID,
       } as any;
       if (options?.headers) {
         for (const k in options.headers) {
@@ -719,7 +718,7 @@ export class CoreStream<T = any> extends EventEmitter {
 
   // get server assigned time of n-th message in stream
   time = (n: number): Date | undefined => {
-    const r = this.raw[n][0];
+    const r = this.raw[n]?.[0];
     if (r == null) {
       return;
     }

src/packages/conat/sync/dstream.ts

@@ -45,6 +45,7 @@ import { waitUntilConnected } from "@cocalc/conat/util";
 import { type Msg } from "@nats-io/nats-core";
 import { headersFromRawMessages } from "./stream";
 import { COCALC_MESSAGE_ID_HEADER } from "./core-stream";
+import type { Client } from "@cocalc/conat/core/client";
 
 const logger = getLogger("dstream");
 
@@ -56,6 +57,7 @@ export interface DStreamOptions extends StreamOptions {
   ephemeral?: boolean;
   persist?: boolean;
   leader?: boolean;
+  client?: Client;
 }
 
 export class DStream<T = any> extends EventEmitter {
@@ -111,7 +113,7 @@ export class DStream<T = any> extends EventEmitter {
       delete this.saved[last(raw).seq];
       const headers = headersFromRawMessages(raw);
       if (headers?.[COCALC_MESSAGE_ID_HEADER]) {
-        // this is critical with conat-stream.ts, since otherwise there is a moment
+        // this is critical with core-stream.ts, since otherwise there is a moment
         // when the same message is in both this.local *and* this.messages, and you'll
         // see it doubled in this.getAll().  I didn't see this ever with
         // stream.ts, but maybe it is possible.  It probably wouldn't impact any application,