import test from 'node:test'; import assert from 'node:assert/strict'; import { EventEmitter } from 'node:events'; import { PassThrough } from 'node:stream'; import { createJobManager, recoverOrphans } from '../src/jobs/manager.mjs'; import { openDb } from '../src/db.mjs'; test('close cancela jobs, aguarda exit e mata ao exceder prazo', async () => { const children = []; const spawn = () => { const child = new EventEmitter(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.stdin = { write: (value) => { child.cancel = value; } }; child.kill = () => { child.killed = true; }; children.push(child); return child; }; const jobs = createJobManager({ spawn, killGraceMs: 5 }); jobs.startJob({ type: 'teste', script: 'fake.mjs' }); await Promise.race([ jobs.close(), new Promise((resolve, reject) => setTimeout(() => reject(new Error('close nao respeitou prazo')), 50)) ]); assert.equal(children[0].cancel, 'cancel\n'); assert.equal(children[0].killed, true); assert.equal(jobs.activeJobs().length, 0); }); test('recoverOrphans recupera heartbeat stale mesmo com PID reutilizado e preserva lease recente', () => { const db = openDb(':memory:'); db.prepare("INSERT INTO capture_runs (params_json, status, pid, heartbeat_at) VALUES ('{}', 'running', ?, '2000-01-01 00:00:00')").run(process.pid); db.prepare("INSERT INTO extractions (uf, status, pid, heartbeat_at) VALUES ('BA', 'pending', 999, datetime('now'))").run(); db.prepare("INSERT INTO exports (tipo, status, pid, heartbeat_at) VALUES ('xlsx', 'running', 999, '2000-01-01 00:00:00')").run(); recoverOrphans(db, () => {}, { processAlive: (pid) => pid === process.pid }); assert.equal(db.prepare('SELECT status FROM capture_runs').get().status, 'error'); assert.equal(db.prepare('SELECT status FROM extractions').get().status, 'pending'); const recovered = db.prepare('SELECT status, error, finished_at FROM exports').get(); assert.equal(recovered.status, 'error'); assert.match(recovered.error, /interrompido/); assert.ok(recovered.finished_at); db.close(); }); test('manager aguarda close para publicar done e converte close nao zero apos done em erro', async () => { const children = []; const spawn = () => { const child = new EventEmitter(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.stdin = { write() {} }; child.kill = () => {}; children.push(child); return child; }; const jobs = createJobManager({ spawn }); const firstId = jobs.startJob({ type: 'teste', script: 'fake' }); const firstEvents = []; jobs.subscribe(firstId, (event) => firstEvents.push(event)); children[0].stdout.write('{"t":"done","summary":{"total":1}}\n'); await new Promise((resolve) => setImmediate(resolve)); assert.equal(jobs.getJob(firstId).status, 'running'); assert.equal(firstEvents.some((event) => event.t === 'done'), false); children[0].emit('exit', 0, null); assert.equal(jobs.getJob(firstId).status, 'running'); children[0].emit('close', 0, null); assert.equal(jobs.getJob(firstId).status, 'done'); assert.equal(firstEvents.at(-1).t, 'done'); const secondId = jobs.startJob({ type: 'teste', script: 'fake' }); const secondEvents = []; jobs.subscribe(secondId, (event) => secondEvents.push(event)); children[1].stdout.write('{"t":"done","summary":{}}\n'); await new Promise((resolve) => setImmediate(resolve)); children[1].emit('exit', 1, null); children[1].emit('close', 1, null); assert.equal(jobs.getJob(secondId).status, 'error'); assert.equal(secondEvents.at(-1).t, 'error'); }); test('exit nao zero corrige estado done persistido prematuramente pelo filho', async () => { const db = openDb(':memory:'); const runId = Number(db.prepare("INSERT INTO capture_runs (params_json, status) VALUES ('{}', 'running')").run().lastInsertRowid); const child = new EventEmitter(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.stdin = { write() {} }; child.kill = () => {}; const jobs = createJobManager({ spawn: () => child, db }); jobs.startJob({ type: 'captura', script: 'fake', meta: { runId } }); db.prepare("UPDATE capture_runs SET status = 'done' WHERE id = ?").run(runId); child.stdout.write('{"t":"done","summary":{}}\n'); await new Promise((resolve) => setImmediate(resolve)); child.emit('exit', 1, null); child.emit('close', 1, null); assert.equal(db.prepare('SELECT status FROM capture_runs WHERE id = ?').get(runId).status, 'error'); db.close(); }); test('manager persiste erro de filho inesperado e cancelamento forcado no DB', async () => { const db = openDb(':memory:'); const captureId = Number(db.prepare("INSERT INTO capture_runs (params_json, status) VALUES ('{}', 'pending')").run().lastInsertRowid); const extractionId = Number(db.prepare("INSERT INTO extractions (uf, status) VALUES ('BA', 'pending')").run().lastInsertRowid); const children = []; const spawn = () => { const child = new EventEmitter(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.stdin = { write() {} }; child.kill = () => { child.emit('exit', null, 'SIGTERM'); child.emit('close', null, 'SIGTERM'); }; children.push(child); return child; }; const jobs = createJobManager({ spawn, db, killGraceMs: 1 }); jobs.startJob({ type: 'captura', script: 'fake', meta: { runId: captureId } }); children[0].stderr.end('boom'); children[0].emit('exit', 1, null); children[0].emit('close', 1, null); assert.equal(db.prepare('SELECT status FROM capture_runs WHERE id = ?').get(captureId).status, 'error'); const jobId = jobs.startJob({ type: 'extracao', script: 'fake', meta: { extractionId } }); assert.equal(jobs.cancelJob(jobId), true); await new Promise((resolve) => setTimeout(resolve, 5)); assert.equal(db.prepare('SELECT status FROM extractions WHERE id = ?').get(extractionId).status, 'cancelled'); db.close(); }); test('manager mantem cancelado quando filho relata erro depois da intencao de cancelar', async () => { const db = openDb(':memory:'); const extractionId = Number(db.prepare("INSERT INTO extractions (uf, status) VALUES ('BA', 'running')").run().lastInsertRowid); const child = new EventEmitter(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.stdin = { write() {} }; child.kill = () => {}; const jobs = createJobManager({ spawn: () => child, db, killGraceMs: 100 }); const jobId = jobs.startJob({ type: 'extracao', script: 'fake', meta: { extractionId } }); jobs.cancelJob(jobId); child.stdout.write('{"t":"error","message":"stream encerrado"}\n'); await new Promise((resolve) => setImmediate(resolve)); assert.equal(db.prepare('SELECT status FROM extractions WHERE id = ?').get(extractionId).status, 'cancelled'); child.emit('exit', 1, null); child.emit('close', 1, null); db.close(); }); test('linha done drenada entre exit e close resulta em done', async () => { const child = new EventEmitter(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.stdin = { write() {} }; child.kill = () => {}; const jobs = createJobManager({ spawn: () => child }); const jobId = jobs.startJob({ type: 'teste', script: 'fake' }); child.emit('exit', 0, null); child.stdout.write('{"t":"done","summary":{"late":true}}\n'); await new Promise((resolve) => setImmediate(resolve)); assert.equal(jobs.getJob(jobId).status, 'running'); child.emit('close', 0, null); assert.equal(jobs.getJob(jobId).status, 'done'); assert.equal(jobs.getJob(jobId).lastEvent.summary.late, true); }); test('manager trata erro do stream stdout como falha terminal persistida', async () => { const db = openDb(':memory:'); const exportId = Number(db.prepare("INSERT INTO exports (tipo, status) VALUES ('excel', 'running')").run().lastInsertRowid); const child = new EventEmitter(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.stdin = { write() {} }; child.kill = () => { child.killed = true; }; const jobs = createJobManager({ spawn: () => child, db }); jobs.startJob({ type: 'exportacao', script: 'fake', meta: { exportId } }); child.stdout.emit('error', new Error('pipe quebrado')); assert.equal(db.prepare('SELECT status FROM exports WHERE id = ?').get(exportId).status, 'error'); assert.equal(child.killed, true); db.close(); });