Conat -- replacing NATS with socketio + sqlite3 + valkey + nodejs

.github/workflows/make-and-test.yml

@@ -65,7 +65,8 @@ jobs:
       matrix:
         # See supported Node.js release schedule at https://nodejs.org/en/about/releases/
         node-version:
-          - "18.17.1"
+          - "18"
+          - "20"
         pg-version:
           - "13.12"
           - "16"
@@ -93,5 +94,4 @@ jobs:
 
       - run: cd src && npm install -g pnpm
       - run: cd src && pnpm run make
-      - run: cd src && pnpm nats-server-daemon
       - run: cd src && pnpm run test

src/compute/compute/lib/filesystem.ts

@@ -22,7 +22,7 @@ import { apiCall } from "@cocalc/api-client";
 import sendFiles from "./send-files";
 import getFiles from "./get-files";
 // ensure that the nats client is initialized so that syncfs can connect to nats properly.
-import "@cocalc/project/nats";
+import "@cocalc/project/conat";
 
 const logger = getLogger("compute:filesystem");
 

src/compute/compute/lib/manager.ts

@@ -7,7 +7,7 @@ The manager does the following:
 */
 
 import debug from "debug";
-import startProjectServers from "@cocalc/project/nats";
+import startProjectServers from "@cocalc/project/conat";
 import { pingProjectUntilSuccess, waitUntilFilesystemIsOfType } from "./util";
 import { apiCall, project } from "@cocalc/api-client";
 

src/compute/compute/package.json

@@ -32,7 +32,7 @@
     "@cocalc/backend": "workspace:*",
     "@cocalc/compute": "link:",
     "@cocalc/jupyter": "workspace:*",
-    "@cocalc/nats": "workspace:*",
+    "@cocalc/conat": "workspace:*",
     "@cocalc/project": "workspace:*",
     "@cocalc/sync": "workspace:*",
     "@cocalc/sync-client": "workspace:*",

src/compute/conat

@@ -0,0 +1 @@
+../packages/conat

src/compute/pnpm-workspace.yaml

@@ -3,7 +3,7 @@ packages:
   - backend
   - comm
   - jupyter
-  - nats
+  - conat
   - project
   - sync
   - sync-fs

src/package.json

@@ -19,15 +19,8 @@
     "test-parallel": "unset DEBUG && pnpm run version-check && cd packages && pnpm run -r --parallel test",
     "test": "unset DEBUG && pnpm run version-check && cd packages && pnpm run -r test",
     "prettier-all": "cd packages/",
-    "nats-server": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/install').main()\" && node -e \"require('@cocalc/backend/nats/conf').main()\" && node -e \"require('@cocalc/backend/nats/server').main()\"",
-    "build-nats": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/util && pnpm install && pnpm build && cd ${COCALC_ROOT:=$INIT_CWD}/packages/nats && pnpm install && pnpm build && cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && pnpm install && pnpm build",
-    "nats-server-ci": "pnpm run build-nats && cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/install').main()\" && node -e \"require('@cocalc/backend/nats/conf').main()\" && node -e \"require('@cocalc/backend/nats/server').main()\"",
-    "nats-server-daemon": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/install').main()\" && node -e \"require('@cocalc/backend/nats/conf').main()\" && node -e \"require('@cocalc/backend/nats/server').main({daemon:true})\"",
-    "nats-server-verbose": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/install').main()\" && node -e \"require('@cocalc/backend/nats/conf').main()\" && node -e \"require('@cocalc/backend/nats/server').main({verbose:true})\"",
-    "nats-cli": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/cli').main()\"",
-    "nats-sys": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/cli').main({user:'sys'})\"",
-    "nats-tiered-storage": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/server && DEBUG=cocalc:* DEBUG_CONSOLE=yes node -e \"require('@cocalc/server/nats/tiered-storage').init()\"",
-    "local-ci": "./scripts/ci.sh"
+    "local-ci": "./scripts/ci.sh",
+    "conat-server": "cd packages/server && pnpm conat-server"
   },
   "repository": {
     "type": "git",

src/packages/backend/bin/conat-watch.cjs

@@ -0,0 +1,10 @@
+const { getEnv } = require('@cocalc/backend/conat')
+
+async function main() {
+    const subject = process.argv[2] ?? '>';
+    console.log("watching ", {subject})
+    const {cn} = await getEnv()
+    cn.watch(subject)
+}
+
+main();

src/packages/backend/conat/conat.ts

@@ -0,0 +1,15 @@
+import { conatPath, conatServer } from "@cocalc/backend/data";
+import {
+  connect as connect0,
+  Client,
+  type ConnectOptions,
+} from "@cocalc/conat/core/client";
+
+export type { Client };
+
+export function connect(address?, options?: ConnectOptions): Client {
+  return connect0(address ? address : conatServer, {
+    path: conatPath,
+    ...options,
+  });
+}

src/packages/backend/conat/env.ts

@@ -0,0 +1,9 @@
+import { sha1 } from "@cocalc/backend/sha1";
+import { connect as getConatClient } from "./conat";
+
+export async function getEnv(options?) {
+  const jc = null as any;
+  const nc = null as any;
+  const cn = getConatClient(options);
+  return { nc, jc, cn, sha1 };
+}

src/packages/backend/nats/index.ts → src/packages/backend/conat/index.ts

@@ -9,17 +9,17 @@ import { readFile } from "node:fs/promises";
 import getLogger from "@cocalc/backend/logger";
 import { getEnv } from "./env";
 export { getEnv };
-import { inboxPrefix } from "@cocalc/nats/names";
-import { setNatsClient } from "@cocalc/nats/client";
+import { inboxPrefix } from "@cocalc/conat/names";
+import { setConatClient } from "@cocalc/conat/client";
 import getConnection, {
   setConnectionOptions,
-} from "@cocalc/backend/nats/persistent-connection";
+} from "@cocalc/backend/conat/persistent-connection";
 import { hostname } from "os";
 
 export { getConnection };
 
 export function init() {
-  setNatsClient({ getNatsEnv: getEnv, getLogger });
+  setConatClient({ getNatsEnv: getEnv, getLogger });
 }
 init();
 

