Fix unbounded session table growth in directory

acs-directory/lib/mqttcli.js

@@ -321,6 +321,12 @@ export default class MQTTCli {
 
         if (notify.length)
             this.publish_changed(notify);
+
+        /* Clean up old sessions now we've finished the change-notify.
+         * Only do this for the current session; the old session's
+         * notify will also fire but we skip it here. */
+        if (session.next_for_device == null)
+            await this.model.cleanup_old_sessions(id);
     }
 
     async on_service_notify(id) {
@@ -446,7 +452,7 @@ export default class MQTTCli {
 
     async on_birth(address, payload) {
         this.log("device", `Registering BIRTH for ${address}`);
-        this.online.add(address);
+        this.online.add(address.toString());
 
         let tree;
         if (payload.uuid === UUIDs.FactoryPlus) {
@@ -479,7 +485,7 @@ export default class MQTTCli {
         this.log("device", `Registering DEATH for ${address}`);
 
         this.alerts.delete(address);
-        this.online.delete(address);
+        this.online.delete(address.toString());
         await this.model.death({address, time});
 
         this.log("device", `Finished DEATH for ${address}`);
@@ -546,7 +552,7 @@ export default class MQTTCli {
          * rebirth any device we haven't seen a birth for, even if the
          * database says it's online. This is important because we might
          * have the wrong schema information. */
-        if (this.online.has(addr) || pending[addr]) return false;
+        if (this.online.has(addr.toString()) || pending[addr]) return false;
 
         /* Mark that we're working on this device and wait 5-10s to see if
          * it rebirths on its own. */
@@ -555,7 +561,7 @@ export default class MQTTCli {
 
         /* Clear our marker first so we retry next time */
         delete (pending[addr]);
-        if (this.online.has(addr)) return false;
+        if (this.online.has(addr.toString())) return false;
 
         sent[addr] = Date.now();
         return true;

acs-directory/lib/queries.js

@@ -183,6 +183,14 @@ export default class Queries {
         return dbr.rows.map(r => r.uuid);
     }
 
+    async cleanup_old_sessions(session_id) {
+        await this.query(`
+            delete from session
+            where device = (select device from session where id = $1)
+              and next_for_device is not null
+        `, [session_id]);
+    }
+
     async record_schema(session, schema) {
         const schid = await this.find_or_create("schema", schema);
         if (schid == null) return;

acs-directory/package.json

@@ -16,16 +16,21 @@
   "type": "module",
   "scripts": {
     "mqtt": "node bin/directory-mqtt",
-    "webapi": "node bin/directory-webapi"
+    "webapi": "node bin/directory-webapi",
+    "test": "vitest run"
   },
   "dependencies": {
+    "@amrc-factoryplus/pg-client": "file:../lib/js-pg-client",
     "@amrc-factoryplus/rx-client": "file:../lib/js-rx-client",
     "@amrc-factoryplus/rx-util": "file:../lib/js-rx-util",
-    "@amrc-factoryplus/service-client": "file:../lib/js-service-client",
     "@amrc-factoryplus/service-api": "file:../lib/js-service-api",
-    "@amrc-factoryplus/pg-client": "file:../lib/js-pg-client",
+    "@amrc-factoryplus/service-client": "file:../lib/js-service-client",
     "async": "^3.2.4",
     "express": "^4.17.1",
-    "express-openapi-validator": "^4.13.2"
+    "express-openapi-validator": "^4.13.2",
+    "long": "^5.3.2"
+  },
+  "devDependencies": {
+    "vitest": "^3.0.4"
   }
 }

acs-directory/test/find-schemas.test.js

@@ -0,0 +1,59 @@
+import { describe, it, expect } from "vitest";
+import { MetricBranch } from "@amrc-factoryplus/service-client";
+import MQTTCli from "../lib/mqttcli.js";
+
+function make_cli() {
+    return Object.create(MQTTCli.prototype);
+}
+
+describe("find_schemas", () => {
+    it("returns empty set for empty tree", () => {
+        const cli = make_cli();
+        const result = cli.find_schemas({});
+        expect(result.size).toBe(0);
+    });
+
+    it("finds top-level Schema_UUID", () => {
+        const cli = make_cli();
+        const tree = {
+            Schema_UUID: { value: "schema-1" },
+        };
+        const result = cli.find_schemas(tree);
+        expect([...result]).toEqual(["schema-1"]);
+    });
+
+    it("finds nested Schema_UUIDs in MetricBranch", () => {
+        const cli = make_cli();
+        const branch = new MetricBranch();
+        branch.Schema_UUID = { value: "schema-nested" };
+        const tree = { sub: branch };
+        const result = cli.find_schemas(tree);
+        expect(result.has("schema-nested")).toBe(true);
+    });
+
+    it("finds multiple schemas at different levels", () => {
+        const cli = make_cli();
+        const branch = new MetricBranch();
+        branch.Schema_UUID = { value: "schema-2" };
+        const tree = {
+            Schema_UUID: { value: "schema-1" },
+            sub: branch,
+        };
+        const result = cli.find_schemas(tree);
+        expect(result.size).toBe(2);
+        expect(result.has("schema-1")).toBe(true);
+        expect(result.has("schema-2")).toBe(true);
+    });
+
+    it("deduplicates repeated schema UUIDs", () => {
+        const cli = make_cli();
+        const branch = new MetricBranch();
+        branch.Schema_UUID = { value: "schema-1" };
+        const tree = {
+            Schema_UUID: { value: "schema-1" },
+            sub: branch,
+        };
+        const result = cli.find_schemas(tree);
+        expect(result.size).toBe(1);
+    });
+});

acs-directory/test/session-cleanup.test.js

@@ -0,0 +1,38 @@
+import { describe, it, expect, vi } from "vitest";
+import Queries from "../lib/queries.js";
+
+describe("cleanup_old_sessions", () => {
+    it("issues a delete for non-current sessions by session id", async () => {
+        const mockQuery = vi.fn().mockResolvedValue({ rows: [], rowCount: 0 });
+        const q = new Queries(mockQuery);
+
+        await q.cleanup_old_sessions(42);
+
+        expect(mockQuery).toHaveBeenCalledOnce();
+        const [sql, params] = mockQuery.mock.calls[0];
+        expect(sql).toMatch(/delete from session/i);
+        expect(sql).toMatch(/next_for_device is not null/i);
+        expect(params).toEqual([42]);
+    });
+});
+
+describe("session_notification_info", () => {
+    it("returns session info with prev_for_device", async () => {
+        const row = {
+            device: "uuid-1",
+            group_id: "G",
+            node_id: "N",
+            device_id: "D",
+            next_for_device: null,
+            next_for_address: null,
+            prev_for_device: 10,
+        };
+        const mockQuery = vi.fn().mockResolvedValue({ rows: [row] });
+        const q = new Queries(mockQuery);
+
+        const result = await q.session_notification_info(42);
+
+        expect(result).toEqual(row);
+        expect(result.prev_for_device).toBe(10);
+    });
+});

acs-directory/vitest.config.js

@@ -0,0 +1,7 @@
+import { defineConfig } from "vitest/config";
+
+export default defineConfig({
+    test: {
+        include: ["test/**/*.test.js"],
+    },
+});

lib/js-service-client/lib/sparkplug/util.js

@@ -67,11 +67,10 @@ export class Address {
      * @arg device Device ID
      */
     constructor (group, node, device) {
-        this.group = group;
-        this.node = node;
-
         if (device == null || device == "")
             device = undefined;
+        this.group = group;
+        this.node = node;
         this.device = device;
     }
 

lib/js-service-client/package.json

@@ -5,7 +5,7 @@
   "main": "lib/index.js",
   "type": "module",
   "scripts": {
-    "test": "echo \"Error: no test specified\" && exit 1"
+    "test": "vitest run"
   },
   "keywords": [],
   "author": "",
@@ -26,6 +26,7 @@
     "@eslint/eslintrc": "^3.1.0",
     "@eslint/js": "^9.13.0",
     "eslint": "^9.13.0",
-    "globals": "^15.11.0"
+    "globals": "^15.11.0",
+    "vitest": "^3.0.4"
   }
 }

lib/js-service-client/test/address.test.js

@@ -0,0 +1,50 @@
+import { describe, it, expect } from "vitest";
+import { Address } from "../lib/sparkplug/util.js";
+
+describe("Address", () => {
+    it("treats empty string device as node address", () => {
+        const a = new Address("G", "N", "");
+        expect(a.device).toBeUndefined();
+    });
+
+    it("treats null device as node address", () => {
+        const a = new Address("G", "N", null);
+        expect(a.device).toBeUndefined();
+    });
+
+    it("toString formats node address", () => {
+        const a = new Address("G", "N");
+        expect(a.toString()).toBe("G/N");
+    });
+
+    it("toString formats device address", () => {
+        const a = new Address("G", "N", "D");
+        expect(a.toString()).toBe("G/N/D");
+    });
+
+    it("equals compares by value", () => {
+        const a = new Address("G", "N", "D");
+        const b = new Address("G", "N", "D");
+        expect(a.equals(b)).toBe(true);
+    });
+
+    it("equals returns false for different addresses", () => {
+        const a = new Address("G", "N", "D1");
+        const b = new Address("G", "N", "D2");
+        expect(a.equals(b)).toBe(false);
+    });
+
+    it("parse round-trips with toString", () => {
+        const a = new Address("G", "N", "D");
+        const b = Address.parse(a.toString());
+        expect(b.equals(a)).toBe(true);
+    });
+
+    it("parent_node returns node address", () => {
+        const a = new Address("G", "N", "D");
+        const parent = a.parent_node();
+        expect(parent.group).toBe("G");
+        expect(parent.node).toBe("N");
+        expect(parent.device).toBeUndefined();
+    });
+});

lib/js-service-client/vitest.config.js

@@ -0,0 +1,7 @@
+import { defineConfig } from "vitest/config";
+
+export default defineConfig({
+    test: {
+        include: ["test/**/*.test.js"],
+    },
+});