hub database changefeeds -- implementing distributed locking

Author: williamstein

src/packages/database/nats/changefeeds.ts

@@ -46,7 +46,7 @@ import { reuseInFlight } from "@cocalc/util/reuse-in-flight";
 import { uuid } from "@cocalc/util/misc";
 import { delay } from "awaiting";
 import { Svcm } from "@nats-io/services";
-import { Coordinator } from "./coordinator";
+import { Coordinator, now } from "./coordinator";
 
 const logger = getLogger("database:nats:changefeeds");
 
@@ -55,9 +55,7 @@ const jc = JSONCodec();
 let api: any | null = null;
 export async function init() {
   const subject = "hub.*.*.db";
-  logger.debug(`init -- subject='${subject}', options=`, {
-    queue: "0",
-  });
+  logger.debug(`init -- subject='${subject}', options=`);
   const coordinator = new Coordinator();
   const nc = await getConnection();
 
@@ -68,12 +66,18 @@ export async function init() {
     name: "db-server",
     version: "0.2.0",
     description: "CoCalc Database Service (changefeeds)",
+    queue: "0",
   });
 
   api = service.addEndpoint("api", { subject });
+  global.x = { coordinator };
 
-  for await (const mesg of api) {
-    handleRequest({ mesg, nc, coordinator });
+  try {
+    for await (const mesg of api) {
+      handleRequest({ mesg, nc, coordinator });
+    }
+  } finally {
+    coordinator.close();
   }
 }
 
