From 6f628c4693c63a7f5ec5009744f5a89a88aa160a Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 20:52:51 +0100 Subject: [PATCH] fix(core): job failure semantics, stale-lock reaper, run-scoped context, reload Co-Authored-By: Claude Sonnet 5.5 --- docs/ARCHITECTURE.md | 6 +- docs/DECISIONS.md | 16 ++ docs/TODO.md | 5 +- packages/core/src/connections.ts | 162 +++++++++++++----- packages/core/src/jobs.ts | 86 +++++++--- packages/core/src/registry.ts | 2 + packages/core/test/connections.test.ts | 66 +++++++ .../core/test/fixtures/plugins/echo/index.ts | 1 + packages/core/test/jobs.test.ts | 74 +++++++- packages/core/test/registry.test.ts | 20 +++ plugins/mangabaka/src/api.ts | 1 - 11 files changed, 356 insertions(+), 83 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index b9d3efa..812fe5e 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -23,9 +23,9 @@ user layer or files a Changes-inbox event. | `docs/design/mockups` | approved UI mockups (reference only) | exists | ## Data flow -Sync lifecycle (`packages/core/src/sync/`; steps 1–2 are Task 9's job worker, planned): -1. The scheduler enqueues a job. -2. The worker loads the connection and plugin and calls `runSync(deps, plugin, connectionId, mode)`, injecting +Sync lifecycle (`packages/core/src/sync/`; steps 1–2 are `jobs.ts` and `connections.ts`): +1. The scheduler enqueues a `sync` job (lockKey `sync:`) for each non-paused connection that is due. +2. The worker claims the job; `syncHandler` loads the connection and plugin and calls `runSync(deps, plugin, connectionId, mode)`, injecting `makeContext` (http client, credentials, cursor). `mode` picks `pull`, `pullChanges` or `importFile`. 3. `runSync` opens a `source_runs` row and iterates the plugin's `SourceRecord`s one at a time. - `listen` records are batched into `ingestListens` (insert-or-ignore into `listens`; no snapshot, no events). diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index a851d38..e432b77 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -4,6 +4,22 @@ Append-only. Never edit or delete an entry. Supersede it with a new one that lin One entry per decision a future maintainer would otherwise have to reverse-engineer. No index: `grep -n '^## ' docs/DECISIONS.md` lists every entry (date, title, line number). +## 2026-10-30: Job failure semantics and plugin reload limits +**Status:** accepted +**Context:** Task 9 fix round 1. +**Decision:** +- **Failures before `runSync` never retry.** A missing/paused connection drops the job (`DropJobError`: `done`, message in + `last_error`). A missing plugin or undecryptable credentials (`CryptoError`) insert a failed `source_runs` row, set the + connection to `error`, advance `last_run_at` and throw `NonRetryableError` (job `failed`, no backoff). The scheduler still + picks `error` connections, one interval later, so fixing the cause heals them. +- **Stale locks.** Every worker tick re-queues `running` jobs whose `locked_at` is over an hour old (attempts unchanged), so a + crash cannot hold a lockKey forever. Job bookkeeping errors are logged (`[jobs]`), never unhandled; `stop()` never rejects + and aborts the `signal` passed to handlers (`handler(payload, { signal })`). +- **Reload** evicts the entry file from `require.cache` and imports with `?v=` (Bun ignores the query alone for + cached `.ts`). Only entry files reload: changes in a plugin's other modules need a server restart. +- `credentials.save` and `updateConnection` read-merge-write in one transaction (row lock on Postgres). `ctx.log` is buffered and + appended to the run's `source_runs.log` when the run ends. + ## 2026-10-30: Jobs, plugin registry, connections: rate limit in the manifest, secret split, per-connection lock **Status:** accepted **Context:** Task 9 (`packages/core/src/{jobs,registry,connections}.ts`). diff --git a/docs/TODO.md b/docs/TODO.md index 1bc3541..4ac4e3c 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -24,8 +24,9 @@ Small, specific, independently actionable. Remove only when actually done. file, and compare steps and token use. ## Known issues -- [ ] Jobs: a job left `running` by a crash is never re-queued; add a stale-lock reaper using `locked_at`. -- [ ] Plugins: `ctx.log` is discarded; route it into `source_runs.log`. +- [ ] Plugins: reload re-imports entry files only; edits to a plugin's other modules need a server restart. +- [ ] Plugins: rate limiting is per run (each run builds its own client), not per plugin (spec §11); share a limiter per plugin in Phase 2. +- [ ] Plugins: `ctx.log` reaches `source_runs.log` only after the run ends, so a crash mid-run loses it. - [ ] Restore: a new child work whose parent+position collides with an existing child of a re-matched parent fails with `ImportError("conflict")` instead of being merged or repositioned. - [ ] Restore: an unlink of identical attachment bytes between the pre-write and the commit could remove the file; diff --git a/packages/core/src/connections.ts b/packages/core/src/connections.ts index 10a9b43..7a2c275 100644 --- a/packages/core/src/connections.ts +++ b/packages/core/src/connections.ts @@ -2,10 +2,11 @@ import type { Db } from "@sttracker/db/db"; import { newId, nowIso } from "@sttracker/db/ids"; import { fromJson, toJson } from "@sttracker/db/json"; import type { ConnectionStatus } from "@sttracker/db/schema"; -import { createHttpClient } from "@sttracker/plugin-sdk/http"; +import { createHttpClient, type HttpClientOptions } from "@sttracker/plugin-sdk/http"; import type { HttpClient, PluginContext, PluginManifest } from "@sttracker/plugin-sdk/types"; import { decryptJson, encryptJson } from "./crypto.ts"; -import { NotFoundError, ValidationError } from "./errors.ts"; +import { CryptoError, NotFoundError, ValidationError } from "./errors.ts"; +import { DropJobError, NonRetryableError, type JobContext } from "./jobs.ts"; import type { PluginRegistry } from "./registry.ts"; import { runSync, type Connection, type RunSummary } from "./sync/run.ts"; @@ -89,26 +90,30 @@ export interface ConnectionPatch { status?: ConnectionStatus; } -/** Secrets omitted from `config` are kept; plain config is replaced wholesale. */ +/** Secrets omitted from `config` are kept; plain config is replaced wholesale. Read-modify-write runs in one transaction. */ export async function updateConnection(db: Db, secret: string, userId: string, id: string, patch: ConnectionPatch, manifest: PluginManifest): Promise { - const row = await db.k.selectFrom("source_connections").selectAll().where("id", "=", id).where("user_id", "=", userId).executeTakeFirst(); - if (!row) throw new NotFoundError("connection"); - if (row.plugin_id !== manifest.id) throw new ValidationError("pluginId", "manifest_mismatch"); checkSchedule(patch.scheduleMinutes); - const set: Record = {}; - if (patch.name !== undefined) { - if (patch.name.trim() === "") throw new ValidationError("name", "required"); - set.name = patch.name.trim(); - } - if (patch.scheduleMinutes !== undefined) set.schedule_minutes = patch.scheduleMinutes; - if (patch.status !== undefined) set.status = patch.status; - if (patch.config !== undefined || patch.credentials !== undefined) { - const split = patch.config ? splitConfig(manifest, patch.config) : { plain: undefined, secrets: {} }; - if (split.plain) set.config = toJson(split.plain); - const merged = { ...((await readCreds(secret, row.credentials_enc)) ?? {}), ...split.secrets, ...patch.credentials }; - set.credentials_enc = await encOrNull(secret, merged); - } - if (Object.keys(set).length > 0) await db.k.updateTable("source_connections").set(set).where("id", "=", id).execute(); + await db.k.transaction().execute(async (tx) => { + let q = tx.selectFrom("source_connections").selectAll().where("id", "=", id).where("user_id", "=", userId); + if (db.dialect === "postgres") q = q.forUpdate(); + const row = await q.executeTakeFirst(); + if (!row) throw new NotFoundError("connection"); + if (row.plugin_id !== manifest.id) throw new ValidationError("pluginId", "manifest_mismatch"); + const set: Record = {}; + if (patch.name !== undefined) { + if (patch.name.trim() === "") throw new ValidationError("name", "required"); + set.name = patch.name.trim(); + } + if (patch.scheduleMinutes !== undefined) set.schedule_minutes = patch.scheduleMinutes; + if (patch.status !== undefined) set.status = patch.status; + if (patch.config !== undefined || patch.credentials !== undefined) { + const split = patch.config ? splitConfig(manifest, patch.config) : { plain: undefined, secrets: {} }; + if (split.plain) set.config = toJson(split.plain); + const merged = { ...((await readCreds(secret, row.credentials_enc)) ?? {}), ...split.secrets, ...patch.credentials }; + set.credentials_enc = await encOrNull(secret, merged); + } + if (Object.keys(set).length > 0) await tx.updateTable("source_connections").set(set).where("id", "=", id).execute(); + }); } /** Soft delete: the connection is paused and its credentials are wiped. Records and runs stay. */ @@ -140,26 +145,43 @@ export async function listConnections(db: Db, userId: string): Promise { - let creds = await readCreds(secret, conn.credentialsEnc); +export interface ContextOpts { + /** Replaces the rate-limited client entirely. */ + http?: HttpClient; + /** Overrides for the built client (fetch, sleep, clock); used by tests. */ + httpOptions?: Pick; + signal?: AbortSignal; + /** Receives `ctx.log` messages. */ + onLog?: (msg: string) => void; +} + +/** Synchronous context builder over already-decrypted credentials. */ +function buildContext(db: Db, secret: string, conn: Connection, manifest: PluginManifest, initial: Creds | null, opts: ContextOpts): PluginContext { + let creds = initial; let cursor = conn.cursor; return { http: - http ?? - createHttpClient({ ratePerMinute: manifest.rateLimit?.perMinute ?? 60, burst: manifest.rateLimit?.burst ?? 5, userAgent: USER_AGENT }), - log: () => {}, - config: { ...conn.config, ...(creds ?? {}) }, + opts.http ?? + createHttpClient({ + ratePerMinute: manifest.rateLimit?.perMinute ?? 60, + burst: manifest.rateLimit?.burst ?? 5, + userAgent: USER_AGENT, + ...opts.httpOptions, + }), + log: (msg) => opts.onLog?.(msg), + config: { ...conn.config, ...(initial ?? {}) }, credentials: { get: async () => creds, async save(c) { - const merged = { ...(creds ?? {}), ...c }; - await db.k.updateTable("source_connections").set({ credentials_enc: await encryptJson(secret, merged) }).where("id", "=", conn.id).execute(); - creds = merged; + // Read the current blob, merge, and write in one transaction so concurrent saves cannot lose keys. + creds = await db.k.transaction().execute(async (tx) => { + let q = tx.selectFrom("source_connections").select("credentials_enc").where("id", "=", conn.id); + if (db.dialect === "postgres") q = q.forUpdate(); + const row = await q.executeTakeFirstOrThrow(); + const merged = { ...((await readCreds(secret, row.credentials_enc)) ?? {}), ...c }; + await tx.updateTable("source_connections").set({ credentials_enc: await encryptJson(secret, merged) }).where("id", "=", conn.id).execute(); + return merged; + }); }, }, cursor: { @@ -170,10 +192,19 @@ export async function makePluginContext(db: Db, secret: string, conn: Connection }, }, now: () => new Date(), - signal: new AbortController().signal, + signal: opts.signal ?? new AbortController().signal, }; } +/** + * Build the plugin context for one run. `ctx.http` is rate limited from `manifest.rateLimit` (default 60/min, burst 5) + * unless `opts.http` is given. `credentials.save` merges into the current blob and persists it encrypted immediately; + * `cursor.set` persists immediately. + */ +export async function makePluginContext(db: Db, secret: string, conn: Connection, _runId: string, manifest: PluginManifest, opts: ContextOpts = {}): Promise { + return buildContext(db, secret, conn, manifest, await readCreds(secret, conn.credentialsEnc), opts); +} + export interface SyncHandlerDeps { db: Db; secret: string; @@ -181,33 +212,47 @@ export interface SyncHandlerDeps { http?: HttpClient; } +const MSG_CRYPTO = "Cannot decrypt this source's credentials. Was STTRACKER_SECRET changed? Reconnect the source to fix it."; + +/** Record a run that failed before `runSync` could start, flag the connection, and stop the job without retry. */ +async function failBeforeRun(db: Db, connectionId: string, mode: "full" | "changes", message: string): Promise { + const now = nowIso(); + await db.k + .insertInto("source_runs") + .values({ id: newId(), connection_id: connectionId, mode, started_at: now, finished_at: now, status: "failed", counts: "{}", log: message, error: message }) + .execute(); + await db.k.updateTable("source_connections").set({ status: "error", last_run_at: now }).where("id", "=", connectionId).execute(); + throw new NonRetryableError(message); +} + /** * The `"sync"` job handler. Payload `{ connectionId, mode? }`. Runs `full` when asked, when the plugin cannot * `pullChanges`, or when the last successful full run is over a week old; otherwise `changes`. - * A failed run is recorded by `runSync` and rethrown so the worker retries with backoff. + * A missing or paused connection drops the job. A missing plugin or undecryptable credentials record a failed + * run, mark the connection `error` and fail the job without retry. A failed run is rethrown so the worker retries. + * `ctx.log` output is appended to the run's `source_runs.log`. */ -export function syncHandler(deps: SyncHandlerDeps): (payload: { connectionId: string; mode?: "full" | "changes" }) => Promise { - return async (payload) => { +export function syncHandler(deps: SyncHandlerDeps): (payload: { connectionId: string; mode?: "full" | "changes" }, jc?: JobContext) => Promise { + return async (payload, jc) => { const { db } = deps; - const row = await db.k.selectFrom("source_connections").select(["id", "plugin_id", "status"]).where("id", "=", payload.connectionId).executeTakeFirst(); - if (!row) throw new NotFoundError("connection"); - if (row.status === "paused") return; - const plugin = deps.registry.get(row.plugin_id); - if (!plugin) throw new NotFoundError(`plugin ${row.plugin_id}`); + const full = await db.k.selectFrom("source_connections").selectAll().where("id", "=", payload.connectionId).executeTakeFirst(); + if (!full) throw new DropJobError("connection not found"); + if (full.status === "paused") throw new DropJobError("connection is paused"); + const plugin = deps.registry.get(full.plugin_id); + if (!plugin) return failBeforeRun(db, full.id, payload.mode ?? "full", `Plugin '${full.plugin_id}' is not installed or failed to load.`); let mode: "full" | "changes" = "full"; if (plugin.pullChanges && payload.mode !== "full") { const cutoff = new Date(Date.now() - WEEK_MS).toISOString(); const recentFull = await db.k .selectFrom("source_runs") .select("id") - .where("connection_id", "=", row.id) + .where("connection_id", "=", full.id) .where("mode", "=", "full") .where("status", "=", "ok") .where("started_at", ">=", cutoff) .executeTakeFirst(); if (recentFull) mode = "changes"; } - const full = (await db.k.selectFrom("source_connections").selectAll().where("id", "=", row.id).executeTakeFirstOrThrow()); const conn: Connection = { id: full.id, userId: full.user_id, @@ -221,8 +266,31 @@ export function syncHandler(deps: SyncHandlerDeps): (payload: { connectionId: st lastRunAt: full.last_run_at, createdAt: full.created_at, }; - const ctx = await makePluginContext(db, deps.secret, conn, "sync", plugin.manifest, deps.http); - const summary = await runSync({ db, makeContext: () => ctx }, plugin, row.id, mode); + let creds: Creds | null; + try { + creds = await readCreds(deps.secret, conn.credentialsEnc); + } catch (e) { + if (e instanceof CryptoError) return failBeforeRun(db, full.id, mode, MSG_CRYPTO); + throw e; + } + const logs: string[] = []; + let runId: string | null = null; + const makeContext = (c: Connection, id: string) => { + runId = id; + return buildContext(db, deps.secret, c, plugin.manifest, creds, { http: deps.http, signal: jc?.signal, onLog: (m) => void logs.push(m) }); + }; + let summary: RunSummary; + try { + summary = await runSync({ db, makeContext }, plugin, full.id, mode); + } finally { + if (runId !== null && logs.length > 0) { + const cur = await db.k.selectFrom("source_runs").select("log").where("id", "=", runId).executeTakeFirst(); + if (cur) { + const log = cur.log === "" ? logs.join("\n") : `${logs.join("\n")}\n${cur.log}`; + await db.k.updateTable("source_runs").set({ log }).where("id", "=", runId).execute(); + } + } + } if (summary.status === "failed") throw new Error(summary.error ?? "sync failed"); return summary; }; diff --git a/packages/core/src/jobs.ts b/packages/core/src/jobs.ts index 3767f53..081a123 100644 --- a/packages/core/src/jobs.ts +++ b/packages/core/src/jobs.ts @@ -5,7 +5,24 @@ import { sql } from "kysely"; const MAX_ATTEMPTS = 5; -export type JobHandler = (payload: any) => Promise; +const STALE_LOCK_MS = 60 * 60 * 1000; + +export interface JobContext { + /** Aborted when the worker is stopped. */ + signal: AbortSignal; +} +export type JobHandler = (payload: any, ctx: JobContext) => Promise; + +/** Throw from a handler to fail the job at once, with no retry. */ +export class NonRetryableError extends Error { + override readonly name = "NonRetryableError"; +} +/** Throw from a handler to drop the job: it is marked `done` and the message is kept in `last_error`. */ +export class DropJobError extends Error { + override readonly name = "DropJobError"; +} + +const logErr = (err: unknown) => console.error("[jobs]", err); /** * Queue a job. Returns its id, or null when a `queued|running` job already holds `lockKey`. @@ -61,28 +78,40 @@ async function claim(db: Db, id: string): Promise { return r.numUpdatedRows === 1n; } -async function runJob(db: Db, handlers: Record, job: { id: string; type: string; payload: string; attempts: number }) { +async function runJob(db: Db, handlers: Record, job: { id: string; type: string; payload: string; attempts: number }, signal: AbortSignal): Promise { try { - const h = handlers[job.type]; - if (!h) throw new Error(`no handler for job type "${job.type}"`); - await h(fromJson(job.payload, null)); - await db.k.updateTable("jobs").set({ status: "done", locked_at: null, last_error: null }).where("id", "=", job.id).execute(); - } catch (e) { - const attempts = job.attempts + 1; - const err = e instanceof Error ? e.message : String(e); - if (attempts >= MAX_ATTEMPTS) { - await db.k.updateTable("jobs").set({ status: "failed", attempts, last_error: err, locked_at: null }).where("id", "=", job.id).execute(); - } else { - const runAt = new Date(Date.now() + 2 ** attempts * 60_000).toISOString(); - await db.k - .updateTable("jobs") - .set({ status: "queued", attempts, last_error: err, run_at: runAt, locked_at: null }) - .where("id", "=", job.id) - .execute(); + try { + const h = handlers[job.type]; + if (!h) throw new NonRetryableError(`no handler for job type "${job.type}"`); + await h(fromJson(job.payload, null), { signal }); + await db.k.updateTable("jobs").set({ status: "done", locked_at: null, last_error: null }).where("id", "=", job.id).execute(); + } catch (e) { + const attempts = job.attempts + 1; + const err = e instanceof Error ? e.message : String(e); + if (e instanceof DropJobError) { + await db.k.updateTable("jobs").set({ status: "done", locked_at: null, last_error: err }).where("id", "=", job.id).execute(); + } else if (e instanceof NonRetryableError || attempts >= MAX_ATTEMPTS) { + await db.k.updateTable("jobs").set({ status: "failed", attempts, last_error: err, locked_at: null }).where("id", "=", job.id).execute(); + } else { + const runAt = new Date(Date.now() + 2 ** attempts * 60_000).toISOString(); + await db.k + .updateTable("jobs") + .set({ status: "queued", attempts, last_error: err, run_at: runAt, locked_at: null }) + .where("id", "=", job.id) + .execute(); + } } + } catch (e) { + logErr(e); } } +/** Re-queue jobs a crashed process left `running`, so their lockKey cannot be held forever. */ +async function reapStale(db: Db): Promise { + const cutoff = new Date(Date.now() - STALE_LOCK_MS).toISOString(); + await db.k.updateTable("jobs").set({ status: "queued", locked_at: null }).where("status", "=", "running").where("locked_at", "<", cutoff).execute(); +} + /** Poll for due jobs and run them with bounded concurrency. Failures retry with backoff `2^attempts` minutes, then `failed`. */ export function startWorker( db: Db, @@ -92,11 +121,13 @@ export function startWorker( const pollMs = opts.pollMs ?? 1000; const concurrency = opts.concurrency ?? 2; const inflight = new Set>(); + const abort = new AbortController(); let stopped = false; let timer: ReturnType | undefined; let tickP: Promise = Promise.resolve(); async function tick(): Promise { + await reapStale(db); while (!stopped && inflight.size < concurrency) { const due = await db.k .selectFrom("jobs") @@ -111,7 +142,9 @@ export function startWorker( if (stopped || inflight.size >= concurrency) break; if (!(await claim(db, job.id))) continue; started++; - const p = runJob(db, handlers, job).finally(() => inflight.delete(p)); + const p: Promise = runJob(db, handlers, job, abort.signal) + .catch(logErr) + .finally(() => inflight.delete(p)); inflight.add(p); } if (started === 0) break; @@ -121,22 +154,23 @@ export function startWorker( function schedule() { if (stopped) return; timer = setTimeout(() => { - tickP = tick().catch(() => {}).finally(schedule); + tickP = tick().catch(logErr).finally(schedule); }, pollMs); } - tickP = tick().catch(() => {}).finally(schedule); + tickP = tick().catch(logErr).finally(schedule); return { async stop() { stopped = true; + abort.abort(); if (timer) clearTimeout(timer); - await tickP; - await Promise.all(inflight); + await tickP.catch(logErr); + await Promise.allSettled([...inflight]); }, }; } -/** Every `tickMs`, enqueue a `sync` job for each active connection whose schedule has come due. */ +/** Every `tickMs`, enqueue a `sync` job for each non-paused (active or errored) connection whose schedule has come due. */ export function startScheduler(db: Db, opts: { tickMs?: number } = {}): { stop(): void; tickNow(): Promise } { const tickMs = opts.tickMs ?? 60_000; let stopped = false; @@ -144,7 +178,7 @@ export function startScheduler(db: Db, opts: { tickMs?: number } = {}): { stop() const rows = await db.k .selectFrom("source_connections") .select(["id", "schedule_minutes", "last_run_at"]) - .where("status", "=", "active") + .where("status", "!=", "paused") .where("schedule_minutes", "is not", null) .execute(); const now = Date.now(); @@ -154,7 +188,7 @@ export function startScheduler(db: Db, opts: { tickMs?: number } = {}): { stop() if (due) await enqueue(db, "sync", { connectionId: r.id }, { lockKey: `sync:${r.id}` }); } } - const run = () => tick().catch(() => {}); + const run = () => tick().catch(logErr); void run(); const timer = setInterval(() => void run(), tickMs); return { diff --git a/packages/core/src/registry.ts b/packages/core/src/registry.ts index 32d26ee..e5734de 100644 --- a/packages/core/src/registry.ts +++ b/packages/core/src/registry.ts @@ -37,6 +37,8 @@ async function scan(dirs: string[]): Promise> { const file = join(dir, name, "index.ts"); if (!(await Bun.file(file).exists())) continue; try { + // Bun ignores the `?v=` query for cached .ts modules; evicting the cache entry is what makes reload work. + delete require.cache[file]; const mod = (await import(`${pathToFileURL(file).href}?v=${stamp}`)) as { default?: unknown }; if (!validShape(mod.default)) throw new Error("default export is not a valid plugin (manifest, sdkVersion 1)"); const id = mod.default.manifest.id; diff --git a/packages/core/test/connections.test.ts b/packages/core/test/connections.test.ts index b3de7a5..1859c9d 100644 --- a/packages/core/test/connections.test.ts +++ b/packages/core/test/connections.test.ts @@ -2,6 +2,7 @@ import { expect, spyOn, test } from "bun:test"; import { join } from "node:path"; import { forEachDialect } from "@sttracker/db/testing"; import { decryptJson } from "../src/crypto.ts"; +import { DropJobError, NonRetryableError } from "../src/jobs.ts"; import { createConnection, deleteConnection, listConnections, makePluginContext, syncHandler, updateConnection } from "../src/connections.ts"; import { loadPlugins } from "../src/registry.ts"; import { makeUser } from "./sync/helpers.ts"; @@ -95,4 +96,69 @@ forEachDialect("connections", (getDb) => { await handler({ connectionId: id }); expect((await modes()).at(-1)).toBe("changes:ok"); }); + + const setup = async (config: Record) => { + const db = getDb(); + const u = await makeUser(db); + const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config }, echo.manifest); + return { db, id }; + }; + const failedRuns = (id: string) => getDb().k.selectFrom("source_runs").selectAll().where("connection_id", "=", id).execute(); + + test("missing plugin: failed run, connection error, no retry", async () => { + const { db, id } = await setup({}); + await db.k.updateTable("source_connections").set({ plugin_id: "gone" }).where("id", "=", id).execute(); + await expect(syncHandler({ db, secret: SECRET, registry: reg })({ connectionId: id })).rejects.toBeInstanceOf(NonRetryableError); + const runs = await failedRuns(id); + expect(runs).toHaveLength(1); + expect(runs[0]).toMatchObject({ status: "failed", mode: "full", error: "Plugin 'gone' is not installed or failed to load." }); + const c = await raw(id); + expect(c.status).toBe("error"); + expect(c.last_run_at).not.toBeNull(); + }); + + test("undecryptable credentials: failed run, connection error, no retry", async () => { + const { db, id } = await setup({ token: "t" }); + await expect(syncHandler({ db, secret: "other-secret", registry: reg })({ connectionId: id })).rejects.toBeInstanceOf(NonRetryableError); + const runs = await failedRuns(id); + expect(runs[0]!.error).toBe("Cannot decrypt this source's credentials. Was STTRACKER_SECRET changed? Reconnect the source to fix it."); + expect((await raw(id)).status).toBe("error"); + }); + + test("missing or paused connection drops the job", async () => { + const { db, id } = await setup({}); + const h = syncHandler({ db, secret: SECRET, registry: reg }); + await expect(h({ connectionId: newId() })).rejects.toBeInstanceOf(DropJobError); + await db.k.updateTable("source_connections").set({ status: "paused" }).where("id", "=", id).execute(); + await expect(h({ connectionId: id })).rejects.toBeInstanceOf(DropJobError); + expect(await failedRuns(id)).toHaveLength(0); + }); + + test("ctx.log lands in the run's log", async () => { + const { db, id } = await setup({ label: "ok" }); + const res = await syncHandler({ db, secret: SECRET, registry: reg })({ connectionId: id }); + const run = await db.k.selectFrom("source_runs").selectAll().where("id", "=", res!.runId).executeTakeFirstOrThrow(); + expect(run.log).toContain("echo pulled"); + }); + + test("concurrent credentials.save calls keep every key", async () => { + const { db, id } = await setup({ token: "t" }); + const conn = await asConn(id); + const ctx = await makePluginContext(db, SECRET, conn, "r", echo.manifest); + await Promise.all([ctx.credentials.save({ a: "1" }), ctx.credentials.save({ b: "2" })]); + expect(await decryptJson(SECRET, (await raw(id)).credentials_enc!)).toEqual({ token: "t", a: "1", b: "2" }); + }); + + test("ctx.http is limited by manifest.rateLimit", async () => { + const { db, id } = await setup({}); + let t = 0; + const sleeps: number[] = []; + const manifest = { ...echo.manifest, rateLimit: { perMinute: 60, burst: 1 } }; + const ctx = await makePluginContext(db, SECRET, await asConn(id), "r", manifest, { + httpOptions: { fetchImpl: (async () => Response.json({})) as unknown as typeof fetch, now: () => t, sleep: async (ms) => { sleeps.push(ms); t += ms; } }, + }); + await ctx.http.json("https://example.test/a"); + await ctx.http.json("https://example.test/b"); + expect(sleeps).toEqual([1000]); + }); }); diff --git a/packages/core/test/fixtures/plugins/echo/index.ts b/packages/core/test/fixtures/plugins/echo/index.ts index e0ed7df..d9a327e 100644 --- a/packages/core/test/fixtures/plugins/echo/index.ts +++ b/packages/core/test/fixtures/plugins/echo/index.ts @@ -20,6 +20,7 @@ export default definePlugin({ await ctx.credentials.save({ refresh: "new-refresh" }); throw new Error("boom after save"); } + ctx.log("echo pulled"); const rec: SourceRecord = { kind: "entry", key: "echo-1", diff --git a/packages/core/test/jobs.test.ts b/packages/core/test/jobs.test.ts index 1e50b0f..70049f6 100644 --- a/packages/core/test/jobs.test.ts +++ b/packages/core/test/jobs.test.ts @@ -1,6 +1,6 @@ -import { expect, test } from "bun:test"; +import { expect, spyOn, test } from "bun:test"; import { forEachDialect } from "@sttracker/db/testing"; -import { enqueue, startScheduler, startWorker } from "../src/jobs.ts"; +import { DropJobError, enqueue, NonRetryableError, startScheduler, startWorker } from "../src/jobs.ts"; import { makeConnection, makeUser } from "./sync/helpers.ts"; const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); @@ -75,12 +75,14 @@ forEachDialect("jobs", (getDb) => { const due = await makeConnection(db, u, "p"); const fresh = await makeConnection(db, u, "p"); const manual = await makeConnection(db, u, "p"); + const errored = await makeConnection(db, u, "p"); const paused = await makeConnection(db, u, "p"); const old = new Date(Date.now() - 120 * 60_000).toISOString(); const recent = new Date(Date.now() - 5 * 60_000).toISOString(); const set = (id: string, v: object) => db.k.updateTable("source_connections").set(v).where("id", "=", id).execute(); await set(due, { schedule_minutes: 60, last_run_at: old }); await set(fresh, { schedule_minutes: 60, last_run_at: recent }); + await set(errored, { schedule_minutes: 60, last_run_at: old, status: "error" }); await set(paused, { schedule_minutes: 60, last_run_at: old, status: "paused" }); void manual; const s = startScheduler(db, { tickMs: 3_600_000 }); @@ -88,7 +90,71 @@ forEachDialect("jobs", (getDb) => { await s.tickNow(); s.stop(); const jobs = await db.k.selectFrom("jobs").selectAll().execute(); - expect(jobs.map((j) => j.lock_key)).toEqual([`sync:${due}`]); - expect(JSON.parse(jobs[0]!.payload)).toEqual({ connectionId: due }); + expect(jobs.map((j) => j.lock_key).sort()).toEqual([`sync:${due}`, `sync:${errored}`].sort()); + expect(JSON.parse(jobs.find((j) => j.lock_key === `sync:${due}`)!.payload)).toEqual({ connectionId: due }); + }); + + const row = (db: ReturnType, id: string) => db.k.selectFrom("jobs").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + async function until(fn: () => Promise) { + for (let i = 0; i < 200 && !(await fn()); i++) await sleep(10); + } + + test("NonRetryableError fails at once; DropJobError completes with last_error", async () => { + const db = getDb(); + const a = (await enqueue(db, "a", {}))!; + const b = (await enqueue(db, "b", {}))!; + const w = startWorker(db, { a: async () => { throw new NonRetryableError("fatal"); }, b: async () => { throw new DropJobError("gone"); } }, { pollMs: 10 }); + await until(async () => (await row(db, a)).status !== "queued" && (await row(db, b)).status !== "queued"); + await w.stop(); + expect(await row(db, a)).toMatchObject({ status: "failed", attempts: 1, last_error: "fatal" }); + expect(await row(db, b)).toMatchObject({ status: "done", last_error: "gone" }); + }); + + test("a stale running job is re-queued and then runs; a fresh one is left", async () => { + const db = getDb(); + const stale = (await enqueue(db, "t", {}, { lockKey: "s" }))!; + const fresh = (await enqueue(db, "never", {}))!; + await db.k.updateTable("jobs").set({ status: "running", locked_at: new Date(Date.now() - 2 * 3600_000).toISOString(), attempts: 2 }).where("id", "=", stale).execute(); + await db.k.updateTable("jobs").set({ status: "running", locked_at: nowStr() }).where("id", "=", fresh).execute(); + let ran = 0; + const w = startWorker(db, { t: async () => void ran++ }, { pollMs: 10 }); + await until(async () => (await row(db, stale)).status === "done"); + await w.stop(); + expect(ran).toBe(1); + expect((await row(db, stale)).attempts).toBe(2); + expect((await row(db, fresh)).status).toBe("running"); + }); + + test("failing bookkeeping is logged and stop() does not reject", async () => { + const db = getDb(); + const err = spyOn(console, "error").mockImplementation(() => {}); + await enqueue(db, "t", {}); + let updates = 0; + const k = new Proxy(db.k, { + get(t, p) { + const v = Reflect.get(t, p, t); + if (p === "updateTable") return (...a: unknown[]) => { if (++updates > 1) throw new Error("db down"); return (v as Function).apply(t, a); }; + return typeof v === "function" ? v.bind(t) : v; + }, + }); + let called = false; + const w = startWorker({ ...db, k } as typeof db, { t: async () => void (called = true) }, { pollMs: 10 }); + await until(async () => called && err.mock.calls.length > 0); + await expect(w.stop()).resolves.toBeUndefined(); + expect(err.mock.calls.some((c) => c[0] === "[jobs]")).toBe(true); + err.mockRestore(); + }); + + test("stop() aborts the signal handed to handlers", async () => { + const db = getDb(); + await enqueue(db, "t", {}); + let aborted = false; + let started = false; + const w = startWorker(db, { t: (_p, { signal }) => new Promise((res) => { started = true; signal.addEventListener("abort", () => { aborted = true; res(); }); }) }, { pollMs: 10 }); + await until(async () => started); + await w.stop(); + expect(aborted).toBe(true); }); }); + +const nowStr = () => new Date().toISOString(); diff --git a/packages/core/test/registry.test.ts b/packages/core/test/registry.test.ts index f019c5b..03473a4 100644 --- a/packages/core/test/registry.test.ts +++ b/packages/core/test/registry.test.ts @@ -1,4 +1,6 @@ import { expect, spyOn, test } from "bun:test"; +import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; import { join } from "node:path"; import { loadPlugins } from "../src/registry.ts"; @@ -22,3 +24,21 @@ test("duplicate ids across dirs keep the first", async () => { expect(reg.list()).toHaveLength(1); warn.mockRestore(); }); + +test("reload picks up an edited entry file", async () => { + const root = await mkdtemp(join(tmpdir(), "sttracker-plugins-")); + try { + await mkdir(join(root, "hot")); + const write = (v: string) => + writeFile(join(root, "hot", "index.ts"), `export default { manifest: { id: "hot", name: "Hot", version: "${v}", sdkVersion: 1, mediaTypes: [], capabilities: [], config: [] } };\n`); + await write("1.0.0"); + const reg = await loadPlugins([root]); + expect(reg.get("hot")?.manifest.version).toBe("1.0.0"); + await new Promise((r) => setTimeout(r, 5)); + await write("2.0.0"); + await reg.reload(); + expect(reg.get("hot")?.manifest.version).toBe("2.0.0"); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/plugins/mangabaka/src/api.ts b/plugins/mangabaka/src/api.ts index ea3e5f7..783a82d 100644 --- a/plugins/mangabaka/src/api.ts +++ b/plugins/mangabaka/src/api.ts @@ -1,7 +1,6 @@ import type { PluginContext } from "@sttracker/plugin-sdk/types"; export const BASE_URL = "https://api.mangabaka.org"; -/** The host builds the HttpClient; it should use this rate for this plugin. */ export interface ApiSource { id?: number | string | null } export interface ApiSeries {