src/packages/backend/conat/persist.ts

@@ -0,0 +1,21 @@
+import "./index";
+import betterSqlite3 from "better-sqlite3";
+import { initContext } from "@cocalc/conat/persist/context";
+import { compress, decompress } from "zstd-napi";
+import { syncFiles } from "@cocalc/backend/data";
+import ensureContainingDirectoryExists from "@cocalc/backend/misc/ensure-containing-directory-exists";
+
+initContext({
+  betterSqlite3,
+  compress,
+  decompress,
+  syncFiles,
+  ensureContainingDirectoryExists,
+});
+
+export { pstream } from "@cocalc/conat/persist/storage";
+export {
+  init as initServer,
+  terminate as terminateServer,
+} from "@cocalc/conat/persist/server";
+export { getAll, set, get } from "@cocalc/conat/persist/client";

src/packages/backend/conat/persistent-connection.ts

@@ -0,0 +1,42 @@
+/*
+Create a nats connection that doesn't break.
+
+The NATS docs
+
+https://github.com/nats-io/nats.js/blob/main/core/README.md#connecting-to-a-nats-server
+
+ensure us that "the client will always attempt to reconnect if the connection is
+disrupted for a reason other than calling close()" but THAT IS NOT TRUE.
+(I think the upstream code in disconnected in nats.js/core/src/protocol.ts is a lazy
+and I disagree with it.  It tries to connect but if anything goes slightly wrong,
+just gives up forever.)
+
+There are definitely situations where the connection gets permanently closed
+and the close() function was not called, at least not by any of our code.
+I've given up on getting them to fix or understand their bugs in general:
+
+https://github.com/williamstein/nats-bugs/issues/8
+
+We thus monitor the connection, and if it closed, we *swap out the protocol
+object*, which is an evil hack to reconnect. This seems to work fine with all
+our other code.
+
+All that said, it's excellent that the NATS library separates the protocol from
+the connection object itself, so it's possible to do this at all! :-)
+*/
+
+import { reuseInFlight } from "@cocalc/util/reuse-in-flight";
+import type { NatsConnection } from "@cocalc/conat/types";
+
+export function setConnectionOptions(_: any) {}
+
+// gets the singleton connection
+const getConnection = reuseInFlight(async (): Promise<NatsConnection> => {
+  return null as any;
+});
+
+export default getConnection;
+
+export async function getNewConnection(): Promise<NatsConnection> {
+  return null as any;
+}

src/packages/backend/conat/sync.ts

@@ -0,0 +1,32 @@
+import {
+  dstream as createDstream,
+  type DStream,
+} from "@cocalc/conat/sync/dstream";
+import { dkv as createDKV, type DKV } from "@cocalc/conat/sync/dkv";
+import { dko as createDKO, type DKO } from "@cocalc/conat/sync/dko";
+import { akv as createAKV, type AKV } from "@cocalc/conat/sync/akv";
+import { createOpenFiles, type OpenFiles } from "@cocalc/conat/sync/open-files";
+export { inventory } from "@cocalc/conat/sync/inventory";
+import "./index";
+
+export type { DStream, DKV, DKO, AKV };
+
+export async function dstream<T = any>(opts): Promise<DStream<T>> {
+  return await createDstream<T>(opts);
+}
+
+export async function dkv<T = any>(opts): Promise<DKV<T>> {
+  return await createDKV<T>(opts);
+}
+
+export function akv<T = any>(opts): AKV<T> {
+  return createAKV<T>(opts);
+}
+
+export async function dko<T = any>(opts): Promise<DKO<T>> {
+  return await createDKO(opts);
+}
+
+export async function openFiles(project_id: string, opts?): Promise<OpenFiles> {
+  return await createOpenFiles({ project_id, ...opts });
+}