openvibers/openvibe.tools Commits
Stop the Billing readings tick when Network refuses a token
Changes
@@@ -311,11 +311,11 @@ reload reattaches.`tool_job_billing_readings` in the transaction that records the end (no network call there); the pruner's timerposts the unsent ones (500 at most a tick) to `POST /api/v1/usage` and marks each sent. A refused post is logged,never thrown: the reading keeps its error and attempt count and is posted again on the next tick (Billing answers-- a replay with 200), and an outage, a 401/403 or a 429 stops that tick's batch. The hourly `tools.usage.recorded`++ a replay with 200), and an outage, a 401/403, a 429 or a refused token stops that tick's batch. The hourly `tools.usage.recorded`rollups above are unchanged and stay subject-less. Off without `OV_BILLING_URL` (nothing is stored). The token isNetwork `client_credentials` for the `tools` client on audience `OV_BILLING_AUDIENCE` (default`openvibe.billing`); Network must grant that client `billing.usage.record` on it — until the grant exists every-- post is refused (401) and only logged, and the readings wait in the table. `GET /api/health` shows++ token request is refused (400 invalid_scope) and only logged, and the readings wait in the table. `GET /api/health` shows`jobs.billing` (pending, posted, refused, invalid, last_error) when it is on.Environment (all in `/etc/openvibe/tools.env`): `TOOLS_WORKERS`, `TOOLS_WORKERS_<APP>`, `TOOLS_WORKER_MEMORY_MB`,@@@ -34,10 +34,12 @@ function createBillingClient({ env = process.env, fetchImpl = globalThis.fetch,/** One reading → Billing. { ok: true, status } when stored (201) or replayed (200); else logged, { ok: false, status, error }. */async function post(reading) {++ let auth;try {++ try { auth = `Bearer ${await bearer()}`; } catch (err) { err.token = true; throw err; }const r = await fetchImpl(`${base}${PATH}`, {method: 'POST',-- headers: { 'Content-Type': 'application/json', Accept: 'application/json', Authorization: `Bearer ${await bearer()}` },++ headers: { 'Content-Type': 'application/json', Accept: 'application/json', Authorization: auth },body: JSON.stringify(reading), signal: AbortSignal.timeout(TIMEOUT_MS),});if (r.ok) return { ok: true, status: r.status };@@@ -49,7 +51,7 @@ function createBillingClient({ env = process.env, fetchImpl = globalThis.fetch,} catch (err) {const error = String(err && err.message || err).slice(0, 300);log.error(`[Billing] reading ${reading && reading.idempotency_key} not sent: ${error}`);-- return { ok: false, status: err && err.status || 0, error };++ return err && err.token ? { ok: false, status: err.status || 0, error, token: true } : { ok: false, status: err && err.status || 0, error };}}@@@ -130,7 +130,7 @@ function createJobUsage({ db, app, contracts, outbox = null, billing = null, nowreadings.lastError = res.error || String(res.status);await r.failed.run({ job_id: row.job_id, error: readings.lastError.slice(0, 500) });// Billing or the token is down (or the grant is missing): stop here, the rest waits for the next call.-- if (!res.status || res.status === 401 || res.status === 403 || res.status === 429 || res.status >= 500) break;++ if (res.token || !res.status || res.status === 401 || res.status === 403 || res.status === 429 || res.status >= 500) break;}await r.prune.run({ app, at: at - KEEP_SENT_MS });readings.posted += posted; readings.refused += refused;@@@ -171,6 +171,7 @@ function define(system) {const form = new URLSearchParams(String(init.body));tokens.push(form.get('audience'));if (mode === 'no-grant') return new Response(JSON.stringify({ error: 'invalid_scope' }), { status: 401 });++ if (mode === 'invalid-scope') return new Response(JSON.stringify({ error: 'invalid_scope' }), { status: 400 });return new Response(JSON.stringify({ access_token: 'tools-billing-token', expires_in: 3600 }), { status: 200 });}assert.strictEqual(String(url), 'http://billing.test/api/v1/usage');@@@ -260,6 +261,21 @@ function define(system) {mode = 'ok';const st = (await bsys.stats()).billing;assert.deepStrictEqual([st.pending, st.posted, st.refused, st.invalid], [0, 5, 3, 0]);++++ // ── Network answers a missing grant with 400 invalid_scope: the tick stops after the first reading ──++ const scopeId = [await brun({ project: PRJ }, 'succeeded'), await brun({ project: PRJ }, 'succeeded'), await brun({ project: PRJ }, 'succeeded')];++ const noScope = createBillingClient({ env: { OV_BILLING_URL: 'http://billing.test', OV_NETWORK_INTERNAL_URL: 'http://network.test' }, fetchImpl: billingFetch, log: blog });++ const ssys = jobs.createJobSystem({ db: bdb, contracts, service: 'img', dataDir: bdir, concurrency: 1, billing: noScope, progressThrottleMs: 0, pruneIntervalMs: 60 * 60 * 1000, log: blog });++ mode = 'invalid-scope';++ const tokensBefore = tokens.length;++ assert.deepStrictEqual(await ssys.flushReadings(), { posted: 0, refused: 1 }, 'a token refusal ends the tick after the first reading');++ assert.strictEqual(tokens.length - tokensBefore, 1, 'one token request in the tick');++ const unsent = (await readingRows()).filter(r => scopeId.includes(r.job_id));++ assert.strictEqual(unsent.length, 3);++ assert.ok(unsent.every(r => r.sent_at == null), 'all three readings are still unsent');++ assert.deepStrictEqual(unsent.map(r => Number(r.attempts)).sort(), [0, 0, 1], 'attempts raised on the first one only');++ assert.ok(unsent.some(r => /^token: 400 invalid_scope/.test(r.last_error)));++ mode = 'ok';bsys.stop();}console.log('job usage: all checks passed');@@@ -61,6 +61,6 @@ MEDIA_PUBLIC_URL=https://openvibe.media#EVENTS_PUBLISH=off#EVENTS_RELAY_INTERVAL_MS=2000# One usage reading per ended job → OpenVibe.Billing POST /api/v1/usage (off while unset). Needs OV_OAUTH_CLIENT_SECRET--# and the Network grant [tools, billing.usage.record, openvibe.billing]; until then every post is 401 and only logged.++# and the Network grant [tools, billing.usage.record, openvibe.billing]; until then every token request is refused (400 invalid_scope) and only logged.#OV_BILLING_URL=http://127.0.0.1:4600#OV_BILLING_AUDIENCE=openvibe.billing@@@ -36,8 +36,8 @@ release wait on the same lock and find it applied.table, nothing is posted, the rollups and job events behave as before.4. **Network grant** (OpenVibe.Network, its own PR): the `tools` client needs`billing.usage.record` on audience `openvibe.billing`. Without it every token request for that audience is-- refused, each post is logged as `[Billing] reading tools:job:<id> not sent: token: 401 …`, and the readings-- wait in the table (no job is affected).++ refused, each post is logged as `[Billing] reading tools:job:<id> not sent: token: 400 invalid_scope`, the tick++ stops after that first reading, and the readings wait in the table (no job is affected).5. **Turn it on** once the grant is live: add `OV_BILLING_URL` (Billing's internal URL) and, only if itdiffers, `OV_BILLING_AUDIENCE` to `/etc/openvibe/tools.env`, then restart through `ovhost deploy tools--restart`.@@@ -49,7 +49,7 @@ release wait on the same lock and find it applied.`0005 | billing_readings | expand`, and `SELECT COUNT(*) FROM tool_job_billing_readings` is 0 while`OV_BILLING_URL` is unset.- After step 5: `GET /api/health` on img, audio or docs shows `jobs.billing` with `pending` going back to 0-- after a tick (5 minutes) and `posted` growing; `refused` with a `last_error` of `401 …` means the grant is++ after a tick (5 minutes) and `posted` growing; `refused` with a `last_error` of `token: 400 invalid_scope` means the grant ismissing. `SELECT COUNT(*) FROM tool_job_billing_readings WHERE sent_at IS NULL` stays small.- In Billing: `GET /api/v1/usage?service=openvibe.tools` (`billing.ledger.admin`) lists the readings, one perended job, `idempotency_key` `tools:job:<job id>`.
@@ -311,11 +311,11 @@ reload reattaches.
@@ -311,11 +311,11 @@ reload reattaches.
`tool_job_billing_readings` in the transaction that records the end (no network call there); the pruner's timer
`tool_job_billing_readings` in the transaction that records the end (no network call there); the pruner's timer
posts the unsent ones (500 at most a tick) to `POST /api/v1/usage` and marks each sent. A refused post is logged,
posts the unsent ones (500 at most a tick) to `POST /api/v1/usage` and marks each sent. A refused post is logged,
never thrown: the reading keeps its error and attempt count and is posted again on the next tick (Billing answers
never thrown: the reading keeps its error and attempt count and is posted again on the next tick (Billing answers
- a replay with 200), and an outage, a 401/403 or a 429 stops that tick's batch. The hourly `tools.usage.recorded`
+ a replay with 200), and an outage, a 401/403, a 429 or a refused token stops that tick's batch. The hourly `tools.usage.recorded`
rollups above are unchanged and stay subject-less. Off without `OV_BILLING_URL` (nothing is stored). The token is
rollups above are unchanged and stay subject-less. Off without `OV_BILLING_URL` (nothing is stored). The token is
Network `client_credentials` for the `tools` client on audience `OV_BILLING_AUDIENCE` (default
Network `client_credentials` for the `tools` client on audience `OV_BILLING_AUDIENCE` (default
`openvibe.billing`); Network must grant that client `billing.usage.record` on it — until the grant exists every
`openvibe.billing`); Network must grant that client `billing.usage.record` on it — until the grant exists every
- post is refused (401) and only logged, and the readings wait in the table. `GET /api/health` shows
+ token request is refused (400 invalid_scope) and only logged, and the readings wait in the table. `GET /api/health` shows
`jobs.billing` (pending, posted, refused, invalid, last_error) when it is on.
`jobs.billing` (pending, posted, refused, invalid, last_error) when it is on.
Environment (all in `/etc/openvibe/tools.env`): `TOOLS_WORKERS`, `TOOLS_WORKERS_<APP>`, `TOOLS_WORKER_MEMORY_MB`,
Environment (all in `/etc/openvibe/tools.env`): `TOOLS_WORKERS`, `TOOLS_WORKERS_<APP>`, `TOOLS_WORKER_MEMORY_MB`,
@@ -34,10 +34,12 @@ function createBillingClient({ env = process.env, fetchImpl = globalThis.fetch,
@@ -34,10 +34,12 @@ function createBillingClient({ env = process.env, fetchImpl = globalThis.fetch,
/** One reading → Billing. { ok: true, status } when stored (201) or replayed (200); else logged, { ok: false, status, error }. */
/** One reading → Billing. { ok: true, status } when stored (201) or replayed (200); else logged, { ok: false, status, error }. */
async function post(reading) {
async function post(reading) {
+ let auth;
try {
try {
+ try { auth = `Bearer ${await bearer()}`; } catch (err) { err.token = true; throw err; }
const r = await fetchImpl(`${base}${PATH}`, {
const r = await fetchImpl(`${base}${PATH}`, {
method: 'POST',
method: 'POST',
- headers: { 'Content-Type': 'application/json', Accept: 'application/json', Authorization: `Bearer ${await bearer()}` },
+ headers: { 'Content-Type': 'application/json', Accept: 'application/json', Authorization: auth },
body: JSON.stringify(reading), signal: AbortSignal.timeout(TIMEOUT_MS),
body: JSON.stringify(reading), signal: AbortSignal.timeout(TIMEOUT_MS),
});
});
if (r.ok) return { ok: true, status: r.status };
if (r.ok) return { ok: true, status: r.status };
@@ -49,7 +51,7 @@ function createBillingClient({ env = process.env, fetchImpl = globalThis.fetch,
@@ -49,7 +51,7 @@ function createBillingClient({ env = process.env, fetchImpl = globalThis.fetch,
} catch (err) {
} catch (err) {
const error = String(err && err.message || err).slice(0, 300);
const error = String(err && err.message || err).slice(0, 300);
log.error(`[Billing] reading ${reading && reading.idempotency_key} not sent: ${error}`);
log.error(`[Billing] reading ${reading && reading.idempotency_key} not sent: ${error}`);
- return { ok: false, status: err && err.status || 0, error };
+ return err && err.token ? { ok: false, status: err.status || 0, error, token: true } : { ok: false, status: err && err.status || 0, error };
}
}
}
}
@@ -130,7 +130,7 @@ function createJobUsage({ db, app, contracts, outbox = null, billing = null, now
@@ -130,7 +130,7 @@ function createJobUsage({ db, app, contracts, outbox = null, billing = null, now
readings.lastError = res.error || String(res.status);
readings.lastError = res.error || String(res.status);
await r.failed.run({ job_id: row.job_id, error: readings.lastError.slice(0, 500) });
await r.failed.run({ job_id: row.job_id, error: readings.lastError.slice(0, 500) });
// Billing or the token is down (or the grant is missing): stop here, the rest waits for the next call.
// Billing or the token is down (or the grant is missing): stop here, the rest waits for the next call.
- if (!res.status || res.status === 401 || res.status === 403 || res.status === 429 || res.status >= 500) break;
+ if (res.token || !res.status || res.status === 401 || res.status === 403 || res.status === 429 || res.status >= 500) break;
}
}
await r.prune.run({ app, at: at - KEEP_SENT_MS });
await r.prune.run({ app, at: at - KEEP_SENT_MS });
readings.posted += posted; readings.refused += refused;
readings.posted += posted; readings.refused += refused;
@@ -171,6 +171,7 @@ function define(system) {
@@ -171,6 +171,7 @@ function define(system) {
const form = new URLSearchParams(String(init.body));
const form = new URLSearchParams(String(init.body));
tokens.push(form.get('audience'));
tokens.push(form.get('audience'));
if (mode === 'no-grant') return new Response(JSON.stringify({ error: 'invalid_scope' }), { status: 401 });
if (mode === 'no-grant') return new Response(JSON.stringify({ error: 'invalid_scope' }), { status: 401 });
+ if (mode === 'invalid-scope') return new Response(JSON.stringify({ error: 'invalid_scope' }), { status: 400 });
return new Response(JSON.stringify({ access_token: 'tools-billing-token', expires_in: 3600 }), { status: 200 });
return new Response(JSON.stringify({ access_token: 'tools-billing-token', expires_in: 3600 }), { status: 200 });
}
}
assert.strictEqual(String(url), 'http://billing.test/api/v1/usage');
assert.strictEqual(String(url), 'http://billing.test/api/v1/usage');
@@ -260,6 +261,21 @@ function define(system) {
@@ -260,6 +261,21 @@ function define(system) {
mode = 'ok';
mode = 'ok';
const st = (await bsys.stats()).billing;
const st = (await bsys.stats()).billing;
assert.deepStrictEqual([st.pending, st.posted, st.refused, st.invalid], [0, 5, 3, 0]);
assert.deepStrictEqual([st.pending, st.posted, st.refused, st.invalid], [0, 5, 3, 0]);
+
+ // ── Network answers a missing grant with 400 invalid_scope: the tick stops after the first reading ──
+ const scopeId = [await brun({ project: PRJ }, 'succeeded'), await brun({ project: PRJ }, 'succeeded'), await brun({ project: PRJ }, 'succeeded')];
+ const noScope = createBillingClient({ env: { OV_BILLING_URL: 'http://billing.test', OV_NETWORK_INTERNAL_URL: 'http://network.test' }, fetchImpl: billingFetch, log: blog });
+ const ssys = jobs.createJobSystem({ db: bdb, contracts, service: 'img', dataDir: bdir, concurrency: 1, billing: noScope, progressThrottleMs: 0, pruneIntervalMs: 60 * 60 * 1000, log: blog });
+ mode = 'invalid-scope';
+ const tokensBefore = tokens.length;
+ assert.deepStrictEqual(await ssys.flushReadings(), { posted: 0, refused: 1 }, 'a token refusal ends the tick after the first reading');
+ assert.strictEqual(tokens.length - tokensBefore, 1, 'one token request in the tick');
+ const unsent = (await readingRows()).filter(r => scopeId.includes(r.job_id));
+ assert.strictEqual(unsent.length, 3);
+ assert.ok(unsent.every(r => r.sent_at == null), 'all three readings are still unsent');
+ assert.deepStrictEqual(unsent.map(r => Number(r.attempts)).sort(), [0, 0, 1], 'attempts raised on the first one only');
+ assert.ok(unsent.some(r => /^token: 400 invalid_scope/.test(r.last_error)));
+ mode = 'ok';
bsys.stop();
bsys.stop();
}
}
console.log('job usage: all checks passed');
console.log('job usage: all checks passed');
@@ -61,6 +61,6 @@ MEDIA_PUBLIC_URL=https://openvibe.media
@@ -61,6 +61,6 @@ MEDIA_PUBLIC_URL=https://openvibe.media
#EVENTS_PUBLISH=off
#EVENTS_PUBLISH=off
#EVENTS_RELAY_INTERVAL_MS=2000
#EVENTS_RELAY_INTERVAL_MS=2000
# One usage reading per ended job → OpenVibe.Billing POST /api/v1/usage (off while unset). Needs OV_OAUTH_CLIENT_SECRET
# One usage reading per ended job → OpenVibe.Billing POST /api/v1/usage (off while unset). Needs OV_OAUTH_CLIENT_SECRET
-# and the Network grant [tools, billing.usage.record, openvibe.billing]; until then every post is 401 and only logged.
+# and the Network grant [tools, billing.usage.record, openvibe.billing]; until then every token request is refused (400 invalid_scope) and only logged.
#OV_BILLING_URL=http://127.0.0.1:4600
#OV_BILLING_URL=http://127.0.0.1:4600
#OV_BILLING_AUDIENCE=openvibe.billing
#OV_BILLING_AUDIENCE=openvibe.billing
@@ -36,8 +36,8 @@ release wait on the same lock and find it applied.
@@ -36,8 +36,8 @@ release wait on the same lock and find it applied.
table, nothing is posted, the rollups and job events behave as before.
table, nothing is posted, the rollups and job events behave as before.
4. **Network grant** (OpenVibe.Network, its own PR): the `tools` client needs
4. **Network grant** (OpenVibe.Network, its own PR): the `tools` client needs
`billing.usage.record` on audience `openvibe.billing`. Without it every token request for that audience is
`billing.usage.record` on audience `openvibe.billing`. Without it every token request for that audience is
- refused, each post is logged as `[Billing] reading tools:job:<id> not sent: token: 401 …`, and the readings
- wait in the table (no job is affected).
+ refused, each post is logged as `[Billing] reading tools:job:<id> not sent: token: 400 invalid_scope`, the tick
+ stops after that first reading, and the readings wait in the table (no job is affected).
5. **Turn it on** once the grant is live: add `OV_BILLING_URL` (Billing's internal URL) and, only if it
5. **Turn it on** once the grant is live: add `OV_BILLING_URL` (Billing's internal URL) and, only if it
differs, `OV_BILLING_AUDIENCE` to `/etc/openvibe/tools.env`, then restart through `ovhost deploy tools
differs, `OV_BILLING_AUDIENCE` to `/etc/openvibe/tools.env`, then restart through `ovhost deploy tools
--restart`.
--restart`.
@@ -49,7 +49,7 @@ release wait on the same lock and find it applied.
@@ -49,7 +49,7 @@ release wait on the same lock and find it applied.
`0005 | billing_readings | expand`, and `SELECT COUNT(*) FROM tool_job_billing_readings` is 0 while
`0005 | billing_readings | expand`, and `SELECT COUNT(*) FROM tool_job_billing_readings` is 0 while
`OV_BILLING_URL` is unset.
`OV_BILLING_URL` is unset.
- After step 5: `GET /api/health` on img, audio or docs shows `jobs.billing` with `pending` going back to 0
- After step 5: `GET /api/health` on img, audio or docs shows `jobs.billing` with `pending` going back to 0
- after a tick (5 minutes) and `posted` growing; `refused` with a `last_error` of `401 …` means the grant is
+ after a tick (5 minutes) and `posted` growing; `refused` with a `last_error` of `token: 400 invalid_scope` means the grant is
missing. `SELECT COUNT(*) FROM tool_job_billing_readings WHERE sent_at IS NULL` stays small.
missing. `SELECT COUNT(*) FROM tool_job_billing_readings WHERE sent_at IS NULL` stays small.
- In Billing: `GET /api/v1/usage?service=openvibe.tools` (`billing.ledger.admin`) lists the readings, one per
- In Billing: `GET /api/v1/usage?service=openvibe.tools` (`billing.ledger.admin`) lists the readings, one per
ended job, `idempotency_key` `tools:job:<job id>`.
ended job, `idempotency_key` `tools:job:<job id>`.