Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,33 @@ Semver applies from 1.0.0. A breaking change to a documented API needs a major

### Added

- **`backfill()` — one pass over a table, declared as a `job`.** `@ultimat3/jobs` gains a factory, not a ninth primitive: a backfill is durable work with an input schema, a retry policy, an idempotency key and a queue, so `backfill({ name, source, batch, handle })` *returns* a `JobHandle` and inherits `.enqueue()`, the worker's cancellation, the dead-letter path, `x jobs show` and its manifest row with no line of its own. Same rule `llm()` follows: a new capability arrives as a factory over an existing primitive.

```ts
export const rewriteSlugs = backfill({
name: 'rewrite-slugs',
batch: 1_000,
source: () => db.posts.where({ published: true }),
async handle({ rows }) {
await db.posts.upsertAll(rows.map(slugged), { onConflict: ['id'] });
},
});
```

`source` is read through `inBatches()` — one statement per page, keyset, never OFFSET — and every page is handled inside its own `step.run`, named `batch:0`, `batch:1`, … So a run killed mid-pass resumes on the page it stopped at: the completed steps replay from storage without touching the database, and the iteration is opened at the cursor they leave behind. What a step persists is that cursor and a row count, **never the page** — `steps.ts` hands a completed step's output back for the whole run, so checkpointing rows would retain every row already processed until the job ended. The body is handed no `step` for the same reason step names are positional: a name minted inside it would collide with itself on the second batch. `idempotencyKey` is the backfill's name, so re-enqueueing a live pass is the same pass.
Comment thread
coderabbitai[bot] marked this conversation as resolved.

**`handle` is at least once, and that is a requirement on the handler.** It runs *before* its checkpoint lands, so an attempt killed, cancelled or lease-expired between the last row and the step record hands the same page to the next attempt: write through `upsertAll`, `updateWhere` or a statement whose second run changes nothing, never `count + 1` and never an unguarded external side effect. The order is the only one that is safe — a checkpoint written first would report a page as swept that no attempt ever wrote, and a lost page is unrecoverable where a repeated one is a handler's problem to be idempotent about.

**`x_backfills` records what has already been swept**, the twin of `x_migrations` one level up: name, definition checksum, status, app version, rows processed, last cursor, started/completed. It ships in the same DDL as `x_jobs` and `x_job_steps` and hangs off the queue driver as `driver.backfills`, so a ledger a pass cannot write is a queue it could not have been claimed from — and a driver without one runs backfills with no bookkeeping rather than refusing them, exactly as `introspect` already degrades. Re-enqueueing a **completed** name is a no-op with a report (`{ skipped: true, previousRunId }`) — no statement, no body call, no row. `enqueue({ force: true })` sweeps again and writes a **new** row: reruns are history, never an edit of the row they rerun. A definition whose checksum moved since the completed pass **warns** and still does not run, because this checksum is over function source text, which a bundler can move without a line of behaviour changing — where `@ultimat3/db`'s `auditLedger` throws on the same fact, since SQL text is what it applied.

The row is a **report**, never a resume source: where a resumed pass restarts is decided by the step checkpoints and by nothing else. A retry adopts its own row (`started_at` is when the pass began, not this attempt) and clears `completed_at`, which `finish` stamps for `failed` as well as for `completed` — every surface projects that column, so a running pass keeping it would render with a completion time in the past. An attempt that failed is recorded as `failed` with its cursor kept, because where a pass stopped is the first thing anyone asks about one.

**A backfill is throttled, and the default is slow.** `rate` is batches per second and defaults to `DEFAULT_BACKFILL_RATE` (5) — 5,000 rows/sec at the default batch, one statement every 200ms, so the pool spends the rest of each interval serving the requests the app is still taking. There is **no unthrottled mode**: to sweep faster you raise the number, and a rate above what the batches can achieve simply produces no wait. The pause is spent **inside** each batch's `step.run`, which is what stops a resumed attempt from re-paying the throttle of the five hundred batches it is replaying, and it unwinds on the run's cancellation (`X_ABORTED`) rather than sitting in a timer nobody is waiting for. `rate` is not part of the definition checksum, for the reason `batch` is not: pacing is tuning, and changing it does not make a completed sweep a different sweep.

**Progress is readable from three surfaces, all reading the one ledger.** `x db backfill --list` prints it (`--name`, `--status`, `--limit`, and `--json` carrying the same rows); `x jobs ls` reports the passes **in flight** with rows-so-far and cursor, and `x jobs show <id>` carries the ledger row for that run under `backfill`; `/_x`'s jobs panel carries the whole ledger plus a live count. One projection behind all of them — `inspectBackfills()` in `@ultimat3/jobs` — so the dashboard renders what `--json` prints. `x db backfill` without `--list` is refused with the command that works, because `x db backfill <name>` will one day *run* a pass and a bare invocation that quietly listed would be a silent no-op the day it lands.

