openvibers/openvibe.events Commits

Events: iovalkey is a dependency, test:pg boots with Valkey, drain() waits for mid-pass completions.

OP OpenVibers committed 28 Sept 2026 · 7bcccebee0 · 5 files, +117 −2

The PostgreSQL release failed to start in production (openvibe-sdk/valkey needs iovalkey, and no test set
VALKEY_URL); ovhost put the SQLite release back. test:pg now boots every service with the Valkey container under a
prefix of its own, so the limits store and the readiness check run against it. drain() (tests) stopped early when
the last delivery in flight finished during a pass's query: that pass still counted it busy, so its successor stayed
due. It now passes again whenever a delivery settled during the pass (the running worker already did, via its
finally). npm test 17/17, test:pg 17/17 five runs in a row.

Changes

ddiff --git a/package-lock.json b/package-lock.json
iindex f4e1cda..eb133e5 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,6 +12,7 @@
          "better-sqlite3": "^11.10.0",
          "dotenv": "^16.4.7",
          "express": "^4.21.2",
++        "iovalkey": "^0.4.0",
          "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
          "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.20.4",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.28.0",
@@@ -31,6 +32,12 @@
        "devOptional": true,
        "license": "Apache-2.0"
      },
++    "node_modules/@iovalkey/commands": {
++      "version": "0.1.0",
++      "resolved": "https://registry.npmjs.org/@iovalkey/commands/-/commands-0.1.0.tgz",
++      "integrity": "sha512-/B9W4qKSSITDii5nkBCHyPkIkAi+ealUtr1oqBJsLxjSRLka4pxun2VvMNSmcwgAMxgXtQfl0qRv7TE+udPJzg==",
++      "license": "MIT"
++    },
      "node_modules/accepts": {
        "version": "1.3.8",
        "resolved": "https://registry.npmjs.org/accepts/-/accepts-1.3.8.tgz",
@@@ -232,6 +239,15 @@
        "integrity": "sha512-jJ0bqzaylmJtVnNgzTeSOs8DPavpbYgEr/b0YL8/2GO3xJEhInFmhKMUnEJQjZumK7KXGFhUy89PrsJWlakBVg==",
        "license": "ISC"
      },
++    "node_modules/cluster-key-slot": {
++      "version": "1.1.2",
++      "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz",
++      "integrity": "sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==",
++      "license": "Apache-2.0",
++      "engines": {
++        "node": ">=0.10.0"
++      }
++    },
      "node_modules/content-disposition": {
        "version": "0.5.4",
        "resolved": "https://registry.npmjs.org/content-disposition/-/content-disposition-0.5.4.tgz",
@@@ -301,6 +317,15 @@
          "node": ">=4.0.0"
        }
      },
++    "node_modules/denque": {
++      "version": "2.1.0",
++      "resolved": "https://registry.npmjs.org/denque/-/denque-2.1.0.tgz",
++      "integrity": "sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==",
++      "license": "Apache-2.0",
++      "engines": {
++        "node": ">=0.10"
++      }
++    },
      "node_modules/depd": {
        "version": "2.0.0",
        "resolved": "https://registry.npmjs.org/depd/-/depd-2.0.0.tgz",
@@@ -710,6 +735,49 @@
        "integrity": "sha512-JV/yugV2uzW5iMRSiZAyDtQd+nxtUnjeLt0acNdw98kKLrvuRVyB80tsREOE7yvGVgalhZ6RNXCmEHkUKBKxew==",
        "license": "ISC"
      },
