conat dstream: fix some bugs and also start getting testing framework for persistent working

Author: williamstein

src/packages/backend/conat/test/setup.ts

@@ -9,6 +9,14 @@ import { Server } from "socket.io";
 import getLogger from "@cocalc/backend/logger";
 import { setNatsClient } from "@cocalc/conat/client";
 import { sha1 } from "@cocalc/backend/sha1";
+import {
+  initServer as initPersistServer,
+  terminateServer as terminatePersistServer,
+} from "@cocalc/backend/conat/persist";
+import { syncFiles } from "@cocalc/conat/persist/context";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "path";
 
 const logger = getLogger("conat:test:setup");
 
@@ -33,10 +41,16 @@ export async function initConatServer(
 export let server;
 export let port;
 export let address;
+export let tempDir;
+
 export async function before() {
+  tempDir = await mkdtemp(join(tmpdir(), "conat-test"));
   port = await getPort();
   address = `http://localhost:${port}`;
   server = await initConatServer({ port, path });
+  syncFiles.local = join(tempDir, "local");
+  syncFiles.archive = join(tempDir, "archive");
+  initPersistServer({ client: server.client() });
   setNatsClient({
     getNatsEnv: async () => {
       return { cn: connect(), sha1 } as any;
@@ -53,6 +67,8 @@ export function connect() {
 }
 
 export async function after() {
+  terminatePersistServer();
+  await rm(tempDir, { force: true, recursive: true });
   await server.close();
   for (const cn of clients) {
     cn.close();

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

@@ -1,15 +1,14 @@
 /*
-Testing basic ops with dsteam (distributed streams)
+Testing basic ops with *persistent* dstreams.
 
 DEVELOPMENT:
 
-pnpm test dstream.test.ts
+pnpm test ./dstream.test.ts
 
 */
 
 import { createDstream as create } from "./util";
 import { dstream as createDstream } from "@cocalc/backend/conat/sync";
-import { once } from "@cocalc/util/async-utils";
 
 describe("create a dstream and do some basic operations", () => {
   let s;
@@ -55,6 +54,7 @@ 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;
@@ -155,7 +155,8 @@ describe("closing also saves by default, but not if autosave is off", () => {
   const name = `test-${Math.random()}`;
 
   it("creates stream and write a message", async () => {
-    s = await createDstream({ name, noAutosave: false /* the default */ });
+    // noAutosave: false is the default:
+    s = await createDstream({ name, noAutosave: false});
     s.push(389);
   });
 
@@ -299,3 +300,4 @@ describe("ensure there are no NATS subscription leaks", () => {
     expect(Math.abs(after - before)).toBeLessThan(SLACK);
   });
 });
+*/

src/packages/conat/persist/server.ts

@@ -42,7 +42,7 @@ Also getAll using start_seq:
 
 import { pstream, type Message as StoredMessage } from "./storage";
 import { getEnv } from "@cocalc/conat/client";
-import { type Subscription } from "@cocalc/conat/core/client";
+import { type Client, type Subscription } from "@cocalc/conat/core/client";
 import { uuid } from "@cocalc/util/misc";
 import { getLogger } from "@cocalc/conat/client";
 import { delay } from "awaiting";
@@ -67,9 +67,7 @@ export const LAST_CHUNK = "last-chunk";
 
 const logger = getLogger("persist:server");
 
-export const SUBJECT = process.env.COCALC_TEST_MODE
-  ? "persist-test"
-  : "persist";
+export const SUBJECT = "persist";
 
 export type User = { account_id?: string; project_id?: string };
 export function persistSubject({ account_id, project_id }: User) {
@@ -108,7 +106,7 @@ function getUserId(subject: string): string {
 
 let terminated = false;
 let sub: Subscription | null = null;
-export function init() {
+export async function init({ client }: { client?: Client } = {}) {
   logger.debug("starting persist server");
   logger.debug({
     DEFAULT_LIFETIME,
@@ -120,8 +118,9 @@ export function init() {
     MAX_PERSISTS_PER_SERVER,
     SUBJECT,
   });
-  persistService();
-  renewService();
+  client = client ?? (await getEnv()).cn;
+  persistService({ client });
+  renewService({ client });
 }
 
 async function noThrow(f) {
@@ -132,16 +131,14 @@ async function noThrow(f) {
   }
 }
 
-async function persistService() {
-  const { cn } = await getEnv();
-  sub = await cn.subscribe(`${SUBJECT}.*.api`, { queue: "q" });
+async function persistService({ client }) {
+  sub = await client.subscribe(`${SUBJECT}.*.api`, { queue: "q" });
   await listenPersist();
 }
 
 let renew: Subscription | null = null;
-async function renewService() {
-  const { cn } = await getEnv();
-  renew = await cn.subscribe(`${SUBJECT}.*.renew`);
+async function renewService({ client }) {
+  renew = await client.subscribe(`${SUBJECT}.*.renew`);
   await listenRenew();
 }
 
@@ -383,16 +380,15 @@ async function getAll({ mesg, request, user_id, stream }) {
   }
 
   try {
-    console.log("send the id", { id, lifetime });
     await respond(undefined, { id, lifetime });
     if (done) {
       return;
     }
 
     // send the current data
     // [ ] TODO: should we just send it all as a single message?
-    //     much faster, but uses much more RAM.  Maybe some
-    //     combination based on actual data!
+    //     much faster, but uses much more RAM.  **Instead, obviously
+    //     some combination based on actual data!**
     for (const message of stream.getAll({ start_seq: request.start_seq })) {
       if (done) {
         return;

src/packages/conat/persist/storage.ts

@@ -74,7 +74,7 @@ export interface Message {
 }
 
 export interface Options {
-  // relative path to sqlite database file.  This needs to be a valid filename
+  // absolute path to sqlite database file.  This needs to be a valid filename
   // path, and must also be kept under 1K so it can be stored in cloud storage.
   path: string;
   // if false (the default) do not require sync writes to disk on every set

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

@@ -259,7 +259,7 @@ export class CoreStream<T = any> extends EventEmitter {
       storage: this.storage,
       start_seq,
     });
-    console.log("getAll got ", { id });
+    console.log("got persistent stream", { id });
     this.persistStream = stream;
     while (true) {
       const { value, done } = await stream.next();
@@ -388,17 +388,24 @@ export class CoreStream<T = any> extends EventEmitter {
   private serveUntilDone = async (sub) => {
     for await (const raw of sub) {
       if (raw.subject.endsWith(".all")) {
+        // batch get
+
         const { start_seq = 0 } = raw.data ?? {};
 
+        // put exactly the entire data the client needs to get updated
+        // into a single payload
         const payload = this.raw
           .filter((x) => x[0].seq >= start_seq)
           .map((x) => {
             const { headers, encoding, raw } = x[0];
             return { headers, encoding, raw };
           });
 
+        // send it out as a single response.
         raw.respond(payload);
       } else if (raw.subject.endsWith(".send")) {
+        // single send:  ([ ] TODO need to support a batch send)
+
         const options = raw.headers?.[COCALC_OPTIONS_HEADER];
         let resp;
         try {
@@ -445,12 +452,7 @@ export class CoreStream<T = any> extends EventEmitter {
       return;
     }
     if (this.persist) {
-      if (this.persistStream == null) {
-        throw Error("persistentStream must be defined");
-      }
-      for await (const m of this.persistStream) {
-        this.processPersistentMessage(m, false);
-      }
+      this.listenLoopPersist();
       return;
     } else {
       this.sub = await this.client.subscribe(this.subject);
@@ -459,6 +461,19 @@ export class CoreStream<T = any> extends EventEmitter {
     this.enforceLimits();
   };
 
+  private listenLoopPersist = async () => {
+    if (this.persistStream == null) {
+      throw Error("persistentStream must be defined");
+    }
+    for await (const m of this.persistStream) {
+      try {
+        this.processPersistentMessage(m, false);
+      } catch (err) {
+        console.log(`WARNING: issue processing persistent message -- ${err}`);
+      }
+    }
+  };
+
   private listenLoop = async () => {
     if (this.sub == null) {
       throw Error("subscription must be setup");