openvibers/openvibe.run Commits
Run: a job may be pinned to one machine, and is never moved (#117)
Changes
@@@ -15,7 +15,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.250.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0","pg": "^8.23.0"@@@ -1147,9 +1147,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.223.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",-- "integrity": "sha512-FWcPL1951IeYcelp3dUiDRJEiFkTbsLMLth9RVUFfuLTuHGQKoQXA5HIu6H2e2BupPTyBH+AZD00LqmA6jDmdg==",++ "version": "0.250.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.250.0",++ "integrity": "sha512-fABKEYreTuvn5OquY9S5MGc90PQZTppO+evcZLkQyH7kWIdFZB0a5pI40ij7Y4JOaYWAFT8TPP1+FWaX/omynQ==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -26,7 +26,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.250.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0","pg": "^8.23.0"@@@ -144,6 +144,14 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, nowconst { project: projectId, requester, via } = apiAuth.requireProject(req, { delegated: true });const body = validCreate(req.body);checkLimits(body, config.limits);++ // A pinned machine (Codes C9c): the job is refused now when it is not connected, never placed on another one.++ if (body.node) {++ let pin;++ try { pin = await dispatch.pinStatus(body.node, body.class, requester, { fresh: true }); } catch (e) {++ fail(503, 'run.dispatch.no_offers', `Run could not read which machines are connected (${e.code || e.message})`);++ }++ if (!pin.ok) fail(pin.code === 'run.job.node_offline' ? 409 : 422, pin.code, pin.detail);++ }const key = idempotencyKey(req, body);const at = now();const out = await db.tx(async (t) => {@@@ -81,6 +81,9 @@ function createDispatcher({ config = null, db = null, store = null, events = nulreturn {enabled, note, place, cancel, poll, sweep,++ // Codes C9c: may this job run pinned to that machine now? → { ok, offer? , code?, detail? }; throws when++ // Network cannot be read (the caller answers 503, never a guessed offline).++ pinStatus: (nodeId, runtimeClass, requester, opts) => placer.pinStatus(nodeId, runtimeClass, requester, opts),start: () => { relay.start(); return poller.start(); },stop: () => { poller.stop(); return relay.stop(); },status: () => ({@@@ -83,9 +83,9 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat: null;let owned = { at: 0, offers: [] };/** Every live user-owned node offer that names its owner, from the internal list, cached like the public ones. */-- async function ownedOffers() {++ async function ownedOffers({ maxAgeMs = offersTtlMs } = {}) {if (!registryTokens) return [];-- if (owned.at && now() - owned.at < offersTtlMs) return owned.offers;++ if (owned.at && now() - owned.at < maxAgeMs) return owned.offers;const href = `${config.network.internalUrl}/internal/resources?kind=node&trust=user-owned`;let res;try {@@@ -110,9 +110,9 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat}/** The published offers of one kind, cached briefly. A Network that does not answer is a retryable error. */-- async function offers(kind = 'node') {++ async function offers(kind = 'node', { maxAgeMs = offersTtlMs } = {}) {const cache = caches.get(kind) || { at: 0, offers: [], fetched: 0 };-- if (cache.at && now() - cache.at < offersTtlMs) return cache.offers;++ if (cache.at && now() - cache.at < maxAgeMs) return cache.offers;const href = offersUrlOf(kind);let res;try {@@@ -139,12 +139,51 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Datreturn good;}++ /**++ * Whether a job may be pinned to this machine now (Codes C9c): a machine of the requester's own (a user-owned++ * offer names them), advertising the class, and up. Run refuses a pinned job when it is not — never places it++ * somewhere else. A caller at submit asks with fresh: true (one read, not the poller's cached one); the++ * placement loop asks from its cache, and a machine that went offline between the two simply keeps the job++ * queued here until it is back or the job's ttl runs out.++ *++ * @returns {Promise<{ok: boolean, offer?: object, code?: string, detail?: string}>}++ */++ async function pinStatus(nodeId, runtimeClass, requester = null, { fresh = false } = {}) {++ const want = workerCapability(runtimeClass);++ if (!/^dev_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(nodeId || ''))) {++ return { ok: false, code: 'run.job.node_offline', detail: 'node must be a machine id (dev_…)' };++ }++ const age = fresh ? 0 : offersTtlMs;++ const offer = [...(await offers('node', { maxAgeMs: age })), ...(await ownedOffers({ maxAgeMs: age }))]++ .find((o) => o && o.kind === 'node' && o.node_id === nodeId);++ if (!offer) return { ok: false, code: 'run.job.node_offline', detail: `machine ${nodeId} is not connected to Run` };++ if (offer.trust === 'user-owned') {++ const person = personOf(requester);++ if (!person || offer.owner !== person) return { ok: false, code: 'run.job.node_offline', detail: `machine ${nodeId} is not one of yours` };++ }++ if (!(offer.capabilities || []).includes(want)) return { ok: false, code: 'run.job.worker_failed', detail: `machine ${nodeId} does not run ${runtimeClass} jobs` };++ if (!offer.health || offer.health.status !== 'up') return { ok: false, code: 'run.job.node_offline', detail: `machine ${nodeId} is not connected` };++ return { ok: true, offer };++ }++/*** The node to run this job on, or null when no offer is eligible. The requirements are the job's own-- * (the contract's default was filled in at submit), plus the class it runs.++ * (the contract's default was filled in at submit), plus the class it runs. A job that names a machine++ * (Codes C9c) is placed only there: the placer is not asked, and a machine that is not usable is an error,++ * never a quieter placement somewhere else.*/async function choose(row) {const want = workerCapability(row.class);++ const pinned = typeof obj(row.job).node === 'string' ? obj(row.job).node : null;++ if (pinned) {++ const status = await pinStatus(pinned, row.class, row.requester);++ if (!status.ok) throw new PlacementError(status.code, status.detail, { retryable: status.code === 'run.job.node_offline' });++ const o = status.offer;++ return {++ node_id: o.node_id, provider: o.provider ?? null, region: o.region ?? null, kind: o.kind,++ offer_id: o.offer_id, trust: o.trust, reasons: [`pinned to ${o.node_id}`],++ };++ }const requirements = { ...obj(row.job).requirements || requirementsDefault(row.class) };const userOwnedNamed = Array.isArray(requirements.trust) && requirements.trust.includes('user-owned');// A user-owned offer is a person's own machine: eligible only for that person's own job, and only when the job@@@ -186,6 +225,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Datreturn {choose,++ pinStatus,offers,workerCapability,status: () => {@@@ -66,6 +66,9 @@ function recordOf(row) {ttl_ms: Number(row.ttl_ms),egress: job.net ?? 'none',requirements: job.requirements ?? requirementsDefault(row.class),++ // A pinned machine (Codes C9c): part of what an idempotent repeat is compared on, so a repeat that adds or++ // drops the pin with the same key is a different request, never silently the first job.++ node: job.node ?? null,};}@@@ -85,6 +88,8 @@ function normalize(body, { projectId, requester, via = null, idempotencyKey, nownet: body.egress || 'none',inputs: body.inputs || [],requirements: body.requirements || requirementsDefault(body.class),++ // The machine the caller pinned (Codes C9c): kept on the job record, never in the platform.job@1 frame.++ ...(body.node ? { node: body.node } : {}),};// The platform.job@1 body the node will be sent (frameOf drops requirements — platform.job@1 has no// field for it), validated here once so a node can never be handed something off-contract.@@@ -432,6 +432,71 @@ async function main() {bot.devices.delete(OTHER_DEVICE);});++ await check('a pinned machine (Codes C9c): placed only there, refused when it is offline, never moved', async () => {++ const ada = ids.newId('user');++ const bo = ids.newId('user');++ const P = PROJECT();++ const PINNED = `dev_${ids.ulid()}`; // Bot's own id for a device (Run's placement node_id)++ const ELSEWHERE = `dev_${ids.ulid()}`;++ t.network.members.set(`${P} ${ada}`, { role: 'owner' });++ t.network.members.set(`${P} ${bo}`, { role: 'owner' });++ bot.devices.set(PINNED, { classes: ['function'] });++ const mine = offer({ offer_id: 'off_mine', trust: 'user-owned', owner: ada, provider: 'self', node_id: PINNED });++ const elsewhere = offer({ offer_id: 'off_other', node_id: ELSEWHERE });++ t.setOffers([elsewhere, mine]);++ const asAda = async (over, key = null) => t.call('POST', '/api/v1/jobs', {++ body: t.request(over), key,++ token: t.network.signService({ sub: 'svc:codes', cap: ['run.job.submit', 'run.job.read'] }),++ headers: { 'X-OV-Subject': ada, 'X-OV-Project': P },++ });++++ // Ada's own machine: created, kept on the job record, and placed there — not on the first-party offer.++ const pinned = await asAda({ node: PINNED });++ assert.strictEqual(pinned.status, 201, pinned.text);++ const stored = await t.db.maybe('SELECT job FROM run_jobs WHERE id = $1', [pinned.json.job.id]);++ assert.strictEqual(stored.job.node, PINNED);++ const placed = (await t.dispatch.poll()).find((a) => a.job_id === pinned.json.job.id);++ assert.strictEqual(placed.state, 'placed');++ assert.strictEqual(placed.node_id, PINNED);++ assert.strictEqual(bot.jobs.get(pinned.json.job.id).job.node, undefined, 'node is Run\'s, never a platform.job@1 field');++++ // Offline: refused now, and nothing is stored.++ const before = Number((await t.db.maybe('SELECT count(*) AS n FROM run_jobs')).n);++ t.setOffers([elsewhere]);++ const offline = await asAda({ node: PINNED });++ assert.strictEqual(offline.status, 409, offline.text);++ assert.strictEqual(offline.json.code, 'run.job.node_offline');++ assert.strictEqual(Number((await t.db.maybe('SELECT count(*) AS n FROM run_jobs')).n), before, 'a refused pin stores no job');++++ // Someone else's machine is never pinnable, and a class it does not run is a plain refusal.++ t.setOffers([elsewhere, offer({ offer_id: 'off_mine', trust: 'user-owned', owner: bo, node_id: PINNED })]);++ assert.strictEqual((await asAda({ node: PINNED })).json.code, 'run.job.node_offline');++ t.setOffers([elsewhere, mine]);++ const wrongClass = await asAda({ class: 'code', artifact: { name: 'openvibe-codes', version: '1.0.0' }, node: PINNED });++ assert.strictEqual(wrongClass.status, 422, wrongClass.text);++ assert.strictEqual(wrongClass.json.code, 'run.job.worker_failed');++++ // Pinned, then the machine goes down before placement: the job waits, and is never moved to the other one.++ const later = await asAda({ node: PINNED });++ assert.strictEqual(later.status, 201, later.text);++ t.setOffers([elsewhere, { ...mine, health: { status: 'down', checked_at: new Date().toISOString() } }]);++ const waiting = (await t.dispatch.poll()).find((a) => a.job_id === later.json.job.id);++ assert.strictEqual(waiting.state, 'queued', 'a pinned job never lands on another machine');++++ // The pin is part of the request an idempotency key answers for.++ t.setOffers([elsewhere, mine]);++ const keyed = await asAda({ node: PINNED }, 'pin:c9c');++ assert.strictEqual(keyed.status, 201);++ assert.strictEqual((await asAda({ node: PINNED }, 'pin:c9c')).status, 200, 'the same request answers the first job');++ const changed = await asAda({}, 'pin:c9c');++ assert.strictEqual(changed.status, 409);++ assert.strictEqual(changed.json.code, 'run.idempotency.conflict');++++ await t.db.exec(`UPDATE run_jobs SET created_at = created_at - 100000000 WHERE state = 'queued'`);++ await t.dispatch.sweep();++ bot.devices.delete(PINNED);++ });++await check('a job cancelled while its frame is in flight is not placed, and Bot is told to stop it', async () => {t.setOffers([offer()]);const project = PROJECT();
@@ -15,7 +15,7 @@
@@ -15,7 +15,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.250.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -1147,9 +1147,9 @@
@@ -1147,9 +1147,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.223.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",
- "integrity": "sha512-FWcPL1951IeYcelp3dUiDRJEiFkTbsLMLth9RVUFfuLTuHGQKoQXA5HIu6H2e2BupPTyBH+AZD00LqmA6jDmdg==",
+ "version": "0.250.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.250.0",
+ "integrity": "sha512-fABKEYreTuvn5OquY9S5MGc90PQZTppO+evcZLkQyH7kWIdFZB0a5pI40ij7Y4JOaYWAFT8TPP1+FWaX/omynQ==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -26,7 +26,7 @@
@@ -26,7 +26,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.223.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.250.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.74.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -144,6 +144,14 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now
@@ -144,6 +144,14 @@ function v1Router({ config, db, apiAuth, stream, tickets, events, dispatch, now
const { project: projectId, requester, via } = apiAuth.requireProject(req, { delegated: true });
const { project: projectId, requester, via } = apiAuth.requireProject(req, { delegated: true });
const body = validCreate(req.body);
const body = validCreate(req.body);
checkLimits(body, config.limits);
checkLimits(body, config.limits);
+ // A pinned machine (Codes C9c): the job is refused now when it is not connected, never placed on another one.
+ if (body.node) {
+ let pin;
+ try { pin = await dispatch.pinStatus(body.node, body.class, requester, { fresh: true }); } catch (e) {
+ fail(503, 'run.dispatch.no_offers', `Run could not read which machines are connected (${e.code || e.message})`);
+ }
+ if (!pin.ok) fail(pin.code === 'run.job.node_offline' ? 409 : 422, pin.code, pin.detail);
+ }
const key = idempotencyKey(req, body);
const key = idempotencyKey(req, body);
const at = now();
const at = now();
const out = await db.tx(async (t) => {
const out = await db.tx(async (t) => {
@@ -81,6 +81,9 @@ function createDispatcher({ config = null, db = null, store = null, events = nul
@@ -81,6 +81,9 @@ function createDispatcher({ config = null, db = null, store = null, events = nul
return {
return {
enabled, note, place, cancel, poll, sweep,
enabled, note, place, cancel, poll, sweep,
+ // Codes C9c: may this job run pinned to that machine now? → { ok, offer? , code?, detail? }; throws when
+ // Network cannot be read (the caller answers 503, never a guessed offline).
+ pinStatus: (nodeId, runtimeClass, requester, opts) => placer.pinStatus(nodeId, runtimeClass, requester, opts),
start: () => { relay.start(); return poller.start(); },
start: () => { relay.start(); return poller.start(); },
stop: () => { poller.stop(); return relay.stop(); },
stop: () => { poller.stop(); return relay.stop(); },
status: () => ({
status: () => ({
@@ -83,9 +83,9 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -83,9 +83,9 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
: null;
: null;
let owned = { at: 0, offers: [] };
let owned = { at: 0, offers: [] };
/** Every live user-owned node offer that names its owner, from the internal list, cached like the public ones. */
/** Every live user-owned node offer that names its owner, from the internal list, cached like the public ones. */
- async function ownedOffers() {
+ async function ownedOffers({ maxAgeMs = offersTtlMs } = {}) {
if (!registryTokens) return [];
if (!registryTokens) return [];
- if (owned.at && now() - owned.at < offersTtlMs) return owned.offers;
+ if (owned.at && now() - owned.at < maxAgeMs) return owned.offers;
const href = `${config.network.internalUrl}/internal/resources?kind=node&trust=user-owned`;
const href = `${config.network.internalUrl}/internal/resources?kind=node&trust=user-owned`;
let res;
let res;
try {
try {
@@ -110,9 +110,9 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -110,9 +110,9 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
}
}
/** The published offers of one kind, cached briefly. A Network that does not answer is a retryable error. */
/** The published offers of one kind, cached briefly. A Network that does not answer is a retryable error. */
- async function offers(kind = 'node') {
+ async function offers(kind = 'node', { maxAgeMs = offersTtlMs } = {}) {
const cache = caches.get(kind) || { at: 0, offers: [], fetched: 0 };
const cache = caches.get(kind) || { at: 0, offers: [], fetched: 0 };
- if (cache.at && now() - cache.at < offersTtlMs) return cache.offers;
+ if (cache.at && now() - cache.at < maxAgeMs) return cache.offers;
const href = offersUrlOf(kind);
const href = offersUrlOf(kind);
let res;
let res;
try {
try {
@@ -139,12 +139,51 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -139,12 +139,51 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
return good;
return good;
}
}
+ /**
+ * Whether a job may be pinned to this machine now (Codes C9c): a machine of the requester's own (a user-owned
+ * offer names them), advertising the class, and up. Run refuses a pinned job when it is not — never places it
+ * somewhere else. A caller at submit asks with fresh: true (one read, not the poller's cached one); the
+ * placement loop asks from its cache, and a machine that went offline between the two simply keeps the job
+ * queued here until it is back or the job's ttl runs out.
+ *
+ * @returns {Promise<{ok: boolean, offer?: object, code?: string, detail?: string}>}
+ */
+ async function pinStatus(nodeId, runtimeClass, requester = null, { fresh = false } = {}) {
+ const want = workerCapability(runtimeClass);
+ if (!/^dev_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(nodeId || ''))) {
+ return { ok: false, code: 'run.job.node_offline', detail: 'node must be a machine id (dev_…)' };
+ }
+ const age = fresh ? 0 : offersTtlMs;
+ const offer = [...(await offers('node', { maxAgeMs: age })), ...(await ownedOffers({ maxAgeMs: age }))]
+ .find((o) => o && o.kind === 'node' && o.node_id === nodeId);
+ if (!offer) return { ok: false, code: 'run.job.node_offline', detail: `machine ${nodeId} is not connected to Run` };
+ if (offer.trust === 'user-owned') {
+ const person = personOf(requester);
+ if (!person || offer.owner !== person) return { ok: false, code: 'run.job.node_offline', detail: `machine ${nodeId} is not one of yours` };
+ }
+ if (!(offer.capabilities || []).includes(want)) return { ok: false, code: 'run.job.worker_failed', detail: `machine ${nodeId} does not run ${runtimeClass} jobs` };
+ if (!offer.health || offer.health.status !== 'up') return { ok: false, code: 'run.job.node_offline', detail: `machine ${nodeId} is not connected` };
+ return { ok: true, offer };
+ }
+
/**
/**
* The node to run this job on, or null when no offer is eligible. The requirements are the job's own
* The node to run this job on, or null when no offer is eligible. The requirements are the job's own
- * (the contract's default was filled in at submit), plus the class it runs.
+ * (the contract's default was filled in at submit), plus the class it runs. A job that names a machine
+ * (Codes C9c) is placed only there: the placer is not asked, and a machine that is not usable is an error,
+ * never a quieter placement somewhere else.
*/
*/
async function choose(row) {
async function choose(row) {
const want = workerCapability(row.class);
const want = workerCapability(row.class);
+ const pinned = typeof obj(row.job).node === 'string' ? obj(row.job).node : null;
+ if (pinned) {
+ const status = await pinStatus(pinned, row.class, row.requester);
+ if (!status.ok) throw new PlacementError(status.code, status.detail, { retryable: status.code === 'run.job.node_offline' });
+ const o = status.offer;
+ return {
+ node_id: o.node_id, provider: o.provider ?? null, region: o.region ?? null, kind: o.kind,
+ offer_id: o.offer_id, trust: o.trust, reasons: [`pinned to ${o.node_id}`],
+ };
+ }
const requirements = { ...obj(row.job).requirements || requirementsDefault(row.class) };
const requirements = { ...obj(row.job).requirements || requirementsDefault(row.class) };
const userOwnedNamed = Array.isArray(requirements.trust) && requirements.trust.includes('user-owned');
const userOwnedNamed = Array.isArray(requirements.trust) && requirements.trust.includes('user-owned');
// A user-owned offer is a person's own machine: eligible only for that person's own job, and only when the job
// A user-owned offer is a person's own machine: eligible only for that person's own job, and only when the job
@@ -186,6 +225,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -186,6 +225,7 @@ function createPlacement({ config, fetchImpl = globalThis.fetch, now = () => Dat
return {
return {
choose,
choose,
+ pinStatus,
offers,
offers,
workerCapability,
workerCapability,
status: () => {
status: () => {
@@ -66,6 +66,9 @@ function recordOf(row) {
@@ -66,6 +66,9 @@ function recordOf(row) {
ttl_ms: Number(row.ttl_ms),
ttl_ms: Number(row.ttl_ms),
egress: job.net ?? 'none',
egress: job.net ?? 'none',
requirements: job.requirements ?? requirementsDefault(row.class),
requirements: job.requirements ?? requirementsDefault(row.class),
+ // A pinned machine (Codes C9c): part of what an idempotent repeat is compared on, so a repeat that adds or
+ // drops the pin with the same key is a different request, never silently the first job.
+ node: job.node ?? null,
};
};
}
}
@@ -85,6 +88,8 @@ function normalize(body, { projectId, requester, via = null, idempotencyKey, now
@@ -85,6 +88,8 @@ function normalize(body, { projectId, requester, via = null, idempotencyKey, now
net: body.egress || 'none',
net: body.egress || 'none',
inputs: body.inputs || [],
inputs: body.inputs || [],
requirements: body.requirements || requirementsDefault(body.class),
requirements: body.requirements || requirementsDefault(body.class),
+ // The machine the caller pinned (Codes C9c): kept on the job record, never in the platform.job@1 frame.
+ ...(body.node ? { node: body.node } : {}),
};
};
// The platform.job@1 body the node will be sent (frameOf drops requirements — platform.job@1 has no
// The platform.job@1 body the node will be sent (frameOf drops requirements — platform.job@1 has no
// field for it), validated here once so a node can never be handed something off-contract.
// field for it), validated here once so a node can never be handed something off-contract.
@@ -432,6 +432,71 @@ async function main() {
@@ -432,6 +432,71 @@ async function main() {
bot.devices.delete(OTHER_DEVICE);
bot.devices.delete(OTHER_DEVICE);
});
});
+ await check('a pinned machine (Codes C9c): placed only there, refused when it is offline, never moved', async () => {
+ const ada = ids.newId('user');
+ const bo = ids.newId('user');
+ const P = PROJECT();
+ const PINNED = `dev_${ids.ulid()}`; // Bot's own id for a device (Run's placement node_id)
+ const ELSEWHERE = `dev_${ids.ulid()}`;
+ t.network.members.set(`${P} ${ada}`, { role: 'owner' });
+ t.network.members.set(`${P} ${bo}`, { role: 'owner' });
+ bot.devices.set(PINNED, { classes: ['function'] });
+ const mine = offer({ offer_id: 'off_mine', trust: 'user-owned', owner: ada, provider: 'self', node_id: PINNED });
+ const elsewhere = offer({ offer_id: 'off_other', node_id: ELSEWHERE });
+ t.setOffers([elsewhere, mine]);
+ const asAda = async (over, key = null) => t.call('POST', '/api/v1/jobs', {
+ body: t.request(over), key,
+ token: t.network.signService({ sub: 'svc:codes', cap: ['run.job.submit', 'run.job.read'] }),
+ headers: { 'X-OV-Subject': ada, 'X-OV-Project': P },
+ });
+
+ // Ada's own machine: created, kept on the job record, and placed there — not on the first-party offer.
+ const pinned = await asAda({ node: PINNED });
+ assert.strictEqual(pinned.status, 201, pinned.text);
+ const stored = await t.db.maybe('SELECT job FROM run_jobs WHERE id = $1', [pinned.json.job.id]);
+ assert.strictEqual(stored.job.node, PINNED);
+ const placed = (await t.dispatch.poll()).find((a) => a.job_id === pinned.json.job.id);
+ assert.strictEqual(placed.state, 'placed');
+ assert.strictEqual(placed.node_id, PINNED);
+ assert.strictEqual(bot.jobs.get(pinned.json.job.id).job.node, undefined, 'node is Run\'s, never a platform.job@1 field');
+
+ // Offline: refused now, and nothing is stored.
+ const before = Number((await t.db.maybe('SELECT count(*) AS n FROM run_jobs')).n);
+ t.setOffers([elsewhere]);
+ const offline = await asAda({ node: PINNED });
+ assert.strictEqual(offline.status, 409, offline.text);
+ assert.strictEqual(offline.json.code, 'run.job.node_offline');
+ assert.strictEqual(Number((await t.db.maybe('SELECT count(*) AS n FROM run_jobs')).n), before, 'a refused pin stores no job');
+
+ // Someone else's machine is never pinnable, and a class it does not run is a plain refusal.
+ t.setOffers([elsewhere, offer({ offer_id: 'off_mine', trust: 'user-owned', owner: bo, node_id: PINNED })]);
+ assert.strictEqual((await asAda({ node: PINNED })).json.code, 'run.job.node_offline');
+ t.setOffers([elsewhere, mine]);
+ const wrongClass = await asAda({ class: 'code', artifact: { name: 'openvibe-codes', version: '1.0.0' }, node: PINNED });
+ assert.strictEqual(wrongClass.status, 422, wrongClass.text);
+ assert.strictEqual(wrongClass.json.code, 'run.job.worker_failed');
+
+ // Pinned, then the machine goes down before placement: the job waits, and is never moved to the other one.
+ const later = await asAda({ node: PINNED });
+ assert.strictEqual(later.status, 201, later.text);
+ t.setOffers([elsewhere, { ...mine, health: { status: 'down', checked_at: new Date().toISOString() } }]);
+ const waiting = (await t.dispatch.poll()).find((a) => a.job_id === later.json.job.id);
+ assert.strictEqual(waiting.state, 'queued', 'a pinned job never lands on another machine');
+
+ // The pin is part of the request an idempotency key answers for.
+ t.setOffers([elsewhere, mine]);
+ const keyed = await asAda({ node: PINNED }, 'pin:c9c');
+ assert.strictEqual(keyed.status, 201);
+ assert.strictEqual((await asAda({ node: PINNED }, 'pin:c9c')).status, 200, 'the same request answers the first job');
+ const changed = await asAda({}, 'pin:c9c');
+ assert.strictEqual(changed.status, 409);
+ assert.strictEqual(changed.json.code, 'run.idempotency.conflict');
+
+ await t.db.exec(`UPDATE run_jobs SET created_at = created_at - 100000000 WHERE state = 'queued'`);
+ await t.dispatch.sweep();
+ bot.devices.delete(PINNED);
+ });
+
await check('a job cancelled while its frame is in flight is not placed, and Bot is told to stop it', async () => {
await check('a job cancelled while its frame is in flight is not placed, and Bot is told to stop it', async () => {
t.setOffers([offer()]);
t.setOffers([offer()]);
const project = PROJECT();
const project = PROJECT();