Corrige quatro defeitos de runtime do worker e dos scripts operacionais
Os quatro apareceram ao exercitar o layout de runtime da imagem — apenas node_modules de produção, src, scripts e migrations — e nenhum aparecia na suíte de testes nem no build. Os dois primeiros formavam uma armadilha mútua. 1. O CLI do `tsx` cria processo filho (PIDs medidos: 20840 contra 21096). O SIGTERM do `docker stop` pararia no pai, o worker nunca rodaria o handler, o lock esperaria o TTL de 120s e o `stop_grace_period` só atrasaria um SIGKILL. Passa a ser `node --import tsx`, mesmo processo. 2. Top-level await nos scripts falha com `--import tsx`, porque sem `"type": "module"` o esbuild gera CJS. `migrate`, `worker`, `seed-staging`, `flag`, `cleanup` e `check-bucket` não executariam: a migração não rodaria e o worker não subiria. `runScript`/`runWithDb` resolvem sem mudar o module type do projeto. 3. `import "server-only"` em `worker-loop.ts`: o pacote lança fora do bundler do Next, e o worker morria no start. 4. `@react-pdf/renderer` carregado no import de `process-job.ts` falha com ERR_PACKAGE_PATH_NOT_EXPORTED ao resolver a hifenização fora do bundler, derrubando worker e cleanup. O PDF sai sob demanda no download, então o import passou a ser tardio, dentro do ramo que o usa. Além disso: - O loop do worker saiu do script para `worker-loop.ts`, testável sem subir processo. O script cuida só de sinais, cadência e bootstrap. - Desligamento gracioso espera o tick corrente para o `finally` liberar o lock. Um job que ainda não começou fica `queued`: nada pela metade. - Heartbeat por tick, lido pelo healthcheck do contêiner. Processo vivo mas travado passa em qualquer verificação de PID. - O worker recusa iniciar com migration pendente, e relê o contexto do dono a cada job: papel e feature flag podem ter mudado desde o enfileiramento. - `verify-config`, `check-bucket`, `seed-staging` e `smoke-staging` novos. O smoke tem fase autenticada que só roda com credencial fornecida pela operação e, sem ela, reporta PENDENTE em vez de aprovado. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
1 parent
d43f370359
commit
ca8a1112f0
13 files changed
+835
-126
No files matched your search
@@ -7,7 +7,6 @@ import { canUseSpreadsheetAdapter } from "@/domain/spreadsheet-adapter/permissio
|
||||
import { transformRows } from "@/domain/spreadsheet-adapter/transform";
|
||||
import type { JobProgress, OutputFormat, RequestContext } from "@/domain/spreadsheet-adapter/types";
|
||||
import { buildCsv, buildXlsx } from "@/infra/export/tabular";
|
||||
import { buildPdf } from "@/infra/export/pdf";
|
||||
import type { Query } from "@/infra/db/types";
|
||||
import { lockKey } from "@/infra/redis/redis";
|
||||
import { sanitizeAuditMetadata } from "@/infra/observability/scrub";
|
||||
@@ -282,6 +281,10 @@ export async function downloadOutput(deps: AppDeps, ctx: RequestContext, jobId:
|
||||
const csv = (await deps.storage.get(file.output_csv_key)).toString("utf8");
|
||||
const grid = parseCsvText(csv);
|
||||
const headers = grid[0] ?? [];
|
||||
// Import tardio: o worker não gera PDF (o PDF sai sob demanda no download),
|
||||
// e carregar @react-pdf/renderer fora do bundler do Next falha ao resolver
|
||||
// o subpath de hifenização. Sem isto o processo do worker nem subia.
|
||||
const { buildPdf } = await import("@/infra/export/pdf");
|
||||
const body = await buildPdf(file.model_name, headers, grid.slice(1));
|
||||
return { filename, contentType: CONTENT_TYPE.pdf, body: body.body };
|
||||
}
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
import { asJson } from "./db-utils";
|
||||
import { cleanupExpired, listQueuedJobIds, readJobForWorker, recoverStaleJobs, runProcess } from "./process-job";
|
||||
import type { AppDeps } from "./deps";
|
||||
import type { JobProgress, RequestContext, Role } from "@/domain/spreadsheet-adapter/types";
|
||||
|
||||
/**
|
||||
* Loop do worker, fora do script de entrada para poder ser testado sem subir
|
||||
* um processo. O script `scripts/worker.ts` só cuida de sinais e de bootstrap.
|
||||
*
|
||||
* Sem `import "server-only"` de propósito: este módulo roda também no processo
|
||||
* do worker, que é Node puro. O pacote `server-only` lança quando é carregado
|
||||
* fora do bundler do Next, e isso derrubava o worker no start.
|
||||
*/
|
||||
|
||||
export type WorkerOptions = {
|
||||
/** Máximo de ids retirados da fila do Redis por tick. */
|
||||
batch?: number;
|
||||
/** Caminho do arquivo de heartbeat lido pelo health check do contêiner. */
|
||||
heartbeatPath?: string | null;
|
||||
/** Injetável para teste. */
|
||||
writeHeartbeat?: (path: string) => Promise<void>;
|
||||
log?: (message: string) => void;
|
||||
onError?: (scope: string, error: unknown) => void;
|
||||
};
|
||||
|
||||
export type TickResult = {
|
||||
/** Jobs que o worker tentou processar neste tick. */
|
||||
processed: string[];
|
||||
/** Temporários expirados removidos neste tick. */
|
||||
cleaned: number;
|
||||
recovered: boolean;
|
||||
};
|
||||
|
||||
const DEFAULT_BATCH = 20;
|
||||
|
||||
async function defaultWriteHeartbeat(target: string): Promise<void> {
|
||||
const { writeFile } = await import("node:fs/promises");
|
||||
await writeFile(target, new Date().toISOString(), "utf8");
|
||||
}
|
||||
|
||||
/**
|
||||
* Contexto do dono do job, relido do banco a cada execução. Nunca vem do
|
||||
* payload do job: papel, flag do módulo e configuração podem ter mudado entre
|
||||
* o enfileiramento e o processamento.
|
||||
*/
|
||||
async function loadContext(deps: AppDeps, userId: string, workspaceId: string): Promise<RequestContext | null> {
|
||||
const rows = await deps.db.withService((query) =>
|
||||
query.query<{
|
||||
display_name: string;
|
||||
is_superadmin: boolean;
|
||||
role: Role;
|
||||
workspace_name: string;
|
||||
module_enabled: boolean;
|
||||
allow_upload: boolean;
|
||||
}>(
|
||||
`select p.display_name, p.is_superadmin, w.name as workspace_name, m.role,
|
||||
coalesce(f.enabled, false) as module_enabled,
|
||||
coalesce(s.allow_non_admin_model_upload, false) as allow_upload
|
||||
from public.profiles p
|
||||
join public.workspace_members m on m.user_id = p.id and m.workspace_id = $2
|
||||
join public.workspaces w on w.id = $2
|
||||
left join public.workspace_module_flags f on f.workspace_id = w.id and f.module_key = 'spreadsheet_adapter'
|
||||
left join public.spreadsheet_adapter_settings s on s.workspace_id = w.id
|
||||
where p.id = $1`,
|
||||
[userId, workspaceId],
|
||||
),
|
||||
);
|
||||
const found = rows[0];
|
||||
if (!found) return null;
|
||||
return {
|
||||
userId,
|
||||
displayName: found.display_name,
|
||||
isSuperadmin: found.is_superadmin,
|
||||
workspaceId,
|
||||
workspaceName: found.workspace_name,
|
||||
role: found.role,
|
||||
moduleEnabled: found.module_enabled,
|
||||
allowNonAdminModelUpload: found.allow_upload,
|
||||
};
|
||||
}
|
||||
|
||||
export function createWorker(deps: AppDeps, options: WorkerOptions = {}) {
|
||||
const batch = options.batch ?? DEFAULT_BATCH;
|
||||
const heartbeatPath = options.heartbeatPath ?? null;
|
||||
const writeHeartbeat = options.writeHeartbeat ?? defaultWriteHeartbeat;
|
||||
const onError = options.onError ?? (() => undefined);
|
||||
|
||||
let running = true;
|
||||
let inTick = false;
|
||||
|
||||
async function tick(): Promise<TickResult> {
|
||||
if (!running || inTick) return { processed: [], cleaned: 0, recovered: false };
|
||||
inTick = true;
|
||||
const processed: string[] = [];
|
||||
let cleaned = 0;
|
||||
let recovered = false;
|
||||
try {
|
||||
recovered = await recoverStaleJobs(deps)
|
||||
.then(() => true)
|
||||
.catch((error) => {
|
||||
onError("recover", error);
|
||||
return false;
|
||||
});
|
||||
cleaned = await cleanupExpired(deps).catch((error) => {
|
||||
onError("cleanup", error);
|
||||
return 0;
|
||||
});
|
||||
|
||||
// Duas origens para o mesmo trabalho: a fila do Redis é o caminho rápido,
|
||||
// a varredura do Postgres garante que nada se perde se o Redis reiniciar.
|
||||
const queued = new Set(
|
||||
await listQueuedJobIds(deps).catch((error) => {
|
||||
onError("list", error);
|
||||
return [] as string[];
|
||||
}),
|
||||
);
|
||||
for (let index = 0; index < batch; index += 1) {
|
||||
const jobId = await deps.coordinator.dequeue().catch(() => null);
|
||||
if (!jobId) break;
|
||||
queued.add(jobId);
|
||||
}
|
||||
|
||||
for (const jobId of queued) {
|
||||
if (!running) break;
|
||||
const rows = await readJobForWorker(deps, jobId).catch((error) => {
|
||||
onError("read", error);
|
||||
return [];
|
||||
});
|
||||
const row = rows[0];
|
||||
if (!row) continue;
|
||||
const progress = asJson<JobProgress | null>(row.progress, null);
|
||||
if (!progress?.sourceId) continue;
|
||||
const ctx = await loadContext(deps, row.user_id, row.workspace_id).catch((error) => {
|
||||
onError("context", error);
|
||||
return null;
|
||||
});
|
||||
if (!ctx) continue;
|
||||
await runProcess(deps, ctx, jobId, progress.sourceId)
|
||||
.then(() => {
|
||||
processed.push(jobId);
|
||||
})
|
||||
.catch((error) => {
|
||||
onError("process", error);
|
||||
});
|
||||
}
|
||||
|
||||
if (heartbeatPath) {
|
||||
await writeHeartbeat(heartbeatPath).catch((error) => onError("heartbeat", error));
|
||||
}
|
||||
} finally {
|
||||
inTick = false;
|
||||
}
|
||||
return { processed, cleaned, recovered };
|
||||
}
|
||||
|
||||
/**
|
||||
* Para de aceitar trabalho novo e espera o tick corrente terminar, para que o
|
||||
* `finally` do processamento libere o lock. Sem essa espera o job ficaria em
|
||||
* `processing` até o lock vencer por TTL.
|
||||
*/
|
||||
async function shutdown(graceMs = 20_000, sleep = (ms: number) => new Promise((r) => setTimeout(r, ms))): Promise<boolean> {
|
||||
running = false;
|
||||
const deadline = Date.now() + graceMs;
|
||||
while (inTick && Date.now() < deadline) {
|
||||
await sleep(100);
|
||||
}
|
||||
return !inTick;
|
||||
}
|
||||
|
||||
return {
|
||||
tick,
|
||||
shutdown,
|
||||
isRunning: () => running,
|
||||
isBusy: () => inTick,
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user