openvibers/openvibe.actor Commits
Actor: the daily spend caps are checked and reserved atomically (no overrun by concurrent task creation) (#10)
Changes
@@@ -36,18 +36,26 @@ async function createTask({ s, config, engine, principal, body }) {}}const retryAfter = untilMidnight(s.now());-- const mine = await store.spendToday(s, principal.requester);-- if (mine.tasks >= tier.tasksPerDay) return { status: 429, code: 'actor.allowance.exhausted', detail: `The free tier runs ${tier.tasksPerDay} tasks a day; it resets at midnight UTC.`, retryAfter };-- if (mine.usd >= perDay) return { status: 429, code: 'actor.budget.day_spent', detail: `Today's tasks already cost $${mine.usd.toFixed(3)} of the $${perDay} daily budget; it resets at midnight UTC.`, retryAfter };-- const all = await store.spendToday(s, '*');-- if (all.usd >= config.spendCapUsdPerDay) return { status: 503, code: 'actor.capacity.spent', detail: 'Actor has used today\'s free capacity across everyone. It resets at midnight UTC.', retryAfter };---- const row = await store.insertTask(s, {-- id: s.newId('tsk'), requester: principal.requester, project_id: principal.project || null, task: body.task, mode: body.mode || 'balanced', agent: body.agent || null,-- budget_task: perTask, budget_day: perDay, idem_key: body.idempotency_key || null, idem_hash: body.idempotency_key ? hash : null, created_at: s.iso(),++ // Check and insert in one transaction under the day's spend-row locks, counting what open tasks may still spend:++ // concurrent creations cannot all pass on the same reading and overrun the person's or the operator's cap.++ const out = await s.tx(async () => {++ const { mine, all } = await store.lockSpendToday(s, principal.requester);++ if (mine.tasks >= tier.tasksPerDay) return { status: 429, code: 'actor.allowance.exhausted', detail: `The free tier runs ${tier.tasksPerDay} tasks a day; it resets at midnight UTC.`, retryAfter };++ if (mine.usd >= perDay || mine.usd + mine.reserved + perTask > perDay + 1e-9) {++ return { status: 429, code: 'actor.budget.day_spent', detail: `Today's tasks cost $${mine.usd.toFixed(3)} and running ones may use $${mine.reserved.toFixed(3)} of the $${perDay} daily budget; it resets at midnight UTC.`, retryAfter };++ }++ if (all.usd >= config.spendCapUsdPerDay || all.usd + all.reserved + perTask > config.spendCapUsdPerDay + 1e-9) {++ return { status: 503, code: 'actor.capacity.spent', detail: 'Actor has used today\'s free capacity across everyone. It resets at midnight UTC.', retryAfter };++ }++ const row = await store.insertTask(s, {++ id: s.newId('tsk'), requester: principal.requester, project_id: principal.project || null, task: body.task, mode: body.mode || 'balanced', agent: body.agent || null,++ budget_task: perTask, budget_day: perDay, idem_key: body.idempotency_key || null, idem_hash: body.idempotency_key ? hash : null, created_at: s.iso(),++ });++ return { status: 201, task: row };});-- engine.submit(row.id);-- return { status: 201, task: row };++ // The engine starts the task only once its row has committed.++ if (out.status === 201) engine.submit(out.task.id);++ return out;}module.exports = { createTask, untilMidnight };@@@ -77,6 +77,25 @@ async function addSpend(s, requester, usd, { tasks = 0 } = {}) {}}++/**++ * Lock today's spend rows for a check-then-insert (the global '*' row first, then the person's, so two creations++ * never deadlock) and read them with what open tasks may still spend (their per-task budgets: spend accrues as a++ * task runs, so a check that only saw accrued spend would let concurrent tasks overrun the caps). Inside a++ * transaction. → { mine: { usd, tasks, reserved }, all: { usd, reserved } }++ */++async function lockSpendToday(s, requester) {++ const day = dayOf(s.now());++ await s.db.query(`INSERT INTO spend_daily (requester, day, usd, tasks) VALUES ('*', $1, 0, 0), ($2, $1, 0, 0) ON CONFLICT (requester, day) DO NOTHING`, [day, requester]);++ const all = await s.db.maybe(`SELECT usd FROM spend_daily WHERE requester = '*' AND day = $1 FOR UPDATE`, [day]);++ const mine = await s.db.maybe('SELECT usd, tasks FROM spend_daily WHERE requester = $1 AND day = $2 FOR UPDATE', [requester, day]);++ const open = await s.db.maybe(`SELECT COALESCE(sum(budget_task), 0) AS all_reserved, COALESCE(sum(budget_task) FILTER (WHERE requester = $1), 0) AS mine_reserved++ FROM tasks WHERE state = ANY($2)`, [requester, OPEN]);++ return {++ mine: { usd: Number(mine.usd), tasks: Number(mine.tasks), reserved: Number(open.mine_reserved) },++ all: { usd: Number(all.usd), reserved: Number(open.all_reserved) },++ };++}++async function spendToday(s, requester) {const row = await s.db.maybe('SELECT usd, tasks FROM spend_daily WHERE requester = $1 AND day = $2', [requester, dayOf(s.now())]);return { usd: row ? Number(row.usd) : 0, tasks: row ? Number(row.tasks) : 0 };@@@ -124,4 +143,4 @@ function toWire(row, { baseUrl }) {return t;}--module.exports = { insertTask, getTask, byIdempotency, listTasks, updateTask, appendEvent, eventsAfter, addSpend, spendToday, failInterrupted, prune, toWire, subjectRef, dayOf, OPEN, RETENTION_DAYS };++module.exports = { insertTask, getTask, byIdempotency, listTasks, updateTask, appendEvent, eventsAfter, addSpend, spendToday, lockSpendToday, failInterrupted, prune, toWire, subjectRef, dayOf, OPEN, RETENTION_DAYS };@@@ -276,6 +276,23 @@ const SAME = { 'sec-fetch-site': 'same-origin' };assert.ok(Number(r.headers.get('retry-after')) > 0);});++ await check('tasks still running count against the daily budget, so creating many at once cannot overrun it', async () => {++ const store = require('../server/tasks/store');++ const u = t.network.addUser('burst');++ const who = `user:${u.subject}`;++ const { perTaskUsd, perDayUsd } = t.config.allowance;++ const fit = Math.floor(perDayUsd / perTaskUsd + 1e-9);++ // `fit` tasks are open (queued) with nothing spent yet: their budgets are reserved.++ for (let i = 0; i < fit; i++) {++ await store.insertTask(t.ctx.s, { id: t.ctx.s.newId('tsk'), requester: who, task: `open ${i}`, mode: 'balanced', budget_task: perTaskUsd, budget_day: perDayUsd, created_at: t.ctx.s.iso() });++ }++ assert.strictEqual((await store.spendToday(t.ctx.s, who)).usd, 0, 'nothing has accrued yet');++ const r = await create({ task: 'one too many' }, { as: u });++ assert.strictEqual(r.status, 429, r.text);++ assert.strictEqual(r.json().code, 'actor.budget.day_spent');++ await t.ctx.s.db.query("UPDATE tasks SET state = 'cancelled' WHERE requester = $1", [who]);++ });++await check('the operator\'s daily ceiling over everyone stops new tasks', async () => {const store = require('../server/tasks/store');await store.addSpend(t.ctx.s, 'user:usr_01JAB2C3D4E5F6G7H8J9K0MNPZ', t.config.spendCapUsdPerDay);
@@ -36,18 +36,26 @@ async function createTask({ s, config, engine, principal, body }) {
@@ -36,18 +36,26 @@ async function createTask({ s, config, engine, principal, body }) {
}
}
}
}
const retryAfter = untilMidnight(s.now());
const retryAfter = untilMidnight(s.now());
- const mine = await store.spendToday(s, principal.requester);
- if (mine.tasks >= tier.tasksPerDay) return { status: 429, code: 'actor.allowance.exhausted', detail: `The free tier runs ${tier.tasksPerDay} tasks a day; it resets at midnight UTC.`, retryAfter };
- if (mine.usd >= perDay) return { status: 429, code: 'actor.budget.day_spent', detail: `Today's tasks already cost $${mine.usd.toFixed(3)} of the $${perDay} daily budget; it resets at midnight UTC.`, retryAfter };
- const all = await store.spendToday(s, '*');
- if (all.usd >= config.spendCapUsdPerDay) return { status: 503, code: 'actor.capacity.spent', detail: 'Actor has used today\'s free capacity across everyone. It resets at midnight UTC.', retryAfter };
-
- const row = await store.insertTask(s, {
- id: s.newId('tsk'), requester: principal.requester, project_id: principal.project || null, task: body.task, mode: body.mode || 'balanced', agent: body.agent || null,
- budget_task: perTask, budget_day: perDay, idem_key: body.idempotency_key || null, idem_hash: body.idempotency_key ? hash : null, created_at: s.iso(),
+ // Check and insert in one transaction under the day's spend-row locks, counting what open tasks may still spend:
+ // concurrent creations cannot all pass on the same reading and overrun the person's or the operator's cap.
+ const out = await s.tx(async () => {
+ const { mine, all } = await store.lockSpendToday(s, principal.requester);
+ if (mine.tasks >= tier.tasksPerDay) return { status: 429, code: 'actor.allowance.exhausted', detail: `The free tier runs ${tier.tasksPerDay} tasks a day; it resets at midnight UTC.`, retryAfter };
+ if (mine.usd >= perDay || mine.usd + mine.reserved + perTask > perDay + 1e-9) {
+ return { status: 429, code: 'actor.budget.day_spent', detail: `Today's tasks cost $${mine.usd.toFixed(3)} and running ones may use $${mine.reserved.toFixed(3)} of the $${perDay} daily budget; it resets at midnight UTC.`, retryAfter };
+ }
+ if (all.usd >= config.spendCapUsdPerDay || all.usd + all.reserved + perTask > config.spendCapUsdPerDay + 1e-9) {
+ return { status: 503, code: 'actor.capacity.spent', detail: 'Actor has used today\'s free capacity across everyone. It resets at midnight UTC.', retryAfter };
+ }
+ const row = await store.insertTask(s, {
+ id: s.newId('tsk'), requester: principal.requester, project_id: principal.project || null, task: body.task, mode: body.mode || 'balanced', agent: body.agent || null,
+ budget_task: perTask, budget_day: perDay, idem_key: body.idempotency_key || null, idem_hash: body.idempotency_key ? hash : null, created_at: s.iso(),
+ });
+ return { status: 201, task: row };
});
});
- engine.submit(row.id);
- return { status: 201, task: row };
+ // The engine starts the task only once its row has committed.
+ if (out.status === 201) engine.submit(out.task.id);
+ return out;
}
}
module.exports = { createTask, untilMidnight };
module.exports = { createTask, untilMidnight };
@@ -77,6 +77,25 @@ async function addSpend(s, requester, usd, { tasks = 0 } = {}) {
@@ -77,6 +77,25 @@ async function addSpend(s, requester, usd, { tasks = 0 } = {}) {
}
}
}
}
+/**
+ * Lock today's spend rows for a check-then-insert (the global '*' row first, then the person's, so two creations
+ * never deadlock) and read them with what open tasks may still spend (their per-task budgets: spend accrues as a
+ * task runs, so a check that only saw accrued spend would let concurrent tasks overrun the caps). Inside a
+ * transaction. → { mine: { usd, tasks, reserved }, all: { usd, reserved } }
+ */
+async function lockSpendToday(s, requester) {
+ const day = dayOf(s.now());
+ await s.db.query(`INSERT INTO spend_daily (requester, day, usd, tasks) VALUES ('*', $1, 0, 0), ($2, $1, 0, 0) ON CONFLICT (requester, day) DO NOTHING`, [day, requester]);
+ const all = await s.db.maybe(`SELECT usd FROM spend_daily WHERE requester = '*' AND day = $1 FOR UPDATE`, [day]);
+ const mine = await s.db.maybe('SELECT usd, tasks FROM spend_daily WHERE requester = $1 AND day = $2 FOR UPDATE', [requester, day]);
+ const open = await s.db.maybe(`SELECT COALESCE(sum(budget_task), 0) AS all_reserved, COALESCE(sum(budget_task) FILTER (WHERE requester = $1), 0) AS mine_reserved
+ FROM tasks WHERE state = ANY($2)`, [requester, OPEN]);
+ return {
+ mine: { usd: Number(mine.usd), tasks: Number(mine.tasks), reserved: Number(open.mine_reserved) },
+ all: { usd: Number(all.usd), reserved: Number(open.all_reserved) },
+ };
+}
+
async function spendToday(s, requester) {
async function spendToday(s, requester) {
const row = await s.db.maybe('SELECT usd, tasks FROM spend_daily WHERE requester = $1 AND day = $2', [requester, dayOf(s.now())]);
const row = await s.db.maybe('SELECT usd, tasks FROM spend_daily WHERE requester = $1 AND day = $2', [requester, dayOf(s.now())]);
return { usd: row ? Number(row.usd) : 0, tasks: row ? Number(row.tasks) : 0 };
return { usd: row ? Number(row.usd) : 0, tasks: row ? Number(row.tasks) : 0 };
@@ -124,4 +143,4 @@ function toWire(row, { baseUrl }) {
@@ -124,4 +143,4 @@ function toWire(row, { baseUrl }) {
return t;
return t;
}
}
-module.exports = { insertTask, getTask, byIdempotency, listTasks, updateTask, appendEvent, eventsAfter, addSpend, spendToday, failInterrupted, prune, toWire, subjectRef, dayOf, OPEN, RETENTION_DAYS };
+module.exports = { insertTask, getTask, byIdempotency, listTasks, updateTask, appendEvent, eventsAfter, addSpend, spendToday, lockSpendToday, failInterrupted, prune, toWire, subjectRef, dayOf, OPEN, RETENTION_DAYS };
@@ -276,6 +276,23 @@ const SAME = { 'sec-fetch-site': 'same-origin' };
@@ -276,6 +276,23 @@ const SAME = { 'sec-fetch-site': 'same-origin' };
assert.ok(Number(r.headers.get('retry-after')) > 0);
assert.ok(Number(r.headers.get('retry-after')) > 0);
});
});
+ await check('tasks still running count against the daily budget, so creating many at once cannot overrun it', async () => {
+ const store = require('../server/tasks/store');
+ const u = t.network.addUser('burst');
+ const who = `user:${u.subject}`;
+ const { perTaskUsd, perDayUsd } = t.config.allowance;
+ const fit = Math.floor(perDayUsd / perTaskUsd + 1e-9);
+ // `fit` tasks are open (queued) with nothing spent yet: their budgets are reserved.
+ for (let i = 0; i < fit; i++) {
+ await store.insertTask(t.ctx.s, { id: t.ctx.s.newId('tsk'), requester: who, task: `open ${i}`, mode: 'balanced', budget_task: perTaskUsd, budget_day: perDayUsd, created_at: t.ctx.s.iso() });
+ }
+ assert.strictEqual((await store.spendToday(t.ctx.s, who)).usd, 0, 'nothing has accrued yet');
+ const r = await create({ task: 'one too many' }, { as: u });
+ assert.strictEqual(r.status, 429, r.text);
+ assert.strictEqual(r.json().code, 'actor.budget.day_spent');
+ await t.ctx.s.db.query("UPDATE tasks SET state = 'cancelled' WHERE requester = $1", [who]);
+ });
+
await check('the operator\'s daily ceiling over everyone stops new tasks', async () => {
await check('the operator\'s daily ceiling over everyone stops new tasks', async () => {
const store = require('../server/tasks/store');
const store = require('../server/tasks/store');
await store.addSpend(t.ctx.s, 'user:usr_01JAB2C3D4E5F6G7H8J9K0MNPZ', t.config.spendCapUsdPerDay);
await store.addSpend(t.ctx.s, 'user:usr_01JAB2C3D4E5F6G7H8J9K0MNPZ', t.config.spendCapUsdPerDay);