fix(core): job failure semantics, stale-lock reaper, run-scoped context, reload

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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<void> {
-  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<string, unknown> = {};
-  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<string, unknown> = {};
+    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<Connectio
 
 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<PluginContext> {
-  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<HttpClientOptions, "fetchImpl" | "sleep" | "now">;
+  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<PluginContext> {
+  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<never> {
+  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<RunSummary | void> {
-  return async (payload) => {
+export function syncHandler(deps: SyncHandlerDeps): (payload: { connectionId: string; mode?: "full" | "changes" }, jc?: JobContext) => Promise<RunSummary | void> {
+  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;
   };