openvibers/openvibe.mediahub Commits
Video: a running AI edit is read by its run id (a second post carried a new signed URL, which AI refused as an idempotency conflict) (#72)
Changes
@@@ -9,8 +9,9 @@* overview a short description, labelled as written by AI, which the owner may take as the video's description** Media gives AI a signed URL to read (private videos too); AI keeps nothing of the video but the run. One run per-- * attempt, keyed (media-hub.video.analyze:<id>:<attempt>): a run still going after a minute is joined again by its key-- * on the next sweep, never started twice. A failure waits and retries (up to 4 attempts).++ * attempt, keyed (media-hub.video.analyze:<id>:<attempt>): a run still going after a minute is read again by its id on++ * the next sweeps (GET /api/v1/runs/:id; posting again would carry a new signed URL, which AI refuses under the same++ * key), never started twice. A failure waits and retries (up to 4 attempts).** sweep() every few minutes: queue ready uploads that have no AI row, then work the due rows (2 at a time)* forUpload(row) the AI row of an upload, parsed (or null)@@@ -54,6 +55,17 @@ function createAiEdit({ s, media, config, fetchImpl = globalThis.fetch, log = coreturn body && body.run;}++ /** A run this service started, as it is now. */++ async function read(runId) {++ const token = await tokens.getToken();++ const res = await fetchImpl(`${url}/api/v1/runs/${encodeURIComponent(runId)}`, {++ headers: { Accept: 'application/json', Authorization: `Bearer ${token}` }, signal: AbortSignal.timeout(15_000),++ });++ const body = await res.json().catch(() => null);++ if (!res.ok) throw new Error(`OpenVibe.AI answered ${res.status}${body && body.code ? ` (${body.code})` : ''}`);++ return body && body.run;++ }++const later = (minutes) => new Date(s.now() + minutes * 60_000).toISOString();const save = (id, patch) => {const cols = Object.keys(patch);@@@ -63,13 +75,18 @@ function createAiEdit({ s, media, config, fetchImpl = globalThis.fetch, log = coasync function work(ai) {const upload = await s.db.maybe('SELECT * FROM video_uploads WHERE id = $1', [ai.upload_id]);if (!upload || upload.state === 'removed' || upload.status !== 'ready') return save(ai.upload_id, { status: 'failed', error: 'The video is gone.', next_at: null });-- // A run still going is joined by the same key; a new attempt gets a new key.++ // A run still going is read again by its id; a new attempt starts a run under a new key.++ const joining = ai.status === 'running' && Boolean(ai.run_id);const attempt = ai.status === 'running' ? ai.attempts : ai.attempts + 1;if (ai.status !== 'running') await save(ai.upload_id, { status: 'running', attempts: attempt, next_at: later(15) });try {-- const signed = await media.vodSignedUrl(upload.media_vod_id, 6 * 3600);-- if (!signed.ok || !signed.data || !signed.data.url) throw new Error(`Media gave no readable URL (${signed.code || signed.status})`);-- const r = await run({ media_url: signed.data.url, language: 'auto', allow_paid: false }, `media-hub.video.analyze:${upload.id}:${attempt}`, { service: 'media-hub', type: 'video', id: upload.id });++ let r;++ if (joining) r = await read(ai.run_id);++ else {++ const signed = await media.vodSignedUrl(upload.media_vod_id, 6 * 3600);++ if (!signed.ok || !signed.data || !signed.data.url) throw new Error(`Media gave no readable URL (${signed.code || signed.status})`);++ r = await run({ media_url: signed.data.url, language: 'auto', allow_paid: false }, `media-hub.video.analyze:${upload.id}:${attempt}`, { service: 'media-hub', type: 'video', id: upload.id });++ }// Still going after the minute's wait: look again on the next sweep or two (an analysis takes a few minutes).if (r && (r.status === 'queued' || r.status === 'running')) return save(ai.upload_id, { run_id: r.id || null, next_at: later(2) });const out = r && r.status === 'succeeded' && r.output;@@@ -61,10 +61,10 @@ const OUTPUT = {const runs = [];let mode = 'running';const ai = await startStub((req, body) => {-- runs.push({ path: req.url, body, auth: req.headers.authorization });++ runs.push({ method: req.method, path: req.url, body, auth: req.headers.authorization });if (mode === 'fail') return { status: 503, body: { code: 'ai.unavailable' } };if (mode === 'running') return { status: 202, body: { run: { id: 'run_1', status: 'running' } } };-- return { status: 201, body: { run: { id: 'run_1', status: 'succeeded', output: OUTPUT } } };++ return { status: req.method === 'GET' ? 200 : 201, body: { run: { id: 'run_1', status: 'succeeded', output: OUTPUT } } };});const live = await startStub(() => ({ body: { vods: [], clips: [], items: [], hasMore: false } }));const t = await boot({ env: { MEDIAHUB_VIDEO_URL: 'https://openvibe.video', MEDIAHUB_LIVE_URL: live.url, MEDIAHUB_LIVE_PUBLIC_URL: 'https://openvibe.live', MEDIAHUB_CHAT_URL: live.url, OV_AI_INTERNAL_URL: ai.url } });@@@ -74,7 +74,7 @@ const OUTPUT = {const samCookie = t.signIn(sam);let id;-- await check('a finished private upload is analysed through a signed URL, never a paid model; a run still going is joined by its key', async () => {++ await check('a finished private upload is analysed through a signed URL, never a paid model; a run still going is read again by its id', async () => {const start = await vreq(t.base, '/api/v1/videos', { method: 'POST', json: { title: 'Coast', size: 10, content_type: 'video/mp4', name: 'c.mp4', visibility: 'private' }, cookie: kimCookie });id = start.json().upload.id;await vreq(t.base, `/api/v1/videos/${id}/parts/1`, { method: 'PUT', body: Buffer.alloc(10, 1), cookie: kimCookie });@@@ -91,11 +91,11 @@ const OUTPUT = {assert.ok(String(r.auth).startsWith('Bearer '));const row = await t.ctx.videoAi.forUpload({ id });assert.strictEqual(row.status, 'running');-- // Due again later: join the same run by the same key.++ // Due again later: read the same run by its id (posting again would carry a new signed URL under the same key).await t.ctx.s.db.query('UPDATE video_upload_ai SET next_at = $1 WHERE upload_id = $2', ['2000-01-01T00:00:00.000Z', id]);mode = 'done';await t.ctx.videoAi.sweep();-- assert.strictEqual(runs[1].body.idempotency_key, `media-hub.video.analyze:${id}:1`, 'the same key: joined, not a second analysis');++ assert.deepStrictEqual([runs[1].method, runs[1].path], ['GET', '/api/v1/runs/run_1'], 'read, not a second analysis');assert.strictEqual((await t.ctx.videoAi.forUpload({ id })).status, 'done');});
@@ -9,8 +9,9 @@
@@ -9,8 +9,9 @@
* overview a short description, labelled as written by AI, which the owner may take as the video's description
* overview a short description, labelled as written by AI, which the owner may take as the video's description
*
*
* Media gives AI a signed URL to read (private videos too); AI keeps nothing of the video but the run. One run per
* Media gives AI a signed URL to read (private videos too); AI keeps nothing of the video but the run. One run per
- * attempt, keyed (media-hub.video.analyze:<id>:<attempt>): a run still going after a minute is joined again by its key
- * on the next sweep, never started twice. A failure waits and retries (up to 4 attempts).
+ * attempt, keyed (media-hub.video.analyze:<id>:<attempt>): a run still going after a minute is read again by its id on
+ * the next sweeps (GET /api/v1/runs/:id; posting again would carry a new signed URL, which AI refuses under the same
+ * key), never started twice. A failure waits and retries (up to 4 attempts).
*
*
* sweep() every few minutes: queue ready uploads that have no AI row, then work the due rows (2 at a time)
* sweep() every few minutes: queue ready uploads that have no AI row, then work the due rows (2 at a time)
* forUpload(row) the AI row of an upload, parsed (or null)
* forUpload(row) the AI row of an upload, parsed (or null)
@@ -54,6 +55,17 @@ function createAiEdit({ s, media, config, fetchImpl = globalThis.fetch, log = co
@@ -54,6 +55,17 @@ function createAiEdit({ s, media, config, fetchImpl = globalThis.fetch, log = co
return body && body.run;
return body && body.run;
}
}
+ /** A run this service started, as it is now. */
+ async function read(runId) {
+ const token = await tokens.getToken();
+ const res = await fetchImpl(`${url}/api/v1/runs/${encodeURIComponent(runId)}`, {
+ headers: { Accept: 'application/json', Authorization: `Bearer ${token}` }, signal: AbortSignal.timeout(15_000),
+ });
+ const body = await res.json().catch(() => null);
+ if (!res.ok) throw new Error(`OpenVibe.AI answered ${res.status}${body && body.code ? ` (${body.code})` : ''}`);
+ return body && body.run;
+ }
+
const later = (minutes) => new Date(s.now() + minutes * 60_000).toISOString();
const later = (minutes) => new Date(s.now() + minutes * 60_000).toISOString();
const save = (id, patch) => {
const save = (id, patch) => {
const cols = Object.keys(patch);
const cols = Object.keys(patch);
@@ -63,13 +75,18 @@ function createAiEdit({ s, media, config, fetchImpl = globalThis.fetch, log = co
@@ -63,13 +75,18 @@ function createAiEdit({ s, media, config, fetchImpl = globalThis.fetch, log = co
async function work(ai) {
async function work(ai) {
const upload = await s.db.maybe('SELECT * FROM video_uploads WHERE id = $1', [ai.upload_id]);
const upload = await s.db.maybe('SELECT * FROM video_uploads WHERE id = $1', [ai.upload_id]);
if (!upload || upload.state === 'removed' || upload.status !== 'ready') return save(ai.upload_id, { status: 'failed', error: 'The video is gone.', next_at: null });
if (!upload || upload.state === 'removed' || upload.status !== 'ready') return save(ai.upload_id, { status: 'failed', error: 'The video is gone.', next_at: null });
- // A run still going is joined by the same key; a new attempt gets a new key.
+ // A run still going is read again by its id; a new attempt starts a run under a new key.
+ const joining = ai.status === 'running' && Boolean(ai.run_id);
const attempt = ai.status === 'running' ? ai.attempts : ai.attempts + 1;
const attempt = ai.status === 'running' ? ai.attempts : ai.attempts + 1;
if (ai.status !== 'running') await save(ai.upload_id, { status: 'running', attempts: attempt, next_at: later(15) });
if (ai.status !== 'running') await save(ai.upload_id, { status: 'running', attempts: attempt, next_at: later(15) });
try {
try {
- const signed = await media.vodSignedUrl(upload.media_vod_id, 6 * 3600);
- if (!signed.ok || !signed.data || !signed.data.url) throw new Error(`Media gave no readable URL (${signed.code || signed.status})`);
- const r = await run({ media_url: signed.data.url, language: 'auto', allow_paid: false }, `media-hub.video.analyze:${upload.id}:${attempt}`, { service: 'media-hub', type: 'video', id: upload.id });
+ let r;
+ if (joining) r = await read(ai.run_id);
+ else {
+ const signed = await media.vodSignedUrl(upload.media_vod_id, 6 * 3600);
+ if (!signed.ok || !signed.data || !signed.data.url) throw new Error(`Media gave no readable URL (${signed.code || signed.status})`);
+ r = await run({ media_url: signed.data.url, language: 'auto', allow_paid: false }, `media-hub.video.analyze:${upload.id}:${attempt}`, { service: 'media-hub', type: 'video', id: upload.id });
+ }
// Still going after the minute's wait: look again on the next sweep or two (an analysis takes a few minutes).
// Still going after the minute's wait: look again on the next sweep or two (an analysis takes a few minutes).
if (r && (r.status === 'queued' || r.status === 'running')) return save(ai.upload_id, { run_id: r.id || null, next_at: later(2) });
if (r && (r.status === 'queued' || r.status === 'running')) return save(ai.upload_id, { run_id: r.id || null, next_at: later(2) });
const out = r && r.status === 'succeeded' && r.output;
const out = r && r.status === 'succeeded' && r.output;
@@ -61,10 +61,10 @@ const OUTPUT = {
@@ -61,10 +61,10 @@ const OUTPUT = {
const runs = [];
const runs = [];
let mode = 'running';
let mode = 'running';
const ai = await startStub((req, body) => {
const ai = await startStub((req, body) => {
- runs.push({ path: req.url, body, auth: req.headers.authorization });
+ runs.push({ method: req.method, path: req.url, body, auth: req.headers.authorization });
if (mode === 'fail') return { status: 503, body: { code: 'ai.unavailable' } };
if (mode === 'fail') return { status: 503, body: { code: 'ai.unavailable' } };
if (mode === 'running') return { status: 202, body: { run: { id: 'run_1', status: 'running' } } };
if (mode === 'running') return { status: 202, body: { run: { id: 'run_1', status: 'running' } } };
- return { status: 201, body: { run: { id: 'run_1', status: 'succeeded', output: OUTPUT } } };
+ return { status: req.method === 'GET' ? 200 : 201, body: { run: { id: 'run_1', status: 'succeeded', output: OUTPUT } } };
});
});
const live = await startStub(() => ({ body: { vods: [], clips: [], items: [], hasMore: false } }));
const live = await startStub(() => ({ body: { vods: [], clips: [], items: [], hasMore: false } }));
const t = await boot({ env: { MEDIAHUB_VIDEO_URL: 'https://openvibe.video', MEDIAHUB_LIVE_URL: live.url, MEDIAHUB_LIVE_PUBLIC_URL: 'https://openvibe.live', MEDIAHUB_CHAT_URL: live.url, OV_AI_INTERNAL_URL: ai.url } });
const t = await boot({ env: { MEDIAHUB_VIDEO_URL: 'https://openvibe.video', MEDIAHUB_LIVE_URL: live.url, MEDIAHUB_LIVE_PUBLIC_URL: 'https://openvibe.live', MEDIAHUB_CHAT_URL: live.url, OV_AI_INTERNAL_URL: ai.url } });
@@ -74,7 +74,7 @@ const OUTPUT = {
@@ -74,7 +74,7 @@ const OUTPUT = {
const samCookie = t.signIn(sam);
const samCookie = t.signIn(sam);
let id;
let id;
- await check('a finished private upload is analysed through a signed URL, never a paid model; a run still going is joined by its key', async () => {
+ await check('a finished private upload is analysed through a signed URL, never a paid model; a run still going is read again by its id', async () => {
const start = await vreq(t.base, '/api/v1/videos', { method: 'POST', json: { title: 'Coast', size: 10, content_type: 'video/mp4', name: 'c.mp4', visibility: 'private' }, cookie: kimCookie });
const start = await vreq(t.base, '/api/v1/videos', { method: 'POST', json: { title: 'Coast', size: 10, content_type: 'video/mp4', name: 'c.mp4', visibility: 'private' }, cookie: kimCookie });
id = start.json().upload.id;
id = start.json().upload.id;
await vreq(t.base, `/api/v1/videos/${id}/parts/1`, { method: 'PUT', body: Buffer.alloc(10, 1), cookie: kimCookie });
await vreq(t.base, `/api/v1/videos/${id}/parts/1`, { method: 'PUT', body: Buffer.alloc(10, 1), cookie: kimCookie });
@@ -91,11 +91,11 @@ const OUTPUT = {
@@ -91,11 +91,11 @@ const OUTPUT = {
assert.ok(String(r.auth).startsWith('Bearer '));
assert.ok(String(r.auth).startsWith('Bearer '));
const row = await t.ctx.videoAi.forUpload({ id });
const row = await t.ctx.videoAi.forUpload({ id });
assert.strictEqual(row.status, 'running');
assert.strictEqual(row.status, 'running');
- // Due again later: join the same run by the same key.
+ // Due again later: read the same run by its id (posting again would carry a new signed URL under the same key).
await t.ctx.s.db.query('UPDATE video_upload_ai SET next_at = $1 WHERE upload_id = $2', ['2000-01-01T00:00:00.000Z', id]);
await t.ctx.s.db.query('UPDATE video_upload_ai SET next_at = $1 WHERE upload_id = $2', ['2000-01-01T00:00:00.000Z', id]);
mode = 'done';
mode = 'done';
await t.ctx.videoAi.sweep();
await t.ctx.videoAi.sweep();
- assert.strictEqual(runs[1].body.idempotency_key, `media-hub.video.analyze:${id}:1`, 'the same key: joined, not a second analysis');
+ assert.deepStrictEqual([runs[1].method, runs[1].path], ['GET', '/api/v1/runs/run_1'], 'read, not a second analysis');
assert.strictEqual((await t.ctx.videoAi.forUpload({ id })).status, 'done');
assert.strictEqual((await t.ctx.videoAi.forUpload({ id })).status, 'done');
});
});