Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297// The write-queue runner (KTD-4, KTD-5). A lane is one (owner, app) pair and// runs strictly head-of-line: its oldest non-terminal row goes first, and a// row waiting on a retry or a rate-limit park blocks its lane rather than// being overtaken. That is what keeps an unfollow and a later re-follow of the// same subject in event order.
import { and, eq, inArray, isNull, lt, or, sql } from "drizzle-orm";import { conn, outsideTransaction } from "../db/tx.js";import { syncQueue, syncSets } from "../db/schema.js";import { executeRow, type ExecuteOutcome } from "./execute.js";import { canSpend, spentPoints, type SpentPoints } from "./pacer.js";import { hasOAuthSession, pauseSet, recordAuthFailure, resetAuthFailures } from "./pause.js";import { appendActivity, getSet, isAppSeeded, type QueueRow, type SyncSet } from "./store.js";import { POINTS } from "./targets.js";import { registerSyncWake } from "./wake.js";
/** Fallback tick; the Jetstream handler and the API wake the runner sooner. */const FALLBACK_TICK_MS = 5_000;/** Lanes executing at once, one row each. Same bound and rationale as the * scheduled-action runner: database pool size and outbound fan-out. */export const LANE_CONCURRENCY = 8;/** Rows per tick, which also bounds profile reads per tick. */const MAX_ROWS_PER_TICK = 64;/** A row claimed longer than this is treated as orphaned and re-queued. Kept * far above the PDS call bound (30s session restore plus 30s write, see * PDS_FETCH_TIMEOUT_MS in lib/automations/pds.ts) so a reclaim can never * race a write that is still in flight. */export const ORPHAN_TIMEOUT_MS = 10 * 60 * 1000;export const MAX_ATTEMPTS = 5;const RETRY_BACKOFF_MS = 30_000;const AUTH_PARK_MS = 60_000;const DEFAULT_RATE_LIMIT_PARK_MS = 15 * 60 * 1000;const SHUTDOWN_WAIT_MIN_MS = 1_000;
type Claimed = { row: QueueRow; set: SyncSet };
async function resetOrphans(now: Date): Promise<void> { await conn() .update(syncQueue) .set({ status: "pending" }) .where( and( eq(syncQueue.status, "running"), or( lt(syncQueue.claimedAt, new Date(now.getTime() - ORPHAN_TIMEOUT_MS)), isNull(syncQueue.claimedAt), ), ), );}
async function laneHeads(): Promise<QueueRow[]> { const heads = ( await conn() .select({ id: sql<string>`min(${syncQueue.id})` }) .from(syncQueue) .where(inArray(syncQueue.status, ["pending", "running"])) .groupBy(syncQueue.did, syncQueue.app) ).map((h) => Number(h.id)); if (heads.length === 0) return []; return conn().select().from(syncQueue).where(inArray(syncQueue.id, heads)).orderBy(syncQueue.id);}
async function claimLanes( now: Date, spent: Map<string, number>, limit: number,): Promise<Claimed[]> { const claimed: Claimed[] = []; const sets = new Map<string, SyncSet | undefined>(); const sessions = new Map<string, boolean>(); const seeded = new Map<string, boolean>(); const budgets = new Map<string, SpentPoints>();
for (const row of await laneHeads()) { if (claimed.length >= limit) break; // A running head is in flight; a future due time is a retry or a park. if (row.status !== "pending" || row.dueAt > now) continue;
if (!sets.has(row.did)) sets.set(row.did, await getSet(row.did)); const set = sets.get(row.did); if (!set || !set.enabled || !set.apps.includes(row.app)) { await conn() .update(syncQueue) .set({ status: "cancelled", finishedAt: now }) .where(eq(syncQueue.id, row.id)); continue; } if (set.pausedAt) continue; if (set.parkedUntil && set.parkedUntil > now) continue; if (!sessions.has(row.did)) sessions.set(row.did, await hasOAuthSession(row.did)); if (!sessions.get(row.did)) { await pauseSet(row.did, now, "session_missing"); continue; } // Seed-before-write gate: never follow on an app whose follows are not // listed yet, or a backfill could duplicate a follow made by hand. if (row.kind === "follow") { const lane = `${row.did}|${row.app}`; if (!seeded.has(lane)) seeded.set(lane, await isAppSeeded(row.did, row.app)); if (!seeded.get(lane)) continue; }
const cost = row.kind === "follow" ? POINTS.create : POINTS.delete; let budget = budgets.get(row.did); if (!budget) { budget = await spentPoints(row.did, now); budgets.set(row.did, budget); } if (!canSpend(budget, cost, spent.get(row.did) ?? 0)) continue;
const ok = await conn() .update(syncQueue) .set({ status: "running", claimedAt: now, attempts: row.attempts + 1 }) .where(and(eq(syncQueue.id, row.id), eq(syncQueue.status, "pending"))) .returning({ id: syncQueue.id }); if (ok.length === 0) continue; spent.set(row.did, (spent.get(row.did) ?? 0) + cost); claimed.push({ row: { ...row, status: "running", attempts: row.attempts + 1 }, set }); } return claimed;}
async function finish( row: QueueRow, status: QueueRow["status"], now: Date, lastError: string | null = null,): Promise<void> { await conn() .update(syncQueue) .set({ status, finishedAt: now, lastError }) .where(eq(syncQueue.id, row.id));}
async function requeue( row: QueueRow, dueAt: Date, attempts: number, lastError: string,): Promise<void> { await conn() .update(syncQueue) .set({ status: "pending", dueAt, attempts, lastError, claimedAt: null }) .where(eq(syncQueue.id, row.id));}
async function applyOutcome(row: QueueRow, outcome: ExecuteOutcome, now: Date): Promise<void> { const base = { did: row.did, app: row.app, source: row.source, subject: row.subject, createdAt: now, }; switch (outcome.kind) { case "written": case "deleted": await finish(row, "done", now); await appendActivity({ ...base, outcome: outcome.kind, points: outcome.points }); await resetAuthFailures(row.did); return; case "skipped": await finish(row, "skipped", now); await appendActivity({ ...base, outcome: "skipped", reason: outcome.reason }); return; case "waiting": await finish(row, "waiting", now); await appendActivity({ ...base, outcome: "skipped", reason: "no_profile" }); await resetAuthFailures(row.did); return; case "failed": { const { failure } = outcome; // Parks do not consume an attempt: waiting out a session or a rate // limit is not the row failing. const previousAttempts = row.attempts - 1; if (failure.kind === "auth") { await requeue( row, new Date(now.getTime() + AUTH_PARK_MS), previousAttempts, failure.message, ); await recordAuthFailure(row.did, now); return; } if (failure.kind === "rate-limited") { const until = new Date(failure.resetAt ?? now.getTime() + DEFAULT_RATE_LIMIT_PARK_MS); await requeue(row, until, previousAttempts, failure.message); await conn().update(syncSets).set({ parkedUntil: until }).where(eq(syncSets.did, row.did)); return; } if (failure.kind === "retryable" && row.attempts < MAX_ATTEMPTS) { await requeue( row, new Date(now.getTime() + RETRY_BACKOFF_MS * row.attempts), row.attempts, failure.message, ); return; } await finish(row, "failed", now, failure.message); await appendActivity({ ...base, outcome: "failed", reason: failure.message.slice(0, 500) }); return; } }}
async function runClaimed({ row, set }: Claimed, now: Date): Promise<void> { let outcome: ExecuteOutcome; try { outcome = await executeRow(row, set, now); } catch (err) { outcome = { kind: "failed", failure: { kind: "retryable", message: err instanceof Error ? err.message : String(err) }, }; } await applyOutcome(row, outcome, now);}
/** One tick: reset orphans, then claim and run lane heads in rounds until * nothing is claimable or the per-tick cap is reached. Exported for tests; * production drives it through `startSyncRunner`. */export async function runSyncQueue(clock: () => Date = () => new Date()): Promise<number> { await resetOrphans(clock()); const spent = new Map<string, number>(); let processed = 0; while (processed < MAX_ROWS_PER_TICK) { const now = clock(); const claimed = await claimLanes( now, spent, Math.min(LANE_CONCURRENCY, MAX_ROWS_PER_TICK - processed), ); if (claimed.length === 0) break; await Promise.all(claimed.map((c) => runClaimed(c, now))); processed += claimed.length; } return processed;}
export type SyncRunnerHandle = { stop(timeoutMs: number): Promise<void> };
export function startSyncRunner( work: () => Promise<unknown> = () => runSyncQueue(), { registerWake = true }: { registerWake?: boolean } = {},): SyncRunnerHandle { let inFlight: Promise<void> | null = null; let dirty = false; let stopped = false;
// Single-flight with a dirty flag, like the Jetstream manager's refresh: a // wake during a tick queues exactly one more tick with the latest state. const tick = (): void => { if (stopped) return; if (inFlight) { dirty = true; return; } // A wake can fire from inside a caller's transaction; the tick must still // run on the pool, not inherit that transaction. inFlight = outsideTransaction(work) .then(() => undefined) .catch((err) => { console.error("[sync-runner] tick failed:", err); }) .finally(() => { inFlight = null; if (dirty && !stopped) { dirty = false; tick(); } }); };
if (registerWake) registerSyncWake(tick); const interval = setInterval(tick, FALLBACK_TICK_MS); tick();
return { async stop(timeoutMs: number) { stopped = true; clearInterval(interval); if (registerWake) registerSyncWake(null); const current = inFlight; if (!current) return; await Promise.race([ current, new Promise<void>((resolve) => setTimeout(resolve, Math.max(timeoutMs, SHUTDOWN_WAIT_MIN_MS)), ), ]); }, };}