fix(database): Mongo live-update interventions (watch, token, tickets, recovery)

docker/docker-compose.standalone.yml

@@ -78,6 +78,8 @@ services:
       MONGO_INITDB_DATABASE: 'conduit'
       MONGO_INITDB_ROOT_USERNAME: '${DB_USER:-conduit}'
       MONGO_INITDB_ROOT_PASSWORD: '${DB_PASS:-pass}'
+    # Existing volumes created before rs0 will not elect a replica set.
+    # Remove the mongo volume (or start with an empty data dir) if hello stays standalone.
     entrypoint:
       - bash
       - -c

docker/docker-compose.yml

@@ -282,6 +282,8 @@ services:
       MONGO_INITDB_ROOT_USERNAME: '${DB_USER:-conduit}'
       MONGO_INITDB_ROOT_PASSWORD: '${DB_PASS:-pass}'
     profiles: ['mongodb']
+    # Existing volumes created before rs0 will not elect a replica set.
+    # Remove the mongo volume (or start with an empty data dir) if hello stays standalone.
     entrypoint:
       - bash
       - -c

libraries/grpc-sdk/package.json

@@ -25,7 +25,8 @@
     "prepublish": "npm run build",
     "prebuild": "npm run protoc",
     "build": "rimraf dist && tsup",