- **`appVersion()` in `@ultimat3/core`.** One reader for `APP_VERSION`, defaulting to `dev`. It was spelled inside `@ultimat3/db`'s `runningAppVersion`, which now delegates — `x_migrations.app_version` and `x_backfills.app_version` are two durable columns an operator reads side by side, `@ultimat3/jobs` cannot reach `db` for the answer, and two packages defaulting the key their own way would put two names on one build.

- **The destructive-SQL rail: a migration that destroys data must say so, and `x verify` refuses one that does not.** `x db gen` now writes `-- destructive: true` into any migration whose `up` drops a table, drops a column, truncates or retypes; `x verify`'s `drift` step reads the committed files back and fails an unmarked one with the new `X_MIGRATION_DESTRUCTIVE`. The strong-migrations idea, enforced rather than documented — a drop is allowed, an *undeclared* drop is not.

```text
Expand Down
2 changes: 2 additions & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,8 @@ Everything in the framework is one of these. **If a feature doesn't fit one of t

**`llm()` is an action factory, not a ninth primitive — decided 2026-08.** A model call is a server-authoritative operation with an input schema, an output schema and a policy, which is the definition of an `action`; so `llm()` ([`packages/ai/src/llm.ts`](packages/ai/src/llm.ts)) *returns* one. That is what gives a model call `.tool()`, `.openapi()`, `.client()`, `.job()` and `.contract()` for free, one authz object across every surface, and a place in the manifest — none of which a ninth primitive would have inherited. The rule generalises: a new capability arrives as a **factory over an existing primitive**, never as a new kind of thing.

**`backfill()` is a job factory — the rule's second instance, decided 2026-08.** A one-pass sweep over a table is durable background work with an input schema, a retry policy, an idempotency key and a queue, which is the definition of a `job`; so `backfill()` ([`packages/jobs/src/backfill.ts`](packages/jobs/src/backfill.ts)) *returns* one, and inherits `.enqueue()`, the worker's cancellation, the dead-letter path, `x jobs show` and its manifest row. The pass is `inBatches()` — one statement per page — with every page in its own `step.run`, so a killed attempt resumes on the page it stopped at. What a step persists is a cursor, never the page. **`handle` is at least once**: it runs before its checkpoint lands, so an attempt cancelled between the two replays that page — the handler must be idempotent (`upsertAll`, `updateWhere`, a statement whose second run changes nothing), never `count + 1`.

## Non-negotiables

- **Bun only.** No Node-specific APIs unless via `node:` and unavoidable, and then with a comment saying why.
Expand Down
39 changes: 24 additions & 15 deletions docs/architecture/08-jobs-internals.md
Original file line number Diff line number Diff line change
Expand Up @@ -51,28 +51,37 @@ Worked trace for the canonical job:

## Driver interface

One interface, three implementations. **Job code never changes.**
One interface, four implementations — three for production plus `memory`, which is the one every framework test and `x dev` run against. **Job code never changes.**

Six required methods, the `steps` store, and three optional members. A driver that ships none of the three is still a working queue: `introspect` absent is `x jobs ls` with nothing to list, `backfills` absent is a `backfill()` pass that runs with no bookkeeping rather than one that is refused, and `close` absent is a driver holding nothing to hand back.

```ts
export interface JobDriver {
enqueue(job: JobRef, input: unknown, opts: EnqueueOpts, tx?: Tx): Promise<JobId>;
claim(queue: string, limit: number): Promise<ClaimedJob[]>;
heartbeat(id: JobId): Promise<void>;
complete(id: JobId, result: unknown): Promise<void>;
fail(id: JobId, err: SerializedError, retryAt: Date | null): Promise<void>;
saveStep(id: JobId, name: string, result: unknown): Promise<void>;
loadSteps(id: JobId): Promise<Record<string, unknown>>;
sleepUntil(id: JobId, at: Date): Promise<void>;
readonly name: string;
/** Step persistence lives with the queue: one store, one transaction boundary. */
readonly steps: StepStore;
enqueue(request: EnqueueRequest): Promise<EnqueueResult>;
claim(options: ClaimOptions): Promise<readonly ClaimedJob[]>;
ack(jobId: string): Promise<void>;
nack(jobId: string, options: NackOptions): Promise<void>;
heartbeat(jobId: string, options: { readonly visibilityTimeoutMs: number }): Promise<void>;
stats(): Promise<readonly QueueStats[]>;
readonly backfills?: BackfillLedger;
readonly introspect?: JobIntrospection;
close?(): Promise<void>;
}
```

| Driver | State | Trade-off |
|---|---|---|
| `pg` (default) | `x_jobs`, `x_job_steps`, `x_outbox`, `x_rate_buckets` | outbox is free (same DB, same tx); `SKIP LOCKED` claiming; zero extra infra |
| `redis` | streams + consumer groups, outbox relay in front | high throughput, short jobs; loses "queue state in one backup" |
| `nats` | JetStream, outbox relay in front | strongest delivery semantics, most operational surface. `As of 2026-07` `claim` throws `X_NOT_IMPLEMENTED` with `fix: set jobs.driver = "pg" in app.config.ts` |
| Driver | State | `backfills` | Trade-off |
|---|---|---|---|
| `pg` (default) | `x_jobs`, `x_job_steps`, `x_backfills`, `x_outbox`, `x_rate_buckets` | yes | outbox is free (same DB, same tx); `SKIP LOCKED` claiming; zero extra infra |
| `memory` | in-process maps, lost with the process | yes | tests and `x dev` only — nothing survives a restart, so it is never a deployment target |
| `redis` | streams + consumer groups, outbox relay in front | no | high throughput, short jobs; loses "queue state in one backup" |
| `nats` | JetStream, outbox relay in front | no | strongest delivery semantics, most operational surface. `As of 2026-07` `claim` throws `X_NOT_IMPLEMENTED` with `fix: set jobs.driver = "pg" in app.config.ts` |

`x_backfills` is the odd one out: it is not queue state but the ledger of what a `backfill()` pass has already swept, hanging off `JobDriver.backfills` because it ships in the same DDL as `x_jobs` — `As of 2026-08` only the `pg` and `memory` drivers carry one, and a driver without it runs backfills with no bookkeeping rather than refusing them.
Comment thread
coderabbitai[bot] marked this conversation as resolved.

Because `saveStep`/`loadSteps` are driver methods, step persistence works identically on all three. Switching is a config line plus `x jobs drain --to redis` for in-flight rows.
Because `steps` is a driver member, step persistence works identically on all four. Switching is a config line plus `x jobs drain --to redis` for in-flight rows.

## The pg claim loop

Expand Down
50 changes: 50 additions & 0 deletions packages/admin/src/dev/data.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,55 @@ describe('hooks', () => {
});
});

describe('the backfill ledger source', () => {
test('the WHOLE ledger comes back, completed passes included', async () => {
const { createMemoryDriver, resetJobDriver, setJobDriver } = await import('@ultimat3/jobs');
const driver = createMemoryDriver();
setJobDriver(driver);
try {
const ledger = driver.backfills;
// An assertion rather than a throw: `backfills` is optional on `JobDriver`, so this is a
// claim about the memory driver worth failing on by name — and it narrows the type.
expect(ledger).toBeDefined();
if (ledger === undefined) return;
await ledger.start({ runId: 'r1', name: 'sweep', checksum: 'aa', appVersion: '1.2.0' });
await ledger.progress('r1', { rows: 40, cursor: 'p40' });
await ledger.finish('r1', { status: 'completed', rows: 40 });
await ledger.start({ runId: 'r2', name: 'recount', checksum: 'bb', appVersion: '1.2.0' });
await ledger.progress('r2', { rows: 7, cursor: 'p7' });

const facts = await defaultDevSources().backfills();

// Filtered to `running`, this panel would report a finished sweep as one that never
// happened — which is the question `x jobs ls` deliberately does NOT answer and this one
// does. `completed` clears the cursor; the row count is what survives.
expect(facts.map((fact) => [fact.runId, fact.status, fact.rows])).toEqual([
['r2', 'running', 7],
['r1', 'completed', 40],
]);
expect(facts[1]?.durationMs).not.toBeNull();
expect(facts[0]?.durationMs).toBeNull();
expect(facts[0]?.cursor).toBe('p7');
} finally {
resetJobDriver();
await driver.close();
}
});

test('no queue at all refuses, naming the source and the panel', async () => {
// The same line `queues` and `jobRuns` draw: a process with no driver cannot answer "what has
// been swept", and `[]` there would claim nothing ever has. A driver that merely ships no
// ledger is the other case — `inspectBackfills` answers `[]` for it, in `@ultimat3/jobs`.
const caught = (await defaultDevSources()
.backfills()
.catch((error: unknown) => error)) as { cause: string; fix: string };
Comment thread
coderabbitai[bot] marked this conversation as resolved.

expect(caught).toBeUltimateError('X_NOT_IMPLEMENTED');
expect(caught.cause).toContain('backfills');
expect(caught.cause).toContain('jobs');
});
});

describe('staticDevSources', () => {
test('still answers empty for everything — the explicit-fixture path must not throw', async () => {
const sources = staticDevSources();
Expand All @@ -174,6 +223,7 @@ describe('staticDevSources', () => {
await expect(sources.jobDefs()).resolves.toEqual([]);
await expect(sources.queues()).resolves.toEqual([]);
await expect(sources.jobRuns()).resolves.toEqual([]);
await expect(sources.backfills()).resolves.toEqual([]);
await expect(sources.tasks()).resolves.toEqual([]);
await expect(sources.tables()).resolves.toEqual([]);
await expect(sources.drift()).resolves.toEqual([]);
Expand Down
27 changes: 27 additions & 0 deletions packages/admin/src/dev/data.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import type { JobState, StepStatus } from '@ultimat3/jobs';
import type { AdminActor, AdminAuthz } from '../authz';
import { DevSourceUnavailableError } from '../errors';
import type {
BackfillFact,
CacheEdgeFact,
DevSources,
DriftFact,
Expand Down Expand Up @@ -46,6 +47,7 @@ export function staticDevSources(facts: Partial<DevSources> = {}): DevSources {
jobDefs: facts.jobDefs ?? ((): Promise<readonly JobDefFact[]> => empty([])),
queues: facts.queues ?? ((): Promise<readonly QueueFact[]> => empty([])),
jobRuns: facts.jobRuns ?? ((): Promise<readonly JobRunFact[]> => empty([])),
backfills: facts.backfills ?? ((): Promise<readonly BackfillFact[]> => empty([])),
tasks: facts.tasks ?? ((): Promise<readonly TaskFact[]> => empty([])),
tables: facts.tables ?? ((): Promise<readonly TableFact[]> => empty([])),
drift: facts.drift ?? ((): Promise<readonly DriftFact[]> => empty([])),
Expand Down Expand Up @@ -253,6 +255,31 @@ export function defaultDevSources(opts: DevSourceOptions = {}): DevSources {
}));
},

/**
* The whole ledger, newest first — not just the passes in flight. `x jobs ls` reports the live
* queue and says so; a panel is read to answer "has this backfill ever run here, and what did
* it sweep", and the completed rows ARE that answer.
*
* `inspectBackfills` returns `[]` for a driver that ships no ledger, so only a process with no
* queue at all refuses here — which is the same line `queues` and `jobRuns` draw.
*/
async backfills(): Promise<readonly BackfillFact[]> {
const { inspectBackfills, jobDriver } = await import('@ultimat3/jobs');
const driver = jobDriver();
if (driver === undefined) throw unavailable('backfills', 'jobs');
return (await inspectBackfills(driver)).map((pass) => ({
runId: pass.runId,
name: pass.name,
status: pass.status,
rows: pass.rows,
cursor: pass.cursor,
startedAt: pass.startedAt,
completedAt: pass.completedAt,
durationMs: pass.durationMs,
appVersion: pass.appVersion,
}));
},

async tasks(): Promise<readonly TaskFact[]> {
const { inspectManifest } = await import('@ultimat3/jobs');
return inspectManifest().tasks.map((task) => ({
Expand Down
23 changes: 23 additions & 0 deletions packages/admin/src/dev/facts.ts
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,28 @@ export interface TaskFact {
readonly nextRunAt: string | null;
}

/**
* One `x_backfills` pass. The queue's own facts answer "is this job moving"; only the ledger
* answers "how much of the table is behind it", which is the one question a sweep is watched for.
* `cursor` is where the pass had got to and `null` before its first batch — never a percentage:
* nothing counted the rows ahead of it, and a made-up denominator is the fact this panel exists
* not to invent.
*/
export interface BackfillFact {
/** The pass, and the run id of the job sweeping it — the join back to `runs` on this panel. */
readonly runId: string;
readonly name: string;
readonly status: 'running' | 'completed' | 'failed';
readonly rows: number;
readonly cursor: string | null;
readonly startedAt: string;
readonly completedAt: string | null;
/** How long the pass took. `null` while it is still running. */
readonly durationMs: number | null;
/** The build that STARTED the pass — a redeploy mid-sweep does not rewrite it. */
readonly appVersion: string;
}

export interface ColumnFact {
readonly name: string;
readonly type: string;
Expand Down Expand Up @@ -192,6 +214,7 @@ export interface DevSources {
jobDefs(): Promise<readonly JobDefFact[]>;
queues(): Promise<readonly QueueFact[]>;
jobRuns(): Promise<readonly JobRunFact[]>;
backfills(): Promise<readonly BackfillFact[]>;
tasks(): Promise<readonly TaskFact[]>;
tables(): Promise<readonly TableFact[]>;
drift(): Promise<readonly DriftFact[]>;
Expand Down
1 change: 1 addition & 0 deletions packages/admin/src/dev/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

export { type DevSourceOptions, defaultDevSources, staticDevSources } from './data';
export type {
BackfillFact,
CacheEdgeFact,
ColumnFact,
DevSources,
Expand Down
Loading