openvibers/openvibe.sdk Commits

Add startTelemetryRetention: prune soon after start, then daily (#44)

OP OpenVibers committed 30 days ago · 5cce8f2b76 · 12 files, +83 −13

Asked: Follow-up of https://github.com/OpenVibers/OpenVibe.Blog/pull/42 (Store request telemetry in Postgres with daily retention): its review asked for a change in OpenVibe.SDK that the PR cannot make from OpenVibe.Blog. The review said: VERDICT: FIX
Harness job sdk-follow-up-of-https-github-com-openvi-020036.

Changes

ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex 3b55b47..b39da0c 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -3,6 +3,14 @@
  All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minor
  release may change an API and says so here.
  
++## 0.46.0 (2026-10-13)
++
++- `startTelemetryRetention({ db, table?, olderThanDays = 30, everyMs = 24 h, delayMs = 60 s, log? })` → `{ run(), stop() }`:
++  the scheduled form of `pruneTelemetry()`. It first prunes `delayMs` after the call, then every `everyMs`, on unref'd
++  timers. A service that waited a full day for the first prune never pruned when it was redeployed more often than that
++  (found in review of OpenVibe.Blog#42). A failing prune is logged and retried at the next tick; runs never overlap;
++  `stop()` cancels the timers and waits for a run in flight.
++
  ## 0.45.0 (2026-10-13)
  
  The telemetry sink every service stores into (plan T1, step 10). Step 5 left `createTelemetry({ sink })` and
ddiff --git a/README.md b/README.md
iindex 8327f90..0efaa4b 100644
---- a/README.md
++++ b/README.md
@@@ -187,7 +187,7 @@ Until OpenVibe.Tools serves the gateway facade (Tools S6), point the jobs client
  | `openvibe-sdk/vip` | server | `createVipClient({baseUrl?, tokenClient \| getToken, fetch?, timeoutMs?})`: `evaluate({subject, resource, owner, fallback, mode})`, `checkEntitlement({subject, creator, product, mode})`, `isMember(subject, creator)`; every failure is a denial. `createVipCache({vip, ttlMs, denyTtlMs, unavailableTtlMs})`: `entitlement`, `peekEntitlement`, `evaluate`, `invalidate`, `handleEvent` (`vip.membership.changed` and Billing's entitlement events drop a member's answers at once), `clear`, `bounds` |
  | `openvibe-sdk/usage` | server | `usageSample(fields)` builds a `platform.usage-sample@1` record (required `id`, `idempotency_key`, `service`, `operation`, `quantity`, `unit`; `at` defaults to now ISO, `source` to `openvibe-sdk/usage`; null fields stripped; money fields never invented); `usageKey(service, ...parts)` joins an idempotency key with `:`; `validateUsageSample(record)` → `{ok, errors}` (openvibe-contracts, required lazily). `createUsageReporter({ db, service, source, table, billingUrl, tokenClient, audience, requireProject, ... })` owns one service's readings: `key(...parts)`/`sample(fields)`, and `record(t, reading)` queues the reading idempotency-keyed in the outbox inside the caller's transaction, relayed to Billing's `billing.usage.record` with backoff (a retry is never dropped; only Billing refusing the reading itself — any 4xx but 401/403/404/408/425/429, which stay pending — marks a row rejected; a 401 drops the cached token). The reporter bills whatever reading it is given: never record first-party or sandbox traffic (pass `requireProject: true` to refuse a reading without a `prj_…` project). `openvibe-sdk/govern` builds its `onUsage` record with it |
  | `openvibe-sdk/commerce` | server | `createCommerceClient(config, {caps, fetchImpl?, tokenClients?, timeoutMs?})`: the Billing client Tips and VIP share. `config` = `{billing: {url, audience, timeoutMs = 10000}, network: {internalUrl}, oauth: {clientId, clientSecret}}`; `caps` maps `intent`, `transfer`, `subscription`, `entitlement` (and optional `rates`, default `entitlement`) to capability strings, one lazily created `serviceAuth` token client per distinct string (`scope` = the string) unless `tokenClients` supplies one, keyed by logical key or capability. Methods: `createIntent` (`kind` purchase with `bits`, or subscription with `creator`/`autoRenew`), `createTransfer`, `refundTransfer`, `subscribeWithCredit`, `cancelSubscription`, `getSubscription`, `listSubscriptions`, `entitlement`, `rates`, `baseUrl`. `Idempotency-Key` is sent verbatim; a 401 invalidates the token and retries once with the same key; problem+json becomes `CommerceError` (`status`, `code`, `body`, `retryable`). `intentKey(prefix, parts)` joins the parts that are set; `receipt(kind, fields)` → `{kind, at, ...fields}` (null stripped, never posted). `openvibe-contracts` is required lazily |
--| `openvibe-sdk/telemetry` | server | `telemetrySample(fields)` / `validateTelemetrySample(record)` for `platform.telemetry-sample@1`; `createTelemetry({service, instance?, sink, intervalMs = 15000, now?, log?, maxBuffered = 1000})` → `{record(name, value, labels), gauge(name, value, labels), count(name, labels), flush(), stop()}`: buffered samples flushed to `await sink(samples)` on an unref'd interval and on `stop()` (a `gracefulStop` stop step); `record` builds `{service, instance, operation, at, latency_ms, ...labels}` (a non-number value goes to `extra[name]`); a rejecting sink is logged once and its samples kept; past `maxBuffered` the oldest sample is dropped, with a warning at most once a minute; the per-service PostgreSQL sink: `telemetrySchema(table = 'telemetry_samples')` (migration DDL), `createPgTelemetrySink({db, table?, log?})` (one multi-row INSERT per flush, malformed samples dropped with a warning), `pruneTelemetry({db, table?, olderThanDays = 30})` |
++| `openvibe-sdk/telemetry` | server | `telemetrySample(fields)` / `validateTelemetrySample(record)` for `platform.telemetry-sample@1`; `createTelemetry({service, instance?, sink, intervalMs = 15000, now?, log?, maxBuffered = 1000})` → `{record(name, value, labels), gauge(name, value, labels), count(name, labels), flush(), stop()}`: buffered samples flushed to `await sink(samples)` on an unref'd interval and on `stop()` (a `gracefulStop` stop step); `record` builds `{service, instance, operation, at, latency_ms, ...labels}` (a non-number value goes to `extra[name]`); a rejecting sink is logged once and its samples kept; past `maxBuffered` the oldest sample is dropped, with a warning at most once a minute; the per-service PostgreSQL sink: `telemetrySchema(table = 'telemetry_samples')` (migration DDL), `createPgTelemetrySink({db, table?, log?})` (one multi-row INSERT per flush, malformed samples dropped with a warning), `pruneTelemetry({db, table?, olderThanDays = 30})`, `startTelemetryRetention({db, table?, olderThanDays = 30, everyMs = 24 h, delayMs = 60 s, log?})` → `{run(), stop()}` (prunes soon after start and then daily on unref'd timers, so a service redeployed more than daily still prunes) |
  | `openvibe-sdk/cost` | server | Build and validate `platform.rate-card@1` and `platform.cost-snapshot@1` records with `rateCard()`, `costSnapshot()`, `validateRateCard()`, and `validateCostSnapshot()`; `estimate()` calculates a charge from a rate card |
  | `openvibe-sdk/runtime` | server | Build and validate `platform.runtime-offer@1` records with `runtimeOffer()` and `validateRuntimeOffer()`; `runtimeClass()`, `RUNTIME_CLASSES`, and `validateRuntimeClass()` handle known runtime classes |
  | `openvibe-sdk/storage` | server | `storageOffer()` and `validateStorageOffer()` for `platform.storage-offer@1` |
ddiff --git a/STATUS.json b/STATUS.json
iindex 3988d77..43f205a 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -3,7 +3,7 @@
    "stage": "alpha",
    "code": true,
    "wave": "W2",
--  "version": "0.45.0",
++  "version": "0.46.0",
    "plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30",
    "deployed": false,
    "deployment": "none: a package released as git tags; services and apps pin a tag",
ddiff --git a/browser/openvibe-sdk.mjs b/browser/openvibe-sdk.mjs
iindex 1e7ee36..5f56634 100644
---- a/browser/openvibe-sdk.mjs
++++ b/browser/openvibe-sdk.mjs
@@@ -1,4 +1,4 @@
--// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.45.0, MIT). Do not edit.
++// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.46.0, MIT). Do not edit.
  // One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces
  // (auth, registry, modules, realtime, media, community, bot, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity.
  /* eslint-disable */
@@@ -42,7 +42,7 @@ const { paginate, offsetPager } = __load(7);
  const semver = __load(6);
  
  /** The package version (test/version.test.js keeps it equal to package.json). */
--const SDK_VERSION = '0.45.0';
++const SDK_VERSION = '0.46.0';
  
  module.exports = {
      createClient,
ddiff --git a/docs/service.md b/docs/service.md
iindex 80922e2..056741d 100644
---- a/docs/service.md
++++ b/docs/service.md
@@@ -212,12 +212,13 @@ time). `registerSignals({ start, stop, lag })` injects a service's own monitor i
    a consumer that needs the fleet view reads the tables (or a later relay copies them) without changing any emitter.
  
  ```js
--const { telemetrySchema, createPgTelemetrySink, pruneTelemetry } = require('openvibe-sdk/telemetry');
++const { telemetrySchema, createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
  // migrations/NNN_telemetry.sql: the text of telemetrySchema() (21 columns in platform.telemetry-sample@1 order,
  // no `instance`, `extra` jsonb, indexes on (at) and (service, operation, at)); the default table is telemetry_samples.
  const collector = svc.createHttpTelemetry({ service: 'blog', sink: createPgTelemetrySink({ db, log }) });
  svc.gracefulStop({ name: 'Blog', server, stop: [() => collector.stop()] });   // the last flush is one INSERT
--await pruneTelemetry({ db });   // a daily job: DELETE … WHERE at < now() - 30 days (olderThanDays to change it)
++const retention = startTelemetryRetention({ db, log });   // DELETE … WHERE at < now() - 30 days: 60 s after start, then daily
++// add `() => retention.stop()` to the stop steps; pruneTelemetry({ db }) is the one-off delete it calls
  ```
  
  The sink does one multi-row parameterized INSERT per flush (a second only past 3120 rows, PostgreSQL's 65535
ddiff --git a/esm/telemetry.mjs b/esm/telemetry.mjs
iindex 03d764c..e37caa5 100644
---- a/esm/telemetry.mjs
++++ b/esm/telemetry.mjs
@@@ -3,5 +3,6 @@ import cjs from '../src/telemetry.js';
  
  export const {
      telemetrySample, validateTelemetrySample, createTelemetry, telemetrySchema, createPgTelemetrySink, pruneTelemetry,
++    startTelemetryRetention,
  } = cjs;
  export default cjs;
ddiff --git a/package-lock.json b/package-lock.json
iindex f836620..ed6c4f1 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -1,12 +1,12 @@
  {
    "name": "openvibe-sdk",
--  "version": "0.45.0",
++  "version": "0.46.0",
    "lockfileVersion": 3,
    "requires": true,
    "packages": {
      "": {
        "name": "openvibe-sdk",
--      "version": "0.45.0",
++      "version": "0.46.0",
        "license": "MIT",
        "devDependencies": {
          "@electric-sql/pglite": "^0.5.8",
ddiff --git a/package.json b/package.json
iindex ea17039..0eb50cc 100644
---- a/package.json
++++ b/package.json
@@@ -1,6 +1,6 @@
  {
    "name": "openvibe-sdk",
--  "version": "0.45.0",
++  "version": "0.46.0",
    "description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, Bot robots, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.",
    "main": "index.js",
    "types": "types/index.d.ts",
ddiff --git a/src/core/index.js b/src/core/index.js
iindex 1f8fdff..679eb8b 100644
---- a/src/core/index.js
++++ b/src/core/index.js
@@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');
  const semver = require('./semver');
  
  /** The package version (test/version.test.js keeps it equal to package.json). */
--const SDK_VERSION = '0.45.0';
++const SDK_VERSION = '0.46.0';
  
  module.exports = {
      createClient,
ddiff --git a/src/telemetry.js b/src/telemetry.js
iindex 112b123..1439d07 100644
---- a/src/telemetry.js
++++ b/src/telemetry.js
@@@ -12,7 +12,7 @@
   *   stop: [() => t.stop()],                                          // a gracefulStop stop step: one flush
   *
   * Where samples land: each service's own PostgreSQL table (telemetrySchema() in a migration, createPgTelemetrySink()
-- * as the sink, pruneTelemetry() for retention), not a central ingest; docs/service.md says why.
++ * as the sink, pruneTelemetry() for retention, startTelemetryRetention() to schedule it), not a central ingest; docs/service.md says why.
   */
  
  /** The fields platform.telemetry-sample@1 requires or allows, in the schema's order; null/undefined are stripped. */
@@@ -236,4 +236,34 @@ async function pruneTelemetry({ db, table = 'telemetry_samples', olderThanDays =
      return r.rowCount;
  }
  
--module.exports = { telemetrySample, validateTelemetrySample, createTelemetry, telemetrySchema, createPgTelemetrySink, pruneTelemetry };
++/** Scheduled retention: runs pruneTelemetry() `delayMs` after the call (default 60 s) and then every `everyMs`
++ *  (default 24 h), both on unref'd timers. The first run is soon, not a day out, so a service that is redeployed
++ *  more often than once a day still prunes. A failing prune is logged and tried again at the next tick; runs never
++ *  overlap. `run()` prunes now; `stop()` cancels the timers. */
++function startTelemetryRetention({ db, table = 'telemetry_samples', olderThanDays = 30, everyMs = 86400e3, delayMs = 60e3, log = console } = {}) {
++    // pruneTelemetry validates db, table and olderThanDays; fail here, not inside a timer
++    if (!db || typeof db.query !== 'function') throw new TypeError('startTelemetryRetention: an openvibe-sdk/db handle is required');
++    checkTable(table, 'startTelemetryRetention');
++    if (!(Number.isFinite(olderThanDays) && olderThanDays > 0)) throw new TypeError('startTelemetryRetention: olderThanDays must be a positive number');
++    if (!(Number.isFinite(everyMs) && everyMs > 0)) throw new TypeError('startTelemetryRetention: everyMs must be a positive number');
++    if (!(Number.isFinite(delayMs) && delayMs >= 0)) throw new TypeError('startTelemetryRetention: delayMs must be a non-negative number');
++    let running = null;
++    let stopped = false;
++    const run = () => {
++        if (running) return running;
++        running = pruneTelemetry({ db, table, olderThanDays })
++            .catch((err) => { log.error(`[telemetry] ${table}: retention prune failed: ${err && err.message}`); return 0; })
++            .finally(() => { running = null; });
++        return running;
++    };
++    const first = setTimeout(() => { if (!stopped) run(); }, delayMs);
++    const every = setInterval(() => { if (!stopped) run(); }, everyMs);
++    if (first.unref) first.unref();
++    if (every.unref) every.unref();
++    return {
++        run,
++        async stop() { stopped = true; clearTimeout(first); clearInterval(every); if (running) await running; },
++    };
++}
++
++module.exports = { telemetrySample, validateTelemetrySample, createTelemetry, telemetrySchema, createPgTelemetrySink, pruneTelemetry, startTelemetryRetention };
ddiff --git a/test/telemetry-sink.test.js b/test/telemetry-sink.test.js
iindex 255ba6b..6e778dd 100644
---- a/test/telemetry-sink.test.js
++++ b/test/telemetry-sink.test.js
@@@ -7,7 +7,7 @@
  const assert = require('node:assert/strict');
  const { run } = require('./helpers');
  const { createDb } = require('../src/db');
--const { telemetrySample, createTelemetry, telemetrySchema, createPgTelemetrySink, pruneTelemetry } = require('../src/telemetry');
++const { telemetrySample, createTelemetry, telemetrySchema, createPgTelemetrySink, pruneTelemetry, startTelemetryRetention } = require('../src/telemetry');
  
  const COLUMNS = ['id', 'at', 'service', 'project', 'subject', 'resource', 'provider', 'node', 'cell', 'region', 'operation',
      'latency_ms', 'queue_delay_ms', 'ttfb_ms', 'throughput_per_second', 'bytes', 'status', 'cache_status',
@@@ -138,4 +138,28 @@ run([
              await assert.rejects(pruneTelemetry({ db, olderThanDays: 0 }), /olderThanDays/);
          } finally { await db.close(); }
      }],
++
++    ['startTelemetryRetention prunes soon after start (not a day later), then again, logs a failure and stops', async () => {
++        const db = await open();
++        try {
++            const old = new Date(Date.now() - 40 * 86400e3).toISOString();
++            await createPgTelemetrySink({ db, log: logger() })([{ service: 'blog', operation: 'old', at: old }, { service: 'blog', operation: 'new', at: new Date().toISOString() }]);
++            const errors = [];
++            const r = startTelemetryRetention({ db, delayMs: 20, everyMs: 60, log: { error: (m) => errors.push(String(m)) } });
++            await new Promise((ok) => setTimeout(ok, 250));
++            assert.deepEqual((await db.many('SELECT operation FROM telemetry_samples')).map((x) => x.operation), ['new']);
++            assert.equal(errors.length, 0);
++            assert.equal(await r.run(), 0);
++            await r.stop();
++            await createPgTelemetrySink({ db, log: logger() })([{ service: 'blog', operation: 'old2', at: old }]);
++            await new Promise((ok) => setTimeout(ok, 150));
++            assert.equal(await db.value('SELECT count(*)::int FROM telemetry_samples'), 2, 'nothing prunes after stop()');
++            const bad = startTelemetryRetention({ db: { query: async () => { throw new Error('db down'); } }, delayMs: 5, everyMs: 1000, log: { error: (m) => errors.push(String(m)) } });
++            assert.equal(await bad.run(), 0);
++            await bad.stop();
++            assert.match(errors[0], /retention prune failed: db down/);
++            assert.throws(() => startTelemetryRetention({ db, everyMs: 0 }), /everyMs/);
++            assert.throws(() => startTelemetryRetention({ db, table: 'x; drop' }), /bad telemetry table name/);
++        } finally { await db.close(); }
++    }],
  ]);
ddiff --git a/types/telemetry.d.ts b/types/telemetry.d.ts
iindex b7d3545..6a6ef8d 100644
---- a/types/telemetry.d.ts
++++ b/types/telemetry.d.ts
@@@ -40,3 +40,9 @@ export declare function telemetrySchema(table?: string): string;
  export declare function createPgTelemetrySink(opts: { db: TelemetryDb; table?: string; log?: { warn(...a: unknown[]): void } }): (samples: TelemetrySample[]) => Promise<void>;
  /** Retention: deletes rows whose `at` is older than olderThanDays (default 30); resolves with the count. */
  export declare function pruneTelemetry(opts: { db: TelemetryDb; table?: string; olderThanDays?: number }): Promise<number>;
++/** Scheduled retention: pruneTelemetry() delayMs after the call (default 60000) and then every everyMs (default
++ *  86400000), on unref'd timers, so a service redeployed more than daily still prunes. A failing prune is logged
++ *  and retried at the next tick; runs never overlap. run() prunes now (resolves with the count, 0 on failure);
++ *  stop() cancels the timers and waits for a run in flight. */
++export declare function startTelemetryRetention(opts: { db: TelemetryDb; table?: string; olderThanDays?: number; everyMs?: number; delayMs?: number;
++    log?: { error(...a: unknown[]): void } }): { run(): Promise<number>; stop(): Promise<void> };