Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 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
25 changes: 25 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,31 @@ 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.

**`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); 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.

## Non-negotiables

- **Bun only.** No Node-specific APIs unless via `node:` and unavoidable, and then with a comment saying why.
Expand Down
4 changes: 3 additions & 1 deletion docs/architecture/08-jobs-internals.md
Original file line number Diff line number Diff line change
Expand Up @@ -68,10 +68,12 @@ export interface JobDriver {

| 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 |
| `pg` (default) | `x_jobs`, `x_job_steps`, `x_backfills`, `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` |

`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.

## The pg claim loop
Expand Down
47 changes: 47 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,52 @@ 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;
if (ledger === undefined) throw new Error('the memory driver must ship a backfill ledger');
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
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 +220,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
77 changes: 77 additions & 0 deletions packages/admin/src/dev/panel-jobs.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
import { describe, expect, test } from 'bun:test';
Comment thread
sebyx07 marked this conversation as resolved.
import { staticDevSources } from './data';
import type { BackfillFact, JobRunFact } from './facts';
import { jobsPanel } from './panel-jobs';

const pass = (fact: Partial<BackfillFact> & { runId: string }): BackfillFact => ({
name: 'recount-likes',
status: 'running',
rows: 0,
cursor: null,
startedAt: '2026-08-14T09:00:00.000Z',
completedAt: null,
durationMs: null,
appVersion: '1.2.0',
...fact,
});

const run = (fact: Partial<JobRunFact> & { id: string }): JobRunFact => ({
job: 'recount-likes',
queue: 'default',
status: 'running',
attempt: 1,
steps: [],
...fact,
});

describe('the jobs panel reports the backfill ledger', () => {
test('every pass is carried, and only the live ones are counted in flight', async () => {
const data = await jobsPanel.data(
staticDevSources({
backfills: () =>
Promise.resolve([
pass({ runId: 'r3', rows: 12_500, cursor: 'post_12500' }),
pass({ runId: 'r2', status: 'failed', rows: 400, cursor: 'post_400' }),
pass({
runId: 'r1',
status: 'completed',
rows: 84_000,
completedAt: '2026-08-14T09:04:00.000Z',
durationMs: 240_000,
}),
]),
}),
new URLSearchParams(),
);

// The whole ledger, because "has this ever run here, and what did it sweep" is the panel's
// question — a filter to `running` would answer it with silence on a finished sweep.
expect(data.backfills.map((entry) => entry.runId)).toEqual(['r3', 'r2', 'r1']);
// A failed pass is an attempt the queue will retry, not a sweep in flight.
expect(data.backfillsInFlight).toBe(1);
expect(data.backfills[0]?.rows).toBe(12_500);
expect(data.backfills[0]?.cursor).toBe('post_12500');
});

test('a ?queue= filter scopes the runs and never the ledger', async () => {
const sources = staticDevSources({
jobRuns: () =>
Promise.resolve([run({ id: 'r3' }), run({ id: 'other', queue: 'mail', job: 'send' })]),
backfills: () => Promise.resolve([pass({ runId: 'r3' })]),
});

const data = await jobsPanel.data(sources, new URLSearchParams('queue=mail'));

// A backfill belongs to a table, not to a queue: scoping the ledger by the queue filter would
// hide the one pass whose job row the same filter just dropped.
expect(data.runs.map((entry) => entry.id)).toEqual(['other']);
expect(data.backfills.map((entry) => entry.runId)).toEqual(['r3']);
expect(data.backfillsInFlight).toBe(1);
});

test('an app that has never swept anything reports an empty ledger, not a failure', async () => {
const data = await jobsPanel.data(staticDevSources(), new URLSearchParams());
expect(data.backfills).toEqual([]);
expect(data.backfillsInFlight).toBe(0);
});
});
Loading