Fas 1a punkt 3+4: worker-härdning (DLQ, larm, multi-instance), AAMOS training export
This commit is contained in:
Sven (AAMOS AI)
+114
-20
@@ -1,6 +1,7 @@
|
||||
import "dotenv/config";
|
||||
import { Queue, Worker, type Job } from "bullmq";
|
||||
import IORedis from "ioredis";
|
||||
import { createServer } from "node:http";
|
||||
import { createContext } from "./context.js";
|
||||
import { processScanJob } from "./processors/scans.js";
|
||||
import { processModerateRecipe } from "./processors/moderation.js";
|
||||
@@ -17,19 +18,29 @@ import {
|
||||
processTrainingExport,
|
||||
} from "./processors/maintenance.js";
|
||||
|
||||
import {
|
||||
DEAD_LETTER_QUEUE_NAME,
|
||||
JOB_QUEUE_NAME as QUEUE_NAME,
|
||||
WORKER_DEFAULT_ATTEMPTS,
|
||||
WORKER_DEFAULT_BACKOFF_MS,
|
||||
} from "@app/shared-types";
|
||||
|
||||
/**
|
||||
* workern. Konsumerar kön "<slug>-jobs" (spec §50):
|
||||
* App → signed S3 upload → API job → worker → AAMOS → result → app
|
||||
* workern. Konsumerar kön "<slug>-jobs" (spec §50).
|
||||
*
|
||||
* Jobbnamn = jobbtyp (spec §54). Deterministiska jobb rör aldrig AAMOS;
|
||||
* AI-jobb går alltid via de typade kontrakten i @app/ai-contracts.
|
||||
* Flera instanser kan köra samtidigt mot samma Redis- och DB-anslutning
|
||||
* utan kodändring (BullMQ hanterar fördelning). Varje instans exponerar
|
||||
* en liten HTTP-healthcheck så att lastbalansering kan se om den lever.
|
||||
*/
|
||||
|
||||
import { JOB_QUEUE_NAME as QUEUE_NAME } from "@app/shared-types";
|
||||
const connection = new IORedis(process.env.REDIS_URL ?? "redis://localhost:6379", {
|
||||
maxRetriesPerRequest: null,
|
||||
});
|
||||
const ctx = createContext();
|
||||
const instanceId = process.env.WORKER_INSTANCE_ID ?? `worker-${process.pid}`;
|
||||
|
||||
const deadLetterQueue = new Queue(DEAD_LETTER_QUEUE_NAME, { connection });
|
||||
const queue = new Queue(QUEUE_NAME, { connection });
|
||||
|
||||
const worker = new Worker(
|
||||
QUEUE_NAME,
|
||||
@@ -116,13 +127,48 @@ const worker = new Worker(
|
||||
{
|
||||
connection,
|
||||
concurrency: Number(process.env.WORKER_CONCURRENCY ?? 5),
|
||||
maxStalledCount: 2,
|
||||
stalledInterval: 30_000,
|
||||
limiter: {
|
||||
max: Number(process.env.WORKER_RATE_LIMIT_MAX ?? 60),
|
||||
duration: Number(process.env.WORKER_RATE_LIMIT_DURATION_MS ?? 60_000),
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
worker.on("completed", (job) => log(`✓ ${job.name} (${job.id})`));
|
||||
worker.on("failed", (job, err) => {
|
||||
console.error(`✗ ${job?.name} (${job?.id}): ${err.message}`);
|
||||
void alertWebhook(`Worker-jobb misslyckades: ${job?.name} (${job?.id}): ${err.message}`);
|
||||
worker.on("failed", async (job, err) => {
|
||||
const jobType = (job?.data?.jobType as string) ?? job?.name ?? "unknown";
|
||||
const attempts = job?.opts?.attempts ?? WORKER_DEFAULT_ATTEMPTS;
|
||||
const attemptsMade = job?.attemptsMade ?? 0;
|
||||
console.error(`✗ ${jobType} (${job?.id}): ${err.message} (attempt ${attemptsMade}/${attempts})`);
|
||||
|
||||
if (job && attemptsMade >= attempts) {
|
||||
try {
|
||||
await deadLetterQueue.add(
|
||||
job.name,
|
||||
{
|
||||
...job.data,
|
||||
_deadLetteredAt: new Date().toISOString(),
|
||||
_instanceId: instanceId,
|
||||
_failureReason: err.message,
|
||||
_failedAt: new Date().toISOString(),
|
||||
},
|
||||
{
|
||||
jobId: `${instanceId}:${job.id}`,
|
||||
removeOnComplete: { age: 30 * 24 * 60 * 60 }, // 30 dagar
|
||||
removeOnFail: { age: 90 * 24 * 60 * 60 }, // 90 dagar
|
||||
},
|
||||
);
|
||||
log(`→ dead letter: ${jobType} (${job.id})`);
|
||||
} catch (dlqErr) {
|
||||
console.error("Kunde inte skriva till dead letter queue:", dlqErr);
|
||||
}
|
||||
}
|
||||
|
||||
await alertWebhook(
|
||||
`Worker-jobb misslyckades: ${jobType} (${job?.id}): ${err.message} (attempt ${attemptsMade}/${attempts})`,
|
||||
);
|
||||
});
|
||||
|
||||
/** Larm till valfri webhook (Slack/Discord/Teams …) – fire-and-forget. */
|
||||
@@ -142,50 +188,98 @@ async function alertWebhook(text: string): Promise<void> {
|
||||
}
|
||||
|
||||
// --- Återkommande jobb via BullMQ Job Schedulers (spec §54) ---
|
||||
const queue = new Queue(QUEUE_NAME, { connection });
|
||||
async function registerRepeatableJobs() {
|
||||
const baseOpts = {
|
||||
attempts: WORKER_DEFAULT_ATTEMPTS,
|
||||
backoff: { type: "exponential" as const, delay: WORKER_DEFAULT_BACKOFF_MS },
|
||||
removeOnComplete: { count: 100 },
|
||||
removeOnFail: { count: 100 },
|
||||
};
|
||||
|
||||
await queue.upsertJobScheduler(
|
||||
"scheduler-outbox",
|
||||
{ every: 30_000 },
|
||||
{ name: "PUBLISH_OUTBOX", data: { jobType: "PUBLISH_OUTBOX" } },
|
||||
{ name: "PUBLISH_OUTBOX", data: { jobType: "PUBLISH_OUTBOX" }, opts: baseOpts },
|
||||
);
|
||||
await queue.upsertJobScheduler(
|
||||
"scheduler-expiry",
|
||||
{ pattern: "0 7 * * *", tz: "Europe/Stockholm" },
|
||||
{ name: "SEND_EXPIRY_NOTIFICATION", data: { jobType: "SEND_EXPIRY_NOTIFICATION" } },
|
||||
{ name: "SEND_EXPIRY_NOTIFICATION", data: { jobType: "SEND_EXPIRY_NOTIFICATION" }, opts: baseOpts },
|
||||
);
|
||||
await queue.upsertJobScheduler(
|
||||
"scheduler-memory",
|
||||
{ pattern: "30 3 * * *", tz: "Europe/Stockholm" },
|
||||
{ name: "UPDATE_USER_MEMORY", data: { jobType: "UPDATE_USER_MEMORY" } },
|
||||
{ name: "UPDATE_USER_MEMORY", data: { jobType: "UPDATE_USER_MEMORY" }, opts: baseOpts },
|
||||
);
|
||||
await queue.upsertJobScheduler(
|
||||
"scheduler-subs",
|
||||
{ pattern: "15 4 * * *", tz: "Europe/Stockholm" },
|
||||
{ name: "VERIFY_SUBSCRIPTION", data: { jobType: "VERIFY_SUBSCRIPTION" } },
|
||||
{ name: "VERIFY_SUBSCRIPTION", data: { jobType: "VERIFY_SUBSCRIPTION" }, opts: baseOpts },
|
||||
);
|
||||
await queue.upsertJobScheduler(
|
||||
"scheduler-retention",
|
||||
{ pattern: "45 4 * * *", tz: "Europe/Stockholm" },
|
||||
{ name: "RUN_RETENTION", data: { jobType: "RUN_RETENTION" } },
|
||||
{ name: "RUN_RETENTION", data: { jobType: "RUN_RETENTION" }, opts: baseOpts },
|
||||
);
|
||||
await queue.upsertJobScheduler(
|
||||
"scheduler-training",
|
||||
{ pattern: "45 2 * * 0", tz: "Europe/Stockholm" },
|
||||
{ name: "BUILD_TRAINING_SAMPLE", data: { jobType: "BUILD_TRAINING_SAMPLE" } },
|
||||
{ name: "BUILD_TRAINING_SAMPLE", data: { jobType: "BUILD_TRAINING_SAMPLE" }, opts: baseOpts },
|
||||
);
|
||||
}
|
||||
|
||||
registerRepeatableJobs()
|
||||
.then(() => log("Worker igång. Väntar på jobb …"))
|
||||
.catch((err) => {
|
||||
console.error("Kunde inte registrera återkommande jobb:", err);
|
||||
// --- Minimal healthcheck-server så att flera instanser kan övervakas ---
|
||||
const healthPort = Number(process.env.WORKER_HEALTH_PORT ?? 4001);
|
||||
let processedCount = 0;
|
||||
let failedCount = 0;
|
||||
worker.on("completed", () => processedCount++);
|
||||
worker.on("failed", () => failedCount++);
|
||||
|
||||
const healthServer = createServer(async (req, res) => {
|
||||
if (req.url === "/healthz") {
|
||||
const [queueCount, dlqCount, workers] = await Promise.all([
|
||||
queue.getJobCounts("wait", "active", "delayed", "completed", "failed"),
|
||||
deadLetterQueue.getJobCounts("wait", "active", "completed", "failed"),
|
||||
queue.getWorkers(),
|
||||
]);
|
||||
res.writeHead(200, { "content-type": "application/json" });
|
||||
res.end(
|
||||
JSON.stringify({
|
||||
ok: true,
|
||||
instanceId,
|
||||
uptimeSeconds: Math.floor(process.uptime()),
|
||||
queue: queueCount,
|
||||
deadLetter: dlqCount,
|
||||
activeWorkers: workers.length,
|
||||
processedSinceStart: processedCount,
|
||||
failedSinceStart: failedCount,
|
||||
}),
|
||||
);
|
||||
return;
|
||||
}
|
||||
res.writeHead(404);
|
||||
res.end("not found");
|
||||
});
|
||||
|
||||
async function start() {
|
||||
await registerRepeatableJobs();
|
||||
log("Worker igång. Väntar på jobb …");
|
||||
healthServer.listen(healthPort, () => {
|
||||
log(`Healthcheck på :${healthPort}`);
|
||||
});
|
||||
}
|
||||
|
||||
start().catch((err) => {
|
||||
console.error("Kunde inte starta worker:", err);
|
||||
process.exit(1);
|
||||
});
|
||||
|
||||
const shutdown = async () => {
|
||||
log("Stänger ner …");
|
||||
healthServer.close();
|
||||
await worker.close();
|
||||
await queue.close();
|
||||
await deadLetterQueue.close();
|
||||
connection.disconnect();
|
||||
await ctx.close();
|
||||
process.exit(0);
|
||||
@@ -194,5 +288,5 @@ process.on("SIGTERM", () => void shutdown());
|
||||
process.on("SIGINT", () => void shutdown());
|
||||
|
||||
function log(msg: string) {
|
||||
console.log(`[worker ${new Date().toISOString()}] ${msg}`);
|
||||
console.log(`[worker ${instanceId} ${new Date().toISOString()}] ${msg}`);
|
||||
}
|
||||
|
||||
@@ -273,7 +273,7 @@ export async function processMemorySync(ctx: WorkerContext): Promise<number> {
|
||||
return updates;
|
||||
}
|
||||
|
||||
/** BUILD_TRAINING_SAMPLE (spec §33): exportera korrigeringar MED samtycke. */
|
||||
/** BUILD_TRAINING_SAMPLE (spec §33): exportera korrigeringar MED samtycke till AAMOS. */
|
||||
export async function processTrainingExport(ctx: WorkerContext): Promise<number> {
|
||||
const corrections = await ctx.db
|
||||
.select()
|
||||
@@ -281,19 +281,48 @@ export async function processTrainingExport(ctx: WorkerContext): Promise<number>
|
||||
.where(isNull(schema.aiCorrections.exportedToTraining))
|
||||
.limit(100);
|
||||
|
||||
let exported = 0;
|
||||
for (const correction of corrections) {
|
||||
const snapshot = correction.consentSnapshot as Record<string, string>;
|
||||
// Endast korrigeringar där anonymiserad förbättring var beviljad vid tillfället.
|
||||
if (snapshot.anonymized_improvement !== "granted") continue;
|
||||
// Här skulle exporten till AAMOS training-pipeline ske (avidentifierad).
|
||||
const eligible = corrections.filter((c) => {
|
||||
const snapshot = c.consentSnapshot as Record<string, string>;
|
||||
return snapshot.anonymized_improvement === "granted";
|
||||
});
|
||||
|
||||
if (eligible.length === 0) return 0;
|
||||
|
||||
const result = await ctx.aamos.runTask(
|
||||
"EXPORT_TRAINING_SAMPLE",
|
||||
{
|
||||
marketLocale: "sv-SE",
|
||||
samples: eligible.map((c) => ({
|
||||
taskType: c.taskType,
|
||||
aiOutput: c.aiOutput as Record<string, unknown>,
|
||||
userCorrection: c.userCorrection as Record<string, unknown>,
|
||||
modelVersion: c.modelVersion ?? null,
|
||||
promptVersion: c.promptVersion ?? null,
|
||||
})),
|
||||
},
|
||||
{
|
||||
correlationId: `training-export-${Date.now()}`,
|
||||
consentFlags: {
|
||||
personalization: false,
|
||||
anonymizedImprovement: true,
|
||||
imageTraining: false,
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
if (result.status !== "ok" || !result.output) {
|
||||
throw new Error(`AAMOS training export failed: ${result.error ?? "unknown"}`);
|
||||
}
|
||||
|
||||
const batchId = result.output.batchId;
|
||||
for (const correction of eligible) {
|
||||
await ctx.db
|
||||
.update(schema.aiCorrections)
|
||||
.set({ exportedToTraining: new Date() })
|
||||
.set({ exportedToTraining: new Date(), trainingBatchId: batchId })
|
||||
.where(eq(schema.aiCorrections.id, correction.id));
|
||||
exported++;
|
||||
}
|
||||
return exported;
|
||||
|
||||
return eligible.length;
|
||||
}
|
||||
|
||||
/** VERIFY_SUBSCRIPTION: flagga prenumerationer som passerat expiry. */
|
||||
|
||||
Reference in New Issue
Block a user