@@ -90,6 +94,7 @@ async function handleRequest({ mesg, nc, coordinator }) {
   try {
     const { account_id, project_id } = getUserId(mesg.subject);
     const { name, args } = jc.decode(mesg.data) ?? ({} as any);
+    //console.log(`got request: "${JSON.stringify({ name, args })}"`);
     // logger.debug(`got request: "${JSON.stringify({ name, args })}"`);
     if (!name) {
       throw Error("api endpoint name must be given in message");
@@ -99,23 +104,37 @@ async function handleRequest({ mesg, nc, coordinator }) {
     //       project_id,
     //       name,
     //     });
-    resp = await getResponse({ name, args, account_id, project_id, nc, coordinator });
+    resp = await getResponse({
+      name,
+      args,
+      account_id,
+      project_id,
+      nc,
+      coordinator,
+    });
   } catch (err) {
-    // logger.debug("ERROR", err);
+    logger.debug(`ERROR -- ${err}`);
     resp = { error: `${err}` };
   }
   // logger.debug(`Responding with "${JSON.stringify(resp)}"`);
   mesg.respond(jc.encode(resp));
 }
 
-async function getResponse({ name, args, account_id, project_id, nc }) {
+async function getResponse({
+  name,
+  args,
+  account_id,
+  project_id,
+  nc,
+  coordinator,
+}) {
   if (name == "userQuery") {
     const opts = { ...args[0], account_id, project_id };
     if (!opts.changes) {
       // a normal query
       return await userQuery(opts);
     } else {
-      return await createChangefeed(opts, nc);
+      return await createChangefeed(opts, nc, coordinator);
     }
   } else {
     throw Error(`name='${name}' not implemented`);
@@ -155,7 +174,7 @@ function cancelAllChangefeeds() {
 // async function, but then handle (and don't return)
 // the subsequent calls to cb generated by the changefeed.
 const createChangefeed = reuseInFlight(
-  async (opts, nc) => {
+  async (opts, nc, coordinator) => {
     const query = opts.query;
     // the query *AND* the user making it define the thing:
     const user = { account_id: opts.account_id, project_id: opts.project_id };
@@ -164,15 +183,28 @@ const createChangefeed = reuseInFlight(
       ...user,
     });
     const hash = sha1(desc);
-    const now = Date.now();
+    const manager = await coordinator.getManager(hash);
+    if (manager && coordinator.managerId != manager) {
+      // somebody else owns it and they are active -- express interest
+      await coordinator.userInterest(hash);
+      return;
+    }
+    // take it
+    await coordinator.takeManagement(hash);
+
     if (changefeedInterest[hash]) {
-      changefeedInterest[hash] = now;
+      changefeedInterest[hash] = now();
       logger.debug("using existing changefeed for", queryTable(query), user);
       return;
     }
     logger.debug("creating new changefeed for", queryTable(query), user);
     const changes = uuid();
     changefeedHashes[changes] = hash;
+    logger.debug(
+      "managing ",
+      Object.keys(changefeedHashes).length,
+      "changefeeds",
+    );
     const env = { nc, jc, sha1 };
     // If you change any settings below (i.e., atomic or immutable), you might also have to change them in
     //   src/packages/sync/table/changefeed-nats.ts
@@ -256,22 +288,26 @@ const createChangefeed = reuseInFlight(
     try {
       await callback(f);
       // it's running successfully
-      changefeedInterest[hash] = Date.now();
+      changefeedInterest[hash] = now();
 
       const watch = async () => {
         // it's all setup and running.  If there's no interest for a while, stop watching
         while (true) {
           await delay(CHANGEFEED_INTEREST_PERIOD_MS);
           if (
-            Date.now() - changefeedInterest[hash] >
+            now() - changefeedInterest[hash] >
             CHANGEFEED_INTEREST_PERIOD_MS
           ) {
-            logger.debug(
-              "insufficient interest in the changefeed, so we stop it.",
-              query,
-            );
-            cancelChangefeed(changes);
-            return;
+            // we check both the local known interest *AND* interest recorded by any other servers!
+            const last = await coordinator.lastUserInterest(hash);
+            if (now() - last >= CHANGEFEED_INTEREST_PERIOD_MS) {
+              logger.debug(
+                "insufficient interest in the changefeed, so we stop it.",
+                query,
+              );
+              cancelChangefeed(changes);
+              return;
+            }
           }
         }
       };

src/packages/database/nats/coordinator.ts

@@ -36,15 +36,18 @@ import { akv, type AKV } from "@cocalc/backend/nats/sync";
 import { randomId } from "@cocalc/nats/names";
 import getTime from "@cocalc/nats/time";
 
-const now = () => getTime({ noError: true });
+//const TIMEOUT = 15000;
+const TIMEOUT = 3000;
+
+export const now = () => getTime({ noError: true });
 
 export class Coordinator {
   public readonly managerId: string;
   public readonly akv: AKV;
   public readonly timeout: number;
   private interval;
 
-  constructor({ timeout = 30000 }: { timeout?: number } = {}) {
+  constructor({ timeout = TIMEOUT }: { timeout?: number } = {}) {
     this.managerId = randomId();
     this.akv = akv({ name: "changefeeds" });
     this.timeout = timeout;
@@ -64,8 +67,8 @@ export class Coordinator {
     await this.akv.set(this.managerId, now());
   };
 
-  owner = async (id: string): Promise<string | undefined> => {
-    const managerId = await this.akv.get(id);
+  getManager = async (id: string): Promise<string | undefined> => {
+    const { managerId } = (await this.akv.get(id)) ?? {};
     if (!managerId) {
       return undefined;
     }
@@ -81,24 +84,25 @@ export class Coordinator {
 
   // use expresses interest in changefeed with given id,
   // which we may or may not be the manager of.
-  interest = async (id: string) => {
+  userInterest = async (id: string) => {
     const x = await this.akv.get(id);
     if (!x) {
       return;
     }
     x.time = now();
-    await this.akv.set(x);
+    await this.akv.set(id, x);
+  };
+
+  lastUserInterest = async (id: string): Promise<number> => {
+    const { time } = (await this.akv.get(id)) ?? { time: 0 };
+    return time;
   };
 
-  take = async (id: string) => {
-    const cur = await this.owner(id);
+  takeManagement = async (id: string) => {
+    const cur = await this.getManager(id);
     if (cur && cur != this.managerId) {
       throw Error(`${id} is locked by another manager`);
     }
-    await this.akv.set(id, this.managerId);
-  };
-
-  free = async (id: string) => {
-    await this.akv.delete(id);
+    await this.akv.set(id, { time: now(), managerId: this.managerId });
   };
 }