From 166b8ecb5e33e6ce99971678a2f099a4ee09eb56 Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 20:39:17 +0100 Subject: [PATCH 1/3] feat(core): jobs, scheduler, plugin registry, connections Co-Authored-By: Claude Sonnet 5.5 --- docs/ARCHITECTURE.md | 2 +- docs/DECISIONS.md | 21 ++ docs/TODO.md | 6 +- packages/core/src/connections.ts | 229 ++++++++++++++++++ packages/core/src/jobs.ts | 168 +++++++++++++ packages/core/src/registry.ts | 63 +++++ packages/core/test/connections.test.ts | 98 ++++++++ .../test/fixtures/plugins/broken/index.ts | 1 + .../core/test/fixtures/plugins/echo/index.ts | 35 +++ packages/core/test/jobs.test.ts | 94 +++++++ packages/core/test/registry.test.ts | 24 ++ packages/plugin-sdk/src/types.ts | 2 +- plugins/hardcover/index.ts | 3 +- plugins/hardcover/src/graphql.ts | 1 - plugins/lastfm/index.ts | 1 + plugins/listenbrainz/index.ts | 1 + plugins/mangabaka/index.ts | 1 + plugins/mangabaka/src/api.ts | 1 - plugins/trakt/index.ts | 1 + plugins/trakt/src/api.ts | 2 - 20 files changed, 744 insertions(+), 10 deletions(-) create mode 100644 packages/core/src/connections.ts create mode 100644 packages/core/src/jobs.ts create mode 100644 packages/core/src/registry.ts create mode 100644 packages/core/test/connections.test.ts create mode 100644 packages/core/test/fixtures/plugins/broken/index.ts create mode 100644 packages/core/test/fixtures/plugins/echo/index.ts create mode 100644 packages/core/test/jobs.test.ts create mode 100644 packages/core/test/registry.test.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 25aead1..b9d3efa 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -16,7 +16,7 @@ user layer or files a Changes-inbox event. |---|---|---| | `apps/server` | `Bun.serve`: `/api/v1`, `/ingest/`, job worker, static web | planned | | `apps/web` | React app (HTML imports) | planned | -| `packages/core` | domain services, one module per area (import by path, no barrel): `auth`, `crypto`, `works`, `matching`, `entries` (entries, library listing, quick add, per-user moves, match suggestions), `engagements`, `history` (`user_history`, restore), `lists`, `copies`, `custom-fields`, `attachments`, `export` (per-user bundle, CSV, restore), `backup` (scheduled bundles + SQLite `VACUUM INTO`), and `sync/` (the change engine: `snapshot` hashing and versions, `apply` per-record diff and apply, `run` = `runSync`, `changes` = Changes inbox `listChanges`/`resolveChange`, `listens` = `ingestListens`) | exists (partial) | +| `packages/core` | domain services, one module per area (import by path, no barrel): `auth`, `crypto`, `works`, `matching`, `entries` (entries, library listing, quick add, per-user moves, match suggestions), `engagements`, `history` (`user_history`, restore), `lists`, `copies`, `custom-fields`, `attachments`, `export` (per-user bundle, CSV, restore), `backup` (scheduled bundles + SQLite `VACUUM INTO`), and `sync/` (the change engine: `snapshot` hashing and versions, `apply` per-record diff and apply, `run` = `runSync`, `changes` = Changes inbox `listChanges`/`resolveChange`, `listens` = `ingestListens`), `jobs` (queue, worker, scheduler), `registry` (`loadPlugins`), `connections` (CRUD, plugin context, `syncHandler`) | exists (partial) | | `packages/db` | Kysely over one Bun.SQL dialect (SQLite + Postgres), ids, JSON helpers, migrations, `forEachDialect` test harness | exists (schema pending) | | `packages/plugin-sdk` | plugin contract, `SourceRecord`, http/rate-limit helpers, fixture test harness | planned | | `plugins/` | one source integration each (hardcover, trakt, mangabaka, listenbrainz, lastfm, csv) | planned | diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index a60f135..a851d38 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -4,6 +4,27 @@ 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: 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`). +**Decision:** +- **Rate limit lives in the manifest** (`PluginManifest.rateLimit?: { perMinute, burst? }`), not in plugin code. The host builds + `ctx.http` from it (default 60/min, burst 5, UA `STTracker/0.1 (+self-hosted)`), so every plugin is throttled the same way + and the per-plugin `RATE_PER_MINUTE` constants are gone. +- **Secrets are split off config.** Config fields typed `"secret"` in the manifest are stored only in the encrypted + `credentials_enc` blob (with OAuth tokens a plugin saves), never in `config`. `makePluginContext` merges them back into + `ctx.config`; `listConnections` returns plain config and `hasCredentials` only. `updateConnection` keeps secrets that are + omitted and replaces plain config wholesale. `credentials.save` merges and persists immediately, so a token refreshed + mid-pull survives a crash. +- **Constructor deviation.** `createConnection`/`updateConnection` take the plugin `manifest` as a trailing argument (needed + for the split), and `makePluginContext` is async (it decrypts) and takes the manifest before an optional `http`. +- **One sync per connection.** Sync jobs carry `lockKey = "sync:"`; `enqueue` returns null while a + `queued|running` job holds the key. On Postgres the check and insert run under `pg_advisory_xact_lock(hashtext(key))`. + Claiming is `update … where status = 'queued'` checked on affected rows. Retries back off `2^attempts` minutes, 5 attempts, then `failed`. +- **Mode choice.** The sync handler runs `full` when asked, when the plugin lacks `pullChanges`, or when no `ok` full run started + within 7 days; else `changes`. A failed run is rethrown so the worker retries it. +**Consequences:** Reload re-imports plugin modules with `?v=`; old module instances are not freed by Bun. + ## 2026-10-30: Export format `sttracker-export/v1`: per-user zip, camelCase JSONL, deleted rows included, no secrets **Status:** accepted **Context:** Task 15 (`packages/core/src/{export,backup}.ts`) fixed the on-disk bundle format that backups and restore share. diff --git a/docs/TODO.md b/docs/TODO.md index 1dfd524..1bc3541 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -24,6 +24,8 @@ 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`. - [ ] 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; @@ -35,8 +37,8 @@ Small, specific, independently actionable. Remove only when actually done. another connection still reports it; consider detaching that connection's `engagement_sources` row instead. - [ ] Sync: pending `changed`/`conflict` events for a value the source has since moved past stay pending; consider superseding them when a newer value for the same field arrives. -- [ ] Sync: two concurrent runs of the same connection can both insert a new `source_records` row for one key - (one fails on the unique key and is logged as a record error). The job worker should lock per connection. +- [ ] Sync: two concurrent runs of the same connection (outside the job queue, e.g. a manual run) can both insert a new + `source_records` row for one key. Queued syncs are serialised by `lockKey`. - [ ] Sync: `ingestListens` does one lookup and one insert per listen; batch it for large backfills. - [ ] Admin merge does not merge colliding children recursively; colliding children stay under the tombstoned parent. - [ ] Matching: a shared parent resolved without external IDs or position (title/year reuse in diff --git a/packages/core/src/connections.ts b/packages/core/src/connections.ts new file mode 100644 index 0000000..10a9b43 --- /dev/null +++ b/packages/core/src/connections.ts @@ -0,0 +1,229 @@ +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 type { HttpClient, PluginContext, PluginManifest } from "@sttracker/plugin-sdk/types"; +import { decryptJson, encryptJson } from "./crypto.ts"; +import { NotFoundError, ValidationError } from "./errors.ts"; +import type { PluginRegistry } from "./registry.ts"; +import { runSync, type Connection, type RunSummary } from "./sync/run.ts"; + +const WEEK_MS = 7 * 24 * 60 * 60 * 1000; + +export interface ConnectionView { + id: string; + pluginId: string; + name: string; + config: Record; + scheduleMinutes: number | null; + status: ConnectionStatus; + lastRunAt: string | null; + createdAt: string; + hasCredentials: boolean; +} + +export interface ConnectionInput { + pluginId: string; + name: string; + config: Record; + credentials?: Record; + scheduleMinutes?: number | null; +} + +type Creds = Record; + +/** Split incoming config: fields typed `"secret"` in the manifest go to the credentials blob, the rest stay config. */ +function splitConfig(manifest: PluginManifest, config: Record): { plain: Record; secrets: Creds } { + const secretKeys = new Set(manifest.config.filter((f) => f.type === "secret").map((f) => f.key)); + const plain: Record = {}; + const secrets: Creds = {}; + for (const [k, v] of Object.entries(config)) { + if (!secretKeys.has(k)) plain[k] = v; + else if (typeof v === "string" && v !== "") secrets[k] = v; + else if (v !== undefined && v !== null && v !== "") throw new ValidationError(k, "secret_must_be_string"); + } + return { plain, secrets }; +} + +async function readCreds(secret: string, enc: string | null): Promise { + return enc === null ? null : decryptJson(secret, enc); +} + +const encOrNull = async (secret: string, c: Creds): Promise => (Object.keys(c).length === 0 ? null : encryptJson(secret, c)); + +function checkSchedule(m: number | null | undefined): void { + if (m !== undefined && m !== null && (!Number.isInteger(m) || m < 1)) throw new ValidationError("scheduleMinutes", "must_be_positive_integer"); +} + +export async function createConnection(db: Db, secret: string, userId: string, input: ConnectionInput, manifest: PluginManifest): Promise { + if (manifest.id !== input.pluginId) throw new ValidationError("pluginId", "manifest_mismatch"); + if (input.name.trim() === "") throw new ValidationError("name", "required"); + checkSchedule(input.scheduleMinutes); + const { plain, secrets } = splitConfig(manifest, input.config); + const id = newId(); + await db.k + .insertInto("source_connections") + .values({ + id, + user_id: userId, + plugin_id: input.pluginId, + name: input.name.trim(), + config: toJson(plain), + credentials_enc: await encOrNull(secret, { ...secrets, ...input.credentials }), + schedule_minutes: input.scheduleMinutes ?? null, + cursor: null, + status: "active", + last_run_at: null, + created_at: nowIso(), + }) + .execute(); + return id; +} + +export interface ConnectionPatch { + name?: string; + config?: Record; + credentials?: Record; + scheduleMinutes?: number | null; + status?: ConnectionStatus; +} + +/** Secrets omitted from `config` are kept; plain config is replaced wholesale. */ +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(); +} + +/** Soft delete: the connection is paused and its credentials are wiped. Records and runs stay. */ +export async function deleteConnection(db: Db, userId: string, id: string): Promise { + const r = await db.k + .updateTable("source_connections") + .set({ status: "paused", credentials_enc: null }) + .where("id", "=", id) + .where("user_id", "=", userId) + .executeTakeFirst(); + if (r.numUpdatedRows !== 1n) throw new NotFoundError("connection"); +} + +/** Never returns secrets: plain config plus `hasCredentials`. */ +export async function listConnections(db: Db, userId: string): Promise { + const rows = await db.k.selectFrom("source_connections").selectAll().where("user_id", "=", userId).orderBy("created_at").execute(); + return rows.map((r) => ({ + id: r.id, + pluginId: r.plugin_id, + name: r.name, + config: fromJson>(r.config, {}), + scheduleMinutes: r.schedule_minutes, + status: r.status, + lastRunAt: r.last_run_at, + createdAt: r.created_at, + hasCredentials: r.credentials_enc !== null, + })); +} + +export const USER_AGENT = "STTracker/0.1 (+self-hosted)"; + +/** + * Build the plugin context for one run. `ctx.http` is rate limited from `manifest.rateLimit` (default 60/min, burst 5) + * unless `http` is given. `credentials.save` merges into the blob and persists it encrypted immediately; + * `cursor.set` persists immediately. + */ +export async function makePluginContext(db: Db, secret: string, conn: Connection, _runId: string, manifest: PluginManifest, http?: HttpClient): Promise { + let creds = await readCreds(secret, conn.credentialsEnc); + 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 ?? {}) }, + 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; + }, + }, + cursor: { + get: async () => cursor, + async set(c) { + await db.k.updateTable("source_connections").set({ cursor: c }).where("id", "=", conn.id).execute(); + cursor = c; + }, + }, + now: () => new Date(), + signal: new AbortController().signal, + }; +} + +export interface SyncHandlerDeps { + db: Db; + secret: string; + registry: PluginRegistry; + http?: HttpClient; +} + +/** + * 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. + */ +export function syncHandler(deps: SyncHandlerDeps): (payload: { connectionId: string; mode?: "full" | "changes" }) => Promise { + return async (payload) => { + 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}`); + 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("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, + pluginId: full.plugin_id, + name: full.name, + config: fromJson>(full.config, {}), + credentialsEnc: full.credentials_enc, + scheduleMinutes: full.schedule_minutes, + cursor: full.cursor, + status: full.status, + 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); + 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 new file mode 100644 index 0000000..3767f53 --- /dev/null +++ b/packages/core/src/jobs.ts @@ -0,0 +1,168 @@ +import type { Db } from "@sttracker/db/db"; +import { newId, nowIso } from "@sttracker/db/ids"; +import { fromJson, toJson } from "@sttracker/db/json"; +import { sql } from "kysely"; + +const MAX_ATTEMPTS = 5; + +export type JobHandler = (payload: any) => Promise; + +/** + * Queue a job. Returns its id, or null when a `queued|running` job already holds `lockKey`. + * On Postgres the check and insert run under an advisory transaction lock so two callers cannot both pass. + */ +export async function enqueue( + db: Db, + type: string, + payload: unknown, + opts: { runAt?: string; lockKey?: string } = {}, +): Promise { + const lockKey = opts.lockKey ?? null; + return db.k.transaction().execute(async (tx) => { + if (lockKey !== null) { + if (db.dialect === "postgres") await sql`select pg_advisory_xact_lock(hashtext(${lockKey}))`.execute(tx); + const held = await tx + .selectFrom("jobs") + .select("id") + .where("lock_key", "=", lockKey) + .where("status", "in", ["queued", "running"]) + .executeTakeFirst(); + if (held) return null; + } + const id = newId(); + const now = nowIso(); + await tx + .insertInto("jobs") + .values({ + id, + type, + payload: toJson(payload ?? null), + run_at: opts.runAt ?? now, + status: "queued", + attempts: 0, + last_error: null, + locked_at: null, + lock_key: lockKey, + created_at: now, + }) + .execute(); + return id; + }); +} + +/** Claim one queued job atomically; true when this caller won it. */ +async function claim(db: Db, id: string): Promise { + const r = await db.k + .updateTable("jobs") + .set({ status: "running", locked_at: nowIso() }) + .where("id", "=", id) + .where("status", "=", "queued") + .executeTakeFirst(); + return r.numUpdatedRows === 1n; +} + +async function runJob(db: Db, handlers: Record, job: { id: string; type: string; payload: string; attempts: number }) { + 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(); + } + } +} + +/** Poll for due jobs and run them with bounded concurrency. Failures retry with backoff `2^attempts` minutes, then `failed`. */ +export function startWorker( + db: Db, + handlers: Record, + opts: { pollMs?: number; concurrency?: number } = {}, +): { stop(): Promise } { + const pollMs = opts.pollMs ?? 1000; + const concurrency = opts.concurrency ?? 2; + const inflight = new Set>(); + let stopped = false; + let timer: ReturnType | undefined; + let tickP: Promise = Promise.resolve(); + + async function tick(): Promise { + while (!stopped && inflight.size < concurrency) { + const due = await db.k + .selectFrom("jobs") + .select(["id", "type", "payload", "attempts"]) + .where("status", "=", "queued") + .where("run_at", "<=", nowIso()) + .orderBy("run_at") + .limit(concurrency - inflight.size) + .execute(); + let started = 0; + for (const job of due) { + if (stopped || inflight.size >= concurrency) break; + if (!(await claim(db, job.id))) continue; + started++; + const p = runJob(db, handlers, job).finally(() => inflight.delete(p)); + inflight.add(p); + } + if (started === 0) break; + } + } + + function schedule() { + if (stopped) return; + timer = setTimeout(() => { + tickP = tick().catch(() => {}).finally(schedule); + }, pollMs); + } + tickP = tick().catch(() => {}).finally(schedule); + + return { + async stop() { + stopped = true; + if (timer) clearTimeout(timer); + await tickP; + await Promise.all(inflight); + }, + }; +} + +/** Every `tickMs`, enqueue a `sync` job for each active 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; + async function tick() { + const rows = await db.k + .selectFrom("source_connections") + .select(["id", "schedule_minutes", "last_run_at"]) + .where("status", "=", "active") + .where("schedule_minutes", "is not", null) + .execute(); + const now = Date.now(); + for (const r of rows) { + if (stopped) return; + const due = r.last_run_at === null || Date.parse(r.last_run_at) + r.schedule_minutes! * 60_000 <= now; + if (due) await enqueue(db, "sync", { connectionId: r.id }, { lockKey: `sync:${r.id}` }); + } + } + const run = () => tick().catch(() => {}); + void run(); + const timer = setInterval(() => void run(), tickMs); + return { + /** Run one scheduling pass now (tests, manual triggers). */ + tickNow: run, + stop() { + stopped = true; + clearInterval(timer); + }, + }; +} diff --git a/packages/core/src/registry.ts b/packages/core/src/registry.ts new file mode 100644 index 0000000..32d26ee --- /dev/null +++ b/packages/core/src/registry.ts @@ -0,0 +1,63 @@ +import { readdir } from "node:fs/promises"; +import { join } from "node:path"; +import { pathToFileURL } from "node:url"; +import type { PluginManifest, SourcePlugin } from "@sttracker/plugin-sdk/types"; + +export interface PluginRegistry { + get(id: string): SourcePlugin | undefined; + list(): PluginManifest[]; + reload(): Promise; +} + +function validShape(p: unknown): p is SourcePlugin { + const m = (p as SourcePlugin | undefined)?.manifest; + return ( + !!m && + typeof m.id === "string" && + m.id.trim() !== "" && + m.sdkVersion === 1 && + typeof m.name === "string" && + Array.isArray(m.capabilities) && + Array.isArray(m.config) + ); +} + +async function scan(dirs: string[]): Promise> { + const found = new Map(); + const stamp = Date.now(); + for (const dir of dirs) { + let names: string[]; + try { + names = (await readdir(dir, { withFileTypes: true })).filter((d) => d.isDirectory()).map((d) => d.name).sort(); + } catch (e) { + console.warn(`plugins: cannot read ${dir}: ${e instanceof Error ? e.message : String(e)}`); + continue; + } + for (const name of names) { + const file = join(dir, name, "index.ts"); + if (!(await Bun.file(file).exists())) continue; + try { + 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; + if (found.has(id)) throw new Error(`duplicate plugin id "${id}"`); + found.set(id, mod.default); + } catch (e) { + console.warn(`plugins: skipping ${file}: ${e instanceof Error ? e.message : String(e)}`); + } + } + } + return found; +} + +/** Load every `//index.ts` plugin. A bad plugin is logged and skipped, never fatal. */ +export async function loadPlugins(dirs: string[]): Promise { + let plugins = await scan(dirs); + return { + get: (id) => plugins.get(id), + list: () => [...plugins.values()].map((p) => p.manifest), + async reload() { + plugins = await scan(dirs); + }, + }; +} diff --git a/packages/core/test/connections.test.ts b/packages/core/test/connections.test.ts new file mode 100644 index 0000000..b3de7a5 --- /dev/null +++ b/packages/core/test/connections.test.ts @@ -0,0 +1,98 @@ +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 { createConnection, deleteConnection, listConnections, makePluginContext, syncHandler, updateConnection } from "../src/connections.ts"; +import { loadPlugins } from "../src/registry.ts"; +import { makeUser } from "./sync/helpers.ts"; +import { newId, nowIso } from "@sttracker/db/ids"; +import type { Connection } from "../src/sync/run.ts"; + +const SECRET = "test-secret"; +const warn = spyOn(console, "warn").mockImplementation(() => {}); +const reg = await loadPlugins([join(import.meta.dir, "fixtures/plugins")]); +warn.mockRestore(); +const echo = reg.get("echo")!; + +forEachDialect("connections", (getDb) => { + const raw = (id: string) => getDb().k.selectFrom("source_connections").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + const asConn = async (id: string): Promise => { + const r = await raw(id); + return { id, userId: r.user_id, pluginId: r.plugin_id, name: r.name, config: JSON.parse(r.config), credentialsEnc: r.credentials_enc, scheduleMinutes: null, cursor: r.cursor, status: r.status, lastRunAt: null, createdAt: r.created_at }; + }; + + test("secrets live only in the encrypted blob and never in list", async () => { + const db = getDb(); + const u = await makeUser(db); + const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: { label: "hi", token: "s3cret" } }, echo.manifest); + const r = await raw(id); + expect(JSON.parse(r.config)).toEqual({ label: "hi" }); + expect(r.credentials_enc).not.toContain("s3cret"); + expect(await decryptJson(SECRET, r.credentials_enc!)).toEqual({ token: "s3cret" }); + const [v] = await listConnections(db, u); + expect(JSON.stringify(v)).not.toContain("s3cret"); + expect(v).toMatchObject({ id, config: { label: "hi" }, hasCredentials: true }); + const ctx = await makePluginContext(db, SECRET, await asConn(id), "r", echo.manifest); + expect(ctx.config).toEqual({ label: "hi", token: "s3cret" }); + expect(await ctx.credentials.get()).toEqual({ token: "s3cret" }); + }); + + test("update keeps omitted secrets; delete pauses and clears credentials", async () => { + const db = getDb(); + const u = await makeUser(db); + const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: { token: "t1" } }, echo.manifest); + await updateConnection(db, SECRET, u, id, { config: { label: "new" }, scheduleMinutes: 30 }, echo.manifest); + let r = await raw(id); + expect(JSON.parse(r.config)).toEqual({ label: "new" }); + expect(r.schedule_minutes).toBe(30); + expect(await decryptJson(SECRET, r.credentials_enc!)).toEqual({ token: "t1" }); + await updateConnection(db, SECRET, u, id, { config: { token: "t2" } }, echo.manifest); + expect(await decryptJson(SECRET, (await raw(id)).credentials_enc!)).toEqual({ token: "t2" }); + const other = await makeUser(db, "bob"); + await expect(deleteConnection(db, other, id)).rejects.toThrow(); + await deleteConnection(db, u, id); + r = await raw(id); + expect(r.status).toBe("paused"); + expect(r.credentials_enc).toBeNull(); + expect((await listConnections(db, u))[0]!.hasCredentials).toBe(false); + }); + + test("cursor persists on set; http uses manifest rate limit", async () => { + const db = getDb(); + const u = await makeUser(db); + const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: {} }, echo.manifest); + const ctx = await makePluginContext(db, SECRET, await asConn(id), "r", echo.manifest); + await ctx.cursor.set("c1"); + expect((await raw(id)).cursor).toBe("c1"); + expect(await ctx.cursor.get()).toBe("c1"); + }); + + test("credentials.save persists encrypted before pull continues", async () => { + const db = getDb(); + const u = await makeUser(db); + const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: { label: "explode", token: "t" } }, echo.manifest); + const handler = syncHandler({ db, secret: SECRET, registry: reg }); + await expect(handler({ connectionId: id })).rejects.toThrow(); + const r = await raw(id); + expect(await decryptJson(SECRET, r.credentials_enc!)).toEqual({ token: "t", refresh: "new-refresh" }); + expect(r.status).toBe("error"); + }); + + test("sync handler: full first, changes after a recent full, full when payload says so or full is stale", async () => { + const db = getDb(); + const u = await makeUser(db); + const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: { label: "ok" } }, echo.manifest); + const handler = syncHandler({ db, secret: SECRET, registry: reg }); + const modes = async () => (await db.k.selectFrom("source_runs").select(["mode", "status"]).where("connection_id", "=", id).orderBy("started_at").orderBy("id").execute()).map((r) => `${r.mode}:${r.status}`); + await handler({ connectionId: id }); + await handler({ connectionId: id }); + await handler({ connectionId: id, mode: "full" }); + expect(await modes()).toEqual(["full:ok", "changes:ok", "full:ok"]); + await db.k.updateTable("source_runs").set({ started_at: new Date(Date.now() - 8 * 86400_000).toISOString() }).where("connection_id", "=", id).execute(); + await handler({ connectionId: id }); + expect((await modes()).at(-1)).toBe("full:ok"); + await db.k.insertInto("source_runs").values({ id: newId(), connection_id: id, mode: "full", started_at: nowIso(), finished_at: null, status: "failed", counts: "{}", log: "", error: "x" }).execute(); + await handler({ connectionId: id }); + expect((await modes()).at(-1)).toBe("changes:ok"); + }); +}); diff --git a/packages/core/test/fixtures/plugins/broken/index.ts b/packages/core/test/fixtures/plugins/broken/index.ts new file mode 100644 index 0000000..413bce7 --- /dev/null +++ b/packages/core/test/fixtures/plugins/broken/index.ts @@ -0,0 +1 @@ +export default { nope: true }; diff --git a/packages/core/test/fixtures/plugins/echo/index.ts b/packages/core/test/fixtures/plugins/echo/index.ts new file mode 100644 index 0000000..e0ed7df --- /dev/null +++ b/packages/core/test/fixtures/plugins/echo/index.ts @@ -0,0 +1,35 @@ +import { definePlugin, type SourceRecord } from "@sttracker/plugin-sdk/types"; + +export default definePlugin({ + manifest: { + id: "echo", + name: "Echo", + version: "0.0.1", + sdkVersion: 1, + mediaTypes: ["book"], + capabilities: ["pull", "pullChanges"], + config: [ + { key: "label", label: "Label", type: "string" }, + { key: "token", label: "Token", type: "secret" }, + ], + rateLimit: { perMinute: 30, burst: 2 }, + }, + async *pull(ctx) { + const cfg = ctx.config; + if (cfg.label === "explode") { + await ctx.credentials.save({ refresh: "new-refresh" }); + throw new Error("boom after save"); + } + const rec: SourceRecord = { + kind: "entry", + key: "echo-1", + raw: { label: cfg.label ?? null }, + work: { title: String(cfg.label ?? "Echo"), mediaType: "book", externalIds: { isbn13: "9780000000002" } }, + entry: { status: "completed" }, + }; + yield rec; + }, + async *pullChanges(ctx) { + yield* this.pull!(ctx); + }, +}); diff --git a/packages/core/test/jobs.test.ts b/packages/core/test/jobs.test.ts new file mode 100644 index 0000000..1e50b0f --- /dev/null +++ b/packages/core/test/jobs.test.ts @@ -0,0 +1,94 @@ +import { expect, test } from "bun:test"; +import { forEachDialect } from "@sttracker/db/testing"; +import { enqueue, startScheduler, startWorker } from "../src/jobs.ts"; +import { makeConnection, makeUser } from "./sync/helpers.ts"; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +forEachDialect("jobs", (getDb) => { + test("enqueue with same lockKey returns null", async () => { + const db = getDb(); + expect(await enqueue(db, "sync", {}, { lockKey: "sync:a" })).toBeString(); + expect(await enqueue(db, "sync", {}, { lockKey: "sync:a" })).toBeNull(); + expect(await enqueue(db, "sync", {}, { lockKey: "sync:b" })).toBeString(); + expect(await enqueue(db, "x", {})).toBeString(); + expect(await enqueue(db, "x", {})).toBeString(); + }); + + test("concurrent enqueue with one lockKey admits exactly one", async () => { + const db = getDb(); + const ids = await Promise.all(Array.from({ length: 6 }, () => enqueue(db, "sync", {}, { lockKey: "k" }))); + expect(ids.filter((i) => i !== null)).toHaveLength(1); + }); + + test("worker runs a job and releases the lock", async () => { + const db = getDb(); + const seen: unknown[] = []; + const id = await enqueue(db, "t", { n: 1 }, { lockKey: "l" }); + const w = startWorker(db, { t: async (p) => void seen.push(p) }, { pollMs: 10 }); + await sleep(150); + await w.stop(); + expect(seen).toEqual([{ n: 1 }]); + expect((await db.k.selectFrom("jobs").selectAll().where("id", "=", id!).executeTakeFirstOrThrow()).status).toBe("done"); + expect(await enqueue(db, "t", {}, { lockKey: "l" })).toBeString(); + }); + + test("a job is claimed by only one of two workers", async () => { + const db = getDb(); + let runs = 0; + const h = { t: async () => { runs++; await sleep(30); } }; + await enqueue(db, "t", {}); + const a = startWorker(db, h, { pollMs: 5 }); + const b = startWorker(db, h, { pollMs: 5 }); + await sleep(200); + await Promise.all([a.stop(), b.stop()]); + expect(runs).toBe(1); + }); + + test("worker retries then fails", async () => { + const db = getDb(); + let calls = 0; + const id = (await enqueue(db, "bad", {}))!; + const w = startWorker(db, { bad: async () => { calls++; throw new Error("nope"); } }, { pollMs: 10 }); + const row = () => db.k.selectFrom("jobs").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + for (let attempt = 1; attempt <= 5; attempt++) { + for (let i = 0; i < 100 && (await row()).attempts < attempt; i++) await sleep(10); + const r = await row(); + expect(r.attempts).toBe(attempt); + expect(r.last_error).toBe("nope"); + if (attempt < 5) { + expect(r.status).toBe("queued"); + const delay = Date.parse(r.run_at) - Date.now(); + expect(delay).toBeGreaterThan(2 ** attempt * 60_000 - 5_000); + expect(delay).toBeLessThanOrEqual(2 ** attempt * 60_000); + await db.k.updateTable("jobs").set({ run_at: new Date(Date.now() - 1000).toISOString() }).where("id", "=", id).execute(); + } + } + await w.stop(); + expect((await row()).status).toBe("failed"); + expect(calls).toBe(5); + }); + + test("scheduler enqueues due connections only", async () => { + const db = getDb(); + const u = await makeUser(db); + const due = await makeConnection(db, u, "p"); + const fresh = await makeConnection(db, u, "p"); + const manual = 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(paused, { schedule_minutes: 60, last_run_at: old, status: "paused" }); + void manual; + const s = startScheduler(db, { tickMs: 3_600_000 }); + await s.tickNow(); + 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 }); + }); +}); diff --git a/packages/core/test/registry.test.ts b/packages/core/test/registry.test.ts new file mode 100644 index 0000000..f019c5b --- /dev/null +++ b/packages/core/test/registry.test.ts @@ -0,0 +1,24 @@ +import { expect, spyOn, test } from "bun:test"; +import { join } from "node:path"; +import { loadPlugins } from "../src/registry.ts"; + +const dir = join(import.meta.dir, "fixtures/plugins"); + +test("registry loads echo plugin and skips broken one", async () => { + const warn = spyOn(console, "warn").mockImplementation(() => {}); + const reg = await loadPlugins([dir, join(dir, "does-not-exist")]); + expect(reg.list().map((m) => m.id)).toEqual(["echo"]); + expect(reg.get("echo")?.manifest.name).toBe("Echo"); + expect(reg.get("broken")).toBeUndefined(); + expect(warn.mock.calls.some((c) => String(c[0]).includes("broken"))).toBe(true); + await reg.reload(); + expect(reg.list()).toHaveLength(1); + warn.mockRestore(); +}); + +test("duplicate ids across dirs keep the first", async () => { + const warn = spyOn(console, "warn").mockImplementation(() => {}); + const reg = await loadPlugins([dir, dir]); + expect(reg.list()).toHaveLength(1); + warn.mockRestore(); +}); diff --git a/packages/plugin-sdk/src/types.ts b/packages/plugin-sdk/src/types.ts index d57df1d..0ea6101 100644 --- a/packages/plugin-sdk/src/types.ts +++ b/packages/plugin-sdk/src/types.ts @@ -7,7 +7,7 @@ export type Precision = "day" | "month" | "year" | "unknown"; export interface PartialDate { value: string; precision: Precision } export type Capability = "pull" | "pullChanges" | "push" | "ownedLibrary" | "ingest" | "lookup" | "importFile" | "exportFile"; export interface ConfigField { key: string; label: string; type: "string" | "secret" | "number" | "boolean" | "select"; required?: boolean; options?: string[]; help?: string } -export interface PluginManifest { id: string; name: string; version: string; sdkVersion: 1; mediaTypes: MediaType[]; capabilities: Capability[]; config: ConfigField[]; attribution?: string } +export interface PluginManifest { id: string; name: string; version: string; sdkVersion: 1; mediaTypes: MediaType[]; capabilities: Capability[]; config: ConfigField[]; attribution?: string; rateLimit?: { perMinute: number; burst?: number } } export interface WorkHint { title: string; mediaType: MediaType; year?: number; creators?: { name: string; role?: string }[]; externalIds: Record; coverUrl?: string; description?: string; parent?: WorkHint; position?: { season?: number; episode?: number; index?: number } } export interface EntryFields { status?: Status; rating?: number | null; review?: string | null; notes?: string | null; favorite?: boolean; tags?: string[] } export interface EngagementFields { startedAt?: PartialDate | null; finishedAt?: PartialDate | null; progress?: Record; platform?: string | null; rating?: number | null; notes?: string | null } diff --git a/plugins/hardcover/index.ts b/plugins/hardcover/index.ts index ed3272d..2298b95 100644 --- a/plugins/hardcover/index.ts +++ b/plugins/hardcover/index.ts @@ -4,8 +4,6 @@ import { gql, LISTS_QUERY, PAGE_SIZE, SEARCH_QUERY, USER_BOOKS_QUERY } from "./s import { mapList, mapUserBook } from "./src/map.ts"; import type { HcList, HcUserBook } from "./src/map.ts"; -export { RATE_PER_MINUTE } from "./src/graphql.ts"; - async function* pageUserBooks(ctx: PluginContext, where: Record | null): AsyncGenerator { for (let offset = 0; ; offset += PAGE_SIZE) { const d = await gql<{ me: { user_books: HcUserBook[] }[] }>(ctx, USER_BOOKS_QUERY, { limit: PAGE_SIZE, offset, where }); @@ -45,6 +43,7 @@ export default definePlugin({ name: "Hardcover", version: "0.1.0", sdkVersion: 1, + rateLimit: { perMinute: 50 }, mediaTypes: ["book"], capabilities: ["pull", "pullChanges", "lookup"], config: [{ key: "token", label: "Personal access token", type: "secret", required: true }], diff --git a/plugins/hardcover/src/graphql.ts b/plugins/hardcover/src/graphql.ts index 959e16b..652d705 100644 --- a/plugins/hardcover/src/graphql.ts +++ b/plugins/hardcover/src/graphql.ts @@ -2,7 +2,6 @@ import { HttpError } from "@sttracker/plugin-sdk/http"; import type { PluginContext } from "@sttracker/plugin-sdk/types"; export const ENDPOINT = "https://api.hardcover.app/v1/graphql"; -export const RATE_PER_MINUTE = 50; export const PAGE_SIZE = 100; const BOOK = `id title release_year contributions { author { name } }`; diff --git a/plugins/lastfm/index.ts b/plugins/lastfm/index.ts index e22470e..fbda414 100644 --- a/plugins/lastfm/index.ts +++ b/plugins/lastfm/index.ts @@ -7,6 +7,7 @@ export default definePlugin({ name: "Last.fm", version: "0.1.0", sdkVersion: 1, + rateLimit: { perMinute: 240 }, mediaTypes: ["track", "album", "artist"], capabilities: ["pull", "pullChanges"], config: [ diff --git a/plugins/listenbrainz/index.ts b/plugins/listenbrainz/index.ts index cc8f901..7c143dd 100644 --- a/plugins/listenbrainz/index.ts +++ b/plugins/listenbrainz/index.ts @@ -7,6 +7,7 @@ export default definePlugin({ name: "ListenBrainz", version: "0.1.0", sdkVersion: 1, + rateLimit: { perMinute: 60 }, mediaTypes: ["track", "album", "artist"], capabilities: ["pull", "pullChanges"], config: [ diff --git a/plugins/mangabaka/index.ts b/plugins/mangabaka/index.ts index 45e59c4..fde54c3 100644 --- a/plugins/mangabaka/index.ts +++ b/plugins/mangabaka/index.ts @@ -8,6 +8,7 @@ export default definePlugin({ name: "MangaBaka", version: "0.1.0", sdkVersion: 1, + rateLimit: { perMinute: 150 }, mediaTypes: ["manga", "comic"], capabilities: ["pull", "pullChanges", "lookup"], config: [{ key: "token", label: "API token", type: "secret", required: true, help: "Sent as the x-api-key header." }], diff --git a/plugins/mangabaka/src/api.ts b/plugins/mangabaka/src/api.ts index 5707c33..ea3e5f7 100644 --- a/plugins/mangabaka/src/api.ts +++ b/plugins/mangabaka/src/api.ts @@ -2,7 +2,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 const RATE_PER_MINUTE = 150; export interface ApiSource { id?: number | string | null } export interface ApiSeries { diff --git a/plugins/trakt/index.ts b/plugins/trakt/index.ts index ea363a4..21c1565 100644 --- a/plugins/trakt/index.ts +++ b/plugins/trakt/index.ts @@ -12,6 +12,7 @@ export default definePlugin({ name: "Trakt", version: "0.1.0", sdkVersion: 1, + rateLimit: { perMinute: 100 }, mediaTypes: ["movie", "show", "season", "episode"], capabilities: ["pull", "pullChanges", "lookup"], config: [ diff --git a/plugins/trakt/src/api.ts b/plugins/trakt/src/api.ts index fa6787e..112e749 100644 --- a/plugins/trakt/src/api.ts +++ b/plugins/trakt/src/api.ts @@ -2,8 +2,6 @@ import { HttpError } from "@sttracker/plugin-sdk/http"; import type { PluginContext } from "@sttracker/plugin-sdk/types"; export const BASE = "https://api.trakt.tv"; -/** Rate the host should configure for this plugin's HttpClient. */ -export const RATE_PER_MINUTE = 100; const REFRESH_WINDOW_MS = 24 * 60 * 60 * 1000; const REDIRECT_URI = "urn:ietf:wg:oauth:2.0:oob"; From 6f628c4693c63a7f5ec5009744f5a89a88aa160a Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 20:52:51 +0100 Subject: [PATCH 2/3] 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 { From f0d41f23c8f7a98465b5b75221b52b56ae7976fb Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 20:57:56 +0100 Subject: [PATCH 3/3] fix(core): reaper skips own in-flight jobs, heartbeat, no attempt burned on shutdown Co-Authored-By: Claude Sonnet 5.5 --- packages/core/src/jobs.ts | 37 +++++++++++++++++++++++++-------- packages/core/test/jobs.test.ts | 35 +++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 9 deletions(-) diff --git a/packages/core/src/jobs.ts b/packages/core/src/jobs.ts index 081a123..4620857 100644 --- a/packages/core/src/jobs.ts +++ b/packages/core/src/jobs.ts @@ -6,6 +6,7 @@ import { sql } from "kysely"; const MAX_ATTEMPTS = 5; const STALE_LOCK_MS = 60 * 60 * 1000; +const HEARTBEAT_MS = 5 * 60 * 1000; export interface JobContext { /** Aborted when the worker is stopped. */ @@ -78,7 +79,11 @@ 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 }, signal: AbortSignal): Promise { +async function runJob(db: Db, handlers: Record, job: { id: string; type: string; payload: string; attempts: number }, signal: AbortSignal, heartbeatMs: number): Promise { + // Keep `locked_at` fresh so another process's reaper does not steal a long-running job. + const beat = setInterval(() => { + db.k.updateTable("jobs").set({ locked_at: nowIso() }).where("id", "=", job.id).where("status", "=", "running").execute().catch(logErr); + }, heartbeatMs); try { try { const h = handlers[job.type]; @@ -88,7 +93,10 @@ async function runJob(db: Db, handlers: Record, job: { id: s } catch (e) { const attempts = job.attempts + 1; const err = e instanceof Error ? e.message : String(e); - if (e instanceof DropJobError) { + if (signal.aborted && !(e instanceof DropJobError) && !(e instanceof NonRetryableError)) { + // Interrupted by shutdown: back to the queue untouched, no attempt burned, no backoff. + await db.k.updateTable("jobs").set({ status: "queued", locked_at: null }).where("id", "=", job.id).execute(); + } else 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(); @@ -103,31 +111,38 @@ async function runJob(db: Db, handlers: Record, job: { id: s } } catch (e) { logErr(e); + } finally { + clearInterval(beat); } } /** 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(); +async function reapStale(db: Db, staleMs: number, mine: Set): Promise { + const cutoff = new Date(Date.now() - staleMs).toISOString(); + let q = db.k.updateTable("jobs").set({ status: "queued", locked_at: null }).where("status", "=", "running").where("locked_at", "<", cutoff); + if (mine.size > 0) q = q.where("id", "not in", [...mine]); + await q.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, handlers: Record, - opts: { pollMs?: number; concurrency?: number } = {}, + opts: { pollMs?: number; concurrency?: number; staleMs?: number; heartbeatMs?: number } = {}, ): { stop(): Promise } { const pollMs = opts.pollMs ?? 1000; const concurrency = opts.concurrency ?? 2; + const staleMs = opts.staleMs ?? STALE_LOCK_MS; + const heartbeatMs = opts.heartbeatMs ?? HEARTBEAT_MS; const inflight = new Set>(); + const mine = 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); + await reapStale(db, staleMs, mine); while (!stopped && inflight.size < concurrency) { const due = await db.k .selectFrom("jobs") @@ -142,9 +157,13 @@ export function startWorker( if (stopped || inflight.size >= concurrency) break; if (!(await claim(db, job.id))) continue; started++; - const p: Promise = runJob(db, handlers, job, abort.signal) + mine.add(job.id); + const p: Promise = runJob(db, handlers, job, abort.signal, heartbeatMs) .catch(logErr) - .finally(() => inflight.delete(p)); + .finally(() => { + inflight.delete(p); + mine.delete(job.id); + }); inflight.add(p); } if (started === 0) break; diff --git a/packages/core/test/jobs.test.ts b/packages/core/test/jobs.test.ts index 70049f6..c17c465 100644 --- a/packages/core/test/jobs.test.ts +++ b/packages/core/test/jobs.test.ts @@ -145,6 +145,41 @@ forEachDialect("jobs", (getDb) => { err.mockRestore(); }); + test("a job running past staleMs is not reaped by its own worker and runs once", async () => { + const db = getDb(); + const id = (await enqueue(db, "t", {}, { lockKey: "slow" }))!; + let runs = 0; + const w = startWorker(db, { t: async () => { runs++; await sleep(300); } }, { pollMs: 10, staleMs: 50, heartbeatMs: 20 }); + await until(async () => (await row(db, id)).status === "done"); + await w.stop(); + expect(runs).toBe(1); + }); + + test("heartbeat refreshes locked_at", async () => { + const db = getDb(); + const id = (await enqueue(db, "t", {}))!; + let release!: () => void; + const gate = new Promise((r) => (release = r)); + const w = startWorker(db, { t: () => gate }, { pollMs: 10, heartbeatMs: 20 }); + await until(async () => (await row(db, id)).status === "running"); + const first = (await row(db, id)).locked_at!; + await until(async () => (await row(db, id)).locked_at !== first); + expect((await row(db, id)).locked_at).not.toBe(first); + release(); + await w.stop(); + }); + + test("stop() re-queues an aborted job without burning an attempt or backoff", async () => { + const db = getDb(); + const id = (await enqueue(db, "t", {}))!; + const before = (await row(db, id)).run_at; + let started = false; + const w = startWorker(db, { t: (_p, { signal }) => new Promise((_, rej) => { started = true; signal.addEventListener("abort", () => rej(new Error("aborted"))); }) }, { pollMs: 10 }); + await until(async () => started); + await w.stop(); + expect(await row(db, id)).toMatchObject({ status: "queued", attempts: 0, run_at: before, locked_at: null }); + }); + test("stop() aborts the signal handed to handlers", async () => { const db = getDb(); await enqueue(db, "t", {});