-    "protoc": "sh build.sh"
+    "protoc": "sh build.sh",
+    "test": "npx tsc -p tsconfig.test.json && node --test dist-test/utilities/EventBus.test.js"
   },
   "license": "MIT",
   "dependencies": {

libraries/grpc-sdk/src/utilities/EventBus.test.ts

@@ -0,0 +1,124 @@
+import assert from 'node:assert/strict';
+import { describe, it } from 'node:test';
+import { EventBus } from './EventBus.js';
+
+type Listener = (channel: string, message: string) => void;
+
+class FakeRedis {
+  handlers: Record<string, Listener[]> = {};
+  subscribed = new Set<string>();
+  failNext = new Set<string>();
+
+  on(event: string, listener: Listener) {
+    this.handlers[event] = this.handlers[event] ?? [];
+    this.handlers[event].push(listener);
+  }
+
+  subscribe(channel: string, cb?: (err?: Error | null) => void) {
+    if (this.failNext.has(channel)) {
+      this.failNext.delete(channel);
+      cb?.(new Error('subscribe failed'));
+      return;
+    }
+    this.subscribed.add(channel);
+    cb?.(null);
+  }
+
+  unsubscribe(channel: string, cb?: () => void) {
+    void channel;
+    cb?.();
+  }
+
+  publish(channel: string, message: string) {
+    void channel;
+    void message;
+  }
+
+  quit() {}
+
+  emitMessage(channel: string, message: string) {
+    for (const listener of this.handlers.message ?? []) {
+      listener(channel, message);
+    }
+  }
+}
+
+function createBus() {
+  const sub = new FakeRedis();
+  const pub = new FakeRedis();
+  const manager = {
+    getClient: () => sub,
+  };
+  const bus = new EventBus(manager as never);
+  (bus as unknown as { _clientSubscriber: FakeRedis })._clientSubscriber = sub;
+  (bus as unknown as { _clientPublisher: FakeRedis })._clientPublisher = pub;
+  return { bus, sub };
+}
+
+describe('EventBus', () => {
+  it('fires once after deactivate/reactivate on the same channel', () => {
+    const { bus, sub } = createBus();
+    let count = 0;
+    bus.subscribe(
+      'database:update:Order',
+      () => {
+        count += 1;
+      },
+      'relay-a',
+    );
+    bus.unsubscribe('relay-a');
+    bus.subscribe(
+      'database:update:Order',
+      () => {
+        count += 1;
+      },
+      'relay-a',
+    );
+    sub.emitMessage('database:update:Order', '{"ok":true}');
+    assert.equal(count, 1);
+  });
+
+  it('keeps the second subscriber when the first is removed', () => {
+    const { bus, sub } = createBus();
+    let first = 0;
+    let second = 0;
+    bus.subscribe(
+      'chan',
+      () => {
+        first += 1;
+      },
+      'one',
+    );
+    bus.subscribe(
+      'chan',
+      () => {
+        second += 1;
+      },
+      'two',
+    );
+    bus.unsubscribe('one');
+    sub.emitMessage('chan', 'x');
+    assert.equal(first, 0);
+    assert.equal(second, 1);
+  });
+
+  it('subscribeAck rejects when Redis subscribe fails', async () => {
+    const { bus, sub } = createBus();
+    sub.failNext.add('chan');
+    await assert.rejects(
+      () => bus.subscribeAck('chan', () => {}, 'relay-a'),
+      /subscribe failed/,
+    );
+    sub.failNext.delete('chan');
+    let count = 0;
+    await bus.subscribeAck(
+      'chan',
+      () => {
+        count += 1;
+      },
+      'relay-a',
+    );
+    sub.emitMessage('chan', 'x');
+    assert.equal(count, 1);
+  });
+});

libraries/grpc-sdk/src/utilities/EventBus.ts

@@ -3,35 +3,58 @@ import { Cluster, Redis } from 'ioredis';
 import crypto from 'crypto';
 import { getLogger } from './GrpcSdkContext.js';
 
+type ChannelCallbacks = Map<string, (message: string) => void>;
+
 export class EventBus {
   private _clientSubscriber: Redis | Cluster;
   private _clientPublisher: Redis | Cluster;
-  private _subscribedChannels: { [listener: string]: ((message: string) => void)[] };
-  private _subscribers: { [listener: string]: [string, number] };
+  /** channelName -> subscriberId -> callback */
+  private _channelCallbacks: Map<string, ChannelCallbacks>;
+  /** subscriberId -> channelName */
+  private _subscriberChannels: Map<string, string>;
+  /** channels with a successful Redis SUBSCRIBE */
+  private _redisSubscribedChannels: Set<string>;
+  private _subscribeInFlight = new Map<string, Promise<void>>();
   private _signature: string;
+  private _anonymousSubscriberSeq = 0;
+  private _shuttingDown = false;
 
   constructor(redisManager: RedisManager) {
-    this._subscribedChannels = {};
-    this._subscribers = {};
+    this._channelCallbacks = new Map();
+    this._subscriberChannels = new Map();
+    this._redisSubscribedChannels = new Set();
     this._clientSubscriber = redisManager.getClient({ keyPrefix: 'bus_' });
     this._clientPublisher = redisManager.getClient({ keyPrefix: 'bus_' });
     this._signature = crypto.randomBytes(20).toString('hex');
     this._clientSubscriber.on('ready', () => {
       getLogger().log('The Bus is in the station...hehe');
     });
+    this._clientSubscriber.on('message', (channel: string, message: string) => {
+      this.dispatch(channel, message);
+    });
     process.on('exit', () => {
-      this._clientSubscriber.quit();
-      this._clientPublisher.quit();
+      this.quit();
     });
   }
 
+  quit(): void {
+    if (this._shuttingDown) return;
+    this._shuttingDown = true;
+    void this._clientSubscriber.quit();
+    void this._clientPublisher.quit();
+  }
+
   unsubscribe(subscriberId: string): void {
-    if (this._subscribers[subscriberId]) {
-      const [channelName, index] = this._subscribers[subscriberId];
-      this._subscribedChannels[channelName].splice(index, 1);
-      delete this._subscribers[subscriberId];
-      if (this._subscribedChannels[channelName].length === 0) {
-        delete this._subscribedChannels[channelName];
+    const channelName = this._subscriberChannels.get(subscriberId);
+    if (!channelName) {
+      return;
+    }
+    const callbacks = this._channelCallbacks.get(channelName);
+    callbacks?.delete(subscriberId);
+    this._subscriberChannels.delete(subscriberId);
+    if (callbacks && callbacks.size === 0) {
+      this._channelCallbacks.delete(channelName);
+      if (this._redisSubscribedChannels.delete(channelName)) {
         this._clientSubscriber.unsubscribe(channelName, () => {});
       }
     }
@@ -42,43 +65,100 @@ export class EventBus {
     callback: (message: string) => void,
     subscriberId?: string,
   ): void {
+    void this.subscribeAck(channelName, callback, subscriberId).catch(err => {
+      getLogger().error(
+        `EventBus subscribe failed for ${channelName}: ${
+          err instanceof Error ? err.message : String(err)
+        }`,
+      );
+    });
+  }
+
+  async subscribeAck(
+    channelName: string,
+    callback: (message: string) => void,
+    subscriberId?: string,
+  ): Promise<void> {
+    if (this._shuttingDown) {
+      return;
+    }
+    const id =
+      subscriberId ??
+      `anon:${channelName}:${++this._anonymousSubscriberSeq}:${crypto.randomBytes(4).toString('hex')}`;
     if (subscriberId) {
-      // if subscriberId is provided, and it is already subscribed, unsubscribe it first
       this.unsubscribe(subscriberId);
     }
-    if (this._subscribedChannels[channelName]) {
-      this._subscribedChannels[channelName].push(callback);
-      if (subscriberId) {
-        this._subscribers[subscriberId] = [
-          channelName,
-          this._subscribedChannels[channelName].length - 1,
-        ];
-      }
+
+    let callbacks = this._channelCallbacks.get(channelName);
+    if (!callbacks) {
+      callbacks = new Map();
+      this._channelCallbacks.set(channelName, callbacks);
+    }
+    callbacks.set(id, callback);
+    this._subscriberChannels.set(id, channelName);
+
+    if (this._redisSubscribedChannels.has(channelName)) {
       return;
     }
-    this._subscribedChannels[channelName] = [callback];
-    this._clientSubscriber.subscribe(channelName, () => {});
-    const self = this;
-    this._clientSubscriber.on('message', (channel: string, message: string) => {
-      if (channel !== channelName) return;
-      // if the message supports the signature
-      if (message.indexOf('CND_Signature') !== -1) {
-        // if the message does not contain this module's signature
-        if (message.indexOf(self._signature) === -1) {
-          self._subscribedChannels[channelName].forEach(fn => {
-            fn(message.split('CND_Signature:')[0]);
-          });
-        }
-      } else {
-        self._subscribedChannels[channelName].forEach(fn => {
-          fn(message);
+
+    let inFlight = this._subscribeInFlight.get(channelName);
+    if (!inFlight) {
+      inFlight = new Promise<void>((resolve, reject) => {
+        this._clientSubscriber.subscribe(channelName, err => {
+          if (err) {
+            reject(err);
+            return;
+          }
+          this._redisSubscribedChannels.add(channelName);
+          resolve();
         });
+      }).finally(() => {
+        this._subscribeInFlight.delete(channelName);
+      });
+      this._subscribeInFlight.set(channelName, inFlight);
+    }
+
+    try {
+      await inFlight;
+    } catch (err) {
+      if (!this._redisSubscribedChannels.has(channelName)) {
+        this.removeChannelCallbacks(channelName);
       }
-    });
+      this.unsubscribe(id);
+      throw err;
+    }
   }
 
   publish(channelName: string, message: string) {
     message = message + `CND_Signature:${this._signature}`;
     this._clientPublisher.publish(channelName, message);
   }
+
+  private removeChannelCallbacks(channelName: string): void {
+    const callbacks = this._channelCallbacks.get(channelName);
+    if (!callbacks) {
+      return;
+    }
+    for (const subId of callbacks.keys()) {
+      this._subscriberChannels.delete(subId);
+    }
+    this._channelCallbacks.delete(channelName);
+  }
+
+  private dispatch(channel: string, message: string): void {
+    const callbacks = this._channelCallbacks.get(channel);
+    if (!callbacks || callbacks.size === 0) {
+      return;
+    }
+    let payload = message;
+    if (message.indexOf('CND_Signature') !== -1) {
+      if (message.indexOf(this._signature) !== -1) {
+        return;
+      }
+      payload = message.split('CND_Signature:')[0];
+    }
+    for (const fn of callbacks.values()) {
+      fn(payload);
+    }
+  }
 }

libraries/grpc-sdk/tsconfig.test.json

@@ -0,0 +1,11 @@
+{
+  "extends": "./tsconfig.json",
+  "compilerOptions": {
+    "outDir": "./dist-test",
+    "rootDir": "./src",
+    "declaration": false,
+    "sourceMap": false,
+    "types": ["node"]
+  },
+  "include": ["src/utilities/EventBus.ts", "src/utilities/EventBus.test.ts"]
+}

libraries/hermes/package.json

@@ -11,7 +11,7 @@
   "scripts": {
     "prepublish": "npm run build",
     "build": "rimraf dist && tsc",
-    "test": "tsc -p tsconfig.test.json && node --test dist-test/Socket/applySocketGlobalMiddlewares.test.js",
+    "test": "npx tsc -p tsconfig.test.json && node --test dist-test/Socket/*.test.js",
     "publish": "npm publish",
     "postbuild": "copyfiles -u 1 src/*.proto src/**/*.json ./dist/"
   },