++    "node_modules/iovalkey": {
++      "version": "0.4.0",
++      "resolved": "https://registry.npmjs.org/iovalkey/-/iovalkey-0.4.0.tgz",
++      "integrity": "sha512-OSUKxJ+s44CLdUTaicX4+pVrBN/zHKIYwr2oKhRDuczr19FwcSCXEZHzcMUYWZNh1mNSCV9bNJHW/T6cHVm9Aw==",
++      "license": "MIT",
++      "dependencies": {
++        "@iovalkey/commands": "^0.1.0",
++        "cluster-key-slot": "^1.1.0",
++        "debug": "^4.3.4",
++        "denque": "^2.1.0",
++        "lodash.defaults": "^4.2.0",
++        "lodash.isarguments": "^3.1.0",
++        "redis-errors": "^1.2.0",
++        "redis-parser": "^3.0.0",
++        "standard-as-callback": "^2.1.0"
++      },
++      "engines": {
++        "node": ">=18.12.0"
++      }
++    },
++    "node_modules/iovalkey/node_modules/debug": {
++      "version": "4.4.3",
++      "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz",
++      "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==",
++      "license": "MIT",
++      "dependencies": {
++        "ms": "^2.1.3"
++      },
++      "engines": {
++        "node": ">=6.0"
++      },
++      "peerDependenciesMeta": {
++        "supports-color": {
++          "optional": true
++        }
++      }
++    },
++    "node_modules/iovalkey/node_modules/ms": {
++      "version": "2.1.3",
++      "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz",
++      "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==",
++      "license": "MIT"
++    },
      "node_modules/ipaddr.js": {
        "version": "1.9.1",
        "resolved": "https://registry.npmjs.org/ipaddr.js/-/ipaddr.js-1.9.1.tgz",
@@@ -774,12 +842,24 @@
          "safe-buffer": "^5.0.1"
        }
      },
++    "node_modules/lodash.defaults": {
++      "version": "4.2.0",
++      "resolved": "https://registry.npmjs.org/lodash.defaults/-/lodash.defaults-4.2.0.tgz",
++      "integrity": "sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==",
++      "license": "MIT"
++    },
      "node_modules/lodash.includes": {
        "version": "4.3.0",
        "resolved": "https://registry.npmjs.org/lodash.includes/-/lodash.includes-4.3.0.tgz",
        "integrity": "sha512-W3Bx6mdkRTGtlJISOvVD/lbqjTlPPUDTMnlXZFnVwi9NKJ6tiAk6LVdlhZMm17VZisqhKcgzpO5Wz91PCt5b0w==",
        "license": "MIT"
      },
++    "node_modules/lodash.isarguments": {
++      "version": "3.1.0",
++      "resolved": "https://registry.npmjs.org/lodash.isarguments/-/lodash.isarguments-3.1.0.tgz",
++      "integrity": "sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==",
++      "license": "MIT"
++    },
      "node_modules/lodash.isboolean": {
        "version": "3.0.3",
        "resolved": "https://registry.npmjs.org/lodash.isboolean/-/lodash.isboolean-3.0.3.tgz",
@@@ -1298,6 +1378,27 @@
          "node": ">= 6"
        }
      },
++    "node_modules/redis-errors": {
++      "version": "1.2.0",
++      "resolved": "https://registry.npmjs.org/redis-errors/-/redis-errors-1.2.0.tgz",
++      "integrity": "sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==",
++      "license": "MIT",
++      "engines": {
++        "node": ">=4"
++      }
++    },
++    "node_modules/redis-parser": {
++      "version": "3.0.0",
++      "resolved": "https://registry.npmjs.org/redis-parser/-/redis-parser-3.0.0.tgz",
++      "integrity": "sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==",
++      "license": "MIT",
++      "dependencies": {
++        "redis-errors": "^1.0.0"
++      },
++      "engines": {
++        "node": ">=4"
++      }
++    },
      "node_modules/require-from-string": {
        "version": "2.0.2",
        "resolved": "https://registry.npmjs.org/require-from-string/-/require-from-string-2.0.2.tgz",
@@@ -1522,6 +1623,12 @@
          "node": ">= 10.x"
        }
      },
++    "node_modules/standard-as-callback": {
++      "version": "2.1.0",
++      "resolved": "https://registry.npmjs.org/standard-as-callback/-/standard-as-callback-2.1.0.tgz",
++      "integrity": "sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==",
++      "license": "MIT"
++    },
      "node_modules/statuses": {
        "version": "2.0.2",
        "resolved": "https://registry.npmjs.org/statuses/-/statuses-2.0.2.tgz",
ddiff --git a/package.json b/package.json
iindex 45a102c..f59ce4f 100644
---- a/package.json
++++ b/package.json
@@@ -24,6 +24,7 @@
      "better-sqlite3": "^11.10.0",
      "dotenv": "^16.4.7",
      "express": "^4.21.2",
++    "iovalkey": "^0.4.0",
      "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
      "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.20.4",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.28.0",
ddiff --git a/server/worker.js b/server/worker.js
iindex 9ffe7e2..c89052f 100644
---- a/server/worker.js
++++ b/server/worker.js
@@@ -124,6 +124,7 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
  
      // One pass at a time: the due query and start() must not interleave, or a second pass would read the rows the
      // first is starting and send them twice. A call during a pass gets one more pass right after it.
++    let finished = 0;   // deliveries settled so far (drain)
      let passing = null;
      let queued = null;
      function dispatch() {
@@@ -164,6 +165,7 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
          const p = send(d)
              .catch(err => log.error(`[worker] delivery ${d.event_id} -> ${d.subscription_id} crashed: ${err.stack || err}`))
              .finally(() => {
++                finished++;
                  busy.delete(d.subscription_id);
                  if (isApp) appInflight--;
                  inflight.delete(p);
@@@ -187,8 +189,10 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
      /** Deliver everything due now (and whatever becomes due as a result), then resolve. Tests. */
      async function drain() {
          for (;;) {
++            // A delivery that finished during the pass's query was still busy for it: its successor needs one more pass.
++            const before = finished;
              const started = await dispatch();
--            if (!started && !inflight.size) return;
++            if (!started && !inflight.size && finished === before) return;
              if (inflight.size) await Promise.race([...inflight]);
          }
      }
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 12a4391..0688a65 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -80,6 +80,9 @@ async function boot({ env = {}, clock = manualClock(), worker = 'manual', fetchI
          EVENTS_DB_PATH: path.join(dir, 'events.db'),
          OV_NETWORK_PUBLIC_KEY: publicKey,
          EVENTS_WORKER: worker === 'manual' ? 'off' : 'on',
++        // test:pg boots with the Valkey container too (limit counters shared across processes), under a prefix of its own.
++        ...(process.env.EVENTS_TEST_STORE === 'pg' && process.env.OV_TEST_VALKEY_URL
++            ? { VALKEY_URL: process.env.OV_TEST_VALKEY_URL, VALKEY_PREFIX: `ov:events-test:${process.pid}:${ids.newId('event').slice(-8)}:` } : {}),
          ...env,
      });
      // One database per boot (PGlite, or EVENTS_TEST_STORE=pg: the containers), dropped when the boot stops.
ddiff --git a/test/observability.test.js b/test/observability.test.js
iindex 4c61b42..e68b856 100644
---- a/test/observability.test.js
++++ b/test/observability.test.js
@@@ -26,7 +26,7 @@ t('/api/ready: every check reports status, latency and checked_at; worker is req
      assert.strictEqual(r.body.service, 'events');
      assert.deepStrictEqual(Object.keys(r.body.checks), ['db', 'valkey', 'network_jwks', 'delivery_worker', 'dlq']);
      for (const [name, c] of Object.entries(r.body.checks)) {
--        assert.strictEqual(c.status, name === 'valkey' ? 'skipped' : 'ok', name);   // test boots have no VALKEY_URL
++        assert.strictEqual(c.status, name === 'valkey' && !h.config.valkey.url ? 'skipped' : 'ok', name);   // Valkey under test:pg only
          assert.strictEqual(typeof c.latency_ms, 'number');
          assert.ok(Date.parse(c.checked_at));
      }