Rawman/directus-extension-generate-image-previews
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: Rawman:73da969ab761d8f967814ab7b8844799d06cbbfc
Branches Tags
Rawman:master
Rawman:v0.2.0
..
compare: Rawman:73e837c16515e226e487cf5a9d1ecfeeb39dbd1e
Branches Tags
Rawman:master
Rawman:v0.2.0

No commits in common. "73da969ab761d8f967814ab7b8844799d06cbbfc" and "73e837c16515e226e487cf5a9d1ecfeeb39dbd1e" have entirely different histories.
73da969ab7 ... 73e837c165

19 changed files with 26 additions and 2872 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -3,4 +3,3 @@ node_modules
dist dist
dev/extension dev/extension
dev/.smoke.env dev/.smoke.env
.opencode/tmp/

48
.opencode/feature_list.json
View file

@ -201,54 +201,6 @@
"Promotion to prod is a human action; the loop proposes, a human deploys.", "Promotion to prod is a human action; the loop proposes, a human deploys.",
"Record the exact cron entry, the scoped grants, and the rollback commands verbatim." "Record the exact cron entry, the scoped grants, and the rollback commands verbatim."
] ]
},
{
"feature_id": "F11",
"title": "Persistent failure journal, bounded retry, failure API and module view",
"acceptance_steps": [
"Every failed (file, preset, format) task from the upload hook, manual `/generate`, and backfill appends an event to `<state-dir>/failures.jsonl`; replay derives the latest state per key across restarts and PM2 workers.",
"Default retry policy is max 3 attempts (initial + 2 retries) with 5 min then 15 min backoff; a failed task gets at least one automatic retry when the scheduler next runs with no active backfill inside the maintenance window.",
"Due retries run under the existing `run.lock`, process only the targeted file/preset/format pairs (bounded batch, default 50), and never trigger a full-library sweep; a manual `Run now` may ignore the window.",
"A recovered task appends a normal `state.jsonl` completion and is marked `recovered`; an exhausted task stays `failed` with attempts, last error, and timestamps, and is not retried again.",
"`GET /image-previews/failures?status=&limit=&offset=` returns persisted records plus file metadata (`filename_download`, `type`, `folder` when available), attempts, first/last failure, next retry, status, and error text; admin-only and read-only.",
"The admin module shows a \"Failed previews\" table with file, preset, format, attempts, last error, next retry, and status; no toasts or user-facing notifications are added.",
"Every failure logs a structured warn with `fileId`, `preset_key`, `format`, `attempt`, `next_retry_at`, and `error`; recovery logs at info. No file contents are logged.",
"Journal compaction under the run lock retains unresolved rows plus the last 1,000 recovered rows, so the file cannot grow without bound.",
"Targeted unit tests cover journal append/replay/idempotency, backoff schedule, max-attempt exhaustion, due-retry selection, and the failure endpoint response shape."
],
"passes": true,
"blocked_by": [
"F08"
],
"notes": [
"Root cause: `/ops` is an in-memory 200-entry ring buffer per PM2 worker; after a long run the 17 failure details were unrecoverable.",
"Do not reuse `/ops` for persistence; keep it as the live view. Journal is append-only for cluster safety; compact only under `run.lock`.",
"Retry policy starts with constants; expose settings only if tuning is needed after the first real run.",
"No user-facing notifications; logs and the read-only API/module table are the observability surface.",
"Accepted 2026-10-30 on code + test evidence (79/79, tsc clean, build OK) after three review rounds; browser module render and live PM2 scheduler/retry tick were waived by the user and remain unverified."
]
},
{
"feature_id": "F12",
"title": "Host fix: `uploaded_on` integrity for TUS and legacy files (no Directus upgrade)",
"acceptance_steps": [
"Before change, record `count(*)`, `count(*) filter (where uploaded_on is null)`, and `count(*) filter (where created_on is null)` for `directus_files`; confirm the legacy null block explains the file-library ordering.",
"In a transaction after a table backup, run `update directus_files set uploaded_on = coalesce(created_on, now()) where uploaded_on is null;` and verify only `uploaded_on` changed (`modified_on`/`modified_by` unchanged, no new activity/revisions).",
"Set the column default to `(now() at time zone 'utc')` when absent, so future TUS inserts that omit `uploaded_on` receive a date.",
"Upload one test file with `TUS_ENABLED=true`; confirm `uploaded_on` is non-null, `tus_id`/`tus_data` are null after finish, and the file appears at the top of the unfiltered folder with `-uploaded_on`.",
"Keep `TUS_ENABLED=true`; no Directus upgrade and no extension code change required for this fix.",
"Document latent 11.17.4 TUS defects not changed here: client-declared `type`/`filesize`, and stale `tus_id` placeholders hidden from `/files` if finish/cleanup fails."
],
"passes": true,
"blocked_by": [],
"notes": [
"Source: v11.17.4 `api/src/services/tus/{data-store,server}.ts` omits `uploaded_on`; only `/files` (`FilesService.uploadOne`) sets it. Upstream #28042 fixed in 12.4.0, not backported to 11.x.",
"Raw SQL is the only way to avoid bumping `modified_on`/`modified_by`; `ItemsService.updateOne` always runs payload specials even with `emitEvents:false`.",
"TUS row insert omits `uploaded_on` entirely, so a DB default applies to future inserts. Host `TimeZone=Etc/UTC`; `(now() at time zone 'utc')::timestamptz` matches Directus's UTC writes.",
"Verified 2026-10-30: 2,836 null rows backfilled from `created_on`; 39,350/39,350 dated; latest 5 TUS rows have non-null `uploaded_on`, `tus_id=null`, delta vs `created_on` 0–0.24s; zero `idle in transaction`.",
"Apparent inverted sort is a UI icon issue, not data: Directus cards `header.vue` flips the sort glyph for descending, opposite to tabular. Network `sort[]=-uploaded_on` = newest first; `sort[]=uploaded_on` = oldest first. No server/data fix needed.",
"This is host data remediation, not extension code; keep it separate from the preview warmer."
]
} }
] ]
} }

45
.opencode/goal.md
View file

@ -3,7 +3,6 @@
Status: SCAFFOLDED — opened 2026-10-25. Standalone extension project (independent of any host Status: SCAFFOLDED — opened 2026-10-25. Standalone extension project (independent of any host
application). application).
Location: `/home/rawman/Projects/DirectusExtensions/directus-extension-generate-image-previews` Location: `/home/rawman/Projects/DirectusExtensions/directus-extension-generate-image-previews`
Branch: `goal/image-previews` (created from master 2026-10-30).
Target: self-hosted Directus **11.17.4** (PM2 cluster, local disk storage). Target: self-hosted Directus **11.17.4** (PM2 cluster, local disk storage).
Project type: Directus **bundle** extension (app module + api hook + api endpoint). Project type: Directus **bundle** extension (app module + api hook + api endpoint).
@ -113,48 +112,6 @@ Our registry (`src/lib/presets.ts`) mirrors this exact list:
- `debug` setting enables verbose structured logs; logs carry ids/metadata only, never file contents. - `debug` setting enables verbose structured logs; logs carry ids/metadata only, never file contents.
- Everything except `/generate` is read-only. - Everything except `/generate` is read-only.
## Failure journal & bounded retries (2026-10-30)
Backfill failures are currently only counted in `run.json` and held in the in-memory `/ops` ring
buffer (last 200 entries, per PM2 worker). After a long run the details are lost. Add a durable
failure journal and a bounded retry path.
- `failures.jsonl` under the state dir = append-only events `{type:'failed'|'retry'|'recovered',
ts, file_id, preset_key, format, attempt, next_retry_at?, error?, ms?}`. Replay derives the latest
state per `(file_id, preset_key, format)`; append-only so PM2 workers can write safely.
- Every failure (upload hook, manual `/generate`, backfill) is journaled. Default retry policy:
max 3 attempts (initial + 2 retries), backoff 5 min then 15 min. Due retries run only while enabled,
inside the maintenance window (manual `Run now` may ignore it), with no active backfill, under the
same `run.lock`; targeted file/preset/format retries — never a full-library sweep.
- Success after retry appends a normal `state.jsonl` completion and marks the failure `recovered`.
Exhausted retries stay `failed` with the last error and remain readable.
- `GET /image-previews/failures?status=&limit=&offset=` returns persisted records + file metadata,
attempts, timestamps, next retry and error text. Admin-only, read-only.
- Module: "Failed previews" table (file, preset, format, attempts, last error, next retry, status).
No toasts or user-facing notifications.
- Log every failure at warn with ids/metadata + error; log recoveries at info. No file contents.
- Journal is compacted under the run lock: keep unresolved rows + last 1,000 recovered; never grow
unbounded. `/ops` remains the live in-memory view.
## Host data fix: `uploaded_on` for TUS uploads (2026-10-30)
Scope split (user decision): DB access and execution stay with the user; this repo only ships the SQL
script + runbook. No extension code change for this item.
Directus 11.17.4 TUS uploads omit `uploaded_on` (`TusDataStore.create` + `onUploadFinish`); the
regular `POST /files` path sets it. Upstream issue #28042 is fixed only in 12.4.0; this host stays on
11.17.4.
- Fix existing rows directly in Postgres: `update directus_files set uploaded_on =
coalesce(created_on, now()) where uploaded_on is null;` in a transaction, after a table backup.
Raw SQL touches only `uploaded_on`; it does not bump `modified_on`/`modified_by`, emit events, or
create revisions.
- Add a UTC column default so future TUS inserts get a date:
`alter table directus_files alter column uploaded_on set default (now() at time zone 'utc');`
Confirm one new TUS upload has non-null `uploaded_on` and `tus_id = null`.
- Keep `TUS_ENABLED=true`; no Directus upgrade and no extension code required. Latent TUS defects
(client-declared `type`/`filesize`, stale `tus_id` placeholders) are documented, not changed.
## Safe deployment & staging (must not crash prod) ## Safe deployment & staging (must not crash prod)
- **Fail-open, inert by default**: every entrypoint is tiny, bodies wrapped in try/catch, hooks never - **Fail-open, inert by default**: every entrypoint is tiny, bodies wrapped in try/catch, hooks never
@ -200,8 +157,6 @@ src/
warmer.ts in-process getAsset(x, { transformationParams, acceptFormat }) + concurrency warmer.ts in-process getAsset(x, { transformationParams, acceptFormat }) + concurrency
backfill.ts enumerate directus_files (image/*), window-aware, resumable backfill.ts enumerate directus_files (image/*), window-aware, resumable
state.ts state.jsonl read/append/resume state.ts state.jsonl read/append/resume
failures.ts failures.jsonl append/replay/compact
retry.ts due-retry selection + backoff/max-attempt policy
settings.ts config file read/write + env defaults settings.ts config file read/write + env defaults
disk.ts sum sizes of __<hash> files (driver.list + stat) disk.ts sum sizes of __<hash> files (driver.list + stat)
window.ts maintenance-window logic (is-now-in-window, next-window) window.ts maintenance-window logic (is-now-in-window, next-window)

166
.opencode/progress.txt
View file

@ -146,169 +146,3 @@ Built the full extension locally and smoke-tested against an **isolated** Direct
maintenance window, no external cron/token, pauses when the window closes and resumes next night. maintenance window, no external cron/token, pauses when the window closes and resumes next night.
- UI now shows last-run errors and `paused (window closed)`. - UI now shows last-run errors and `paused (window closed)`.
- Tests 43/43, typecheck clean. - Tests 43/43, typecheck clean.
## 2026-10-30 — file-library ordering / TUS `uploaded_on` / failure observability
- User report: fresh files absent from folder list unless filtered by `uploaded_on`; 4,649-item Documents
folder; page size raised to 1000 remained slow.
- Directus 11.17.4 source check: filtered result is an `_and` subset of unfiltered; no app item cache.
Postgres `ORDER BY uploaded_on DESC` = NULLS FIRST, so a large null-date block fills every page before
fresh rows. TUS (`TusDataStore.create` + `onUploadFinish`) never sets `uploaded_on`; regular
`POST /files` does. Upstream #28042 fixed only in 12.4.0; host stays 11.17.4.
- Host checks (pgAdmin): `created_on IS NULL` = **0**; `directus_files` total **39,350**,
`uploaded_on IS NULL` = **2,836**, dated = **36,514**.
- `uploaded_on` default added successfully: `DEFAULT (now() AT TIME ZONE 'utc')`.
- **Backfill UPDATE was not applied**: `uploaded_on IS NULL` still 2,836. Defaults do not backfill
history. Re-run `UPDATE directus_files SET uploaded_on = created_on WHERE uploaded_on IS NULL;`
(transaction, backup) and verify null count = 0.
- Slow thumbnails: backfill only ~95,744 / 661,232 tasks done (~14%); loading 1,000 cards issues 1,000
transform requests, most cold → on-demand sharp. Use 25–100 page size; let backfill finish or run
outside-window with low concurrency.
- Added feature specs F11 (durable `failures.jsonl` + bounded retry + `/failures` + module table) and
F12 (host `uploaded_on` remediation). F12 remains `passes:false` until the backfill UPDATE is
actually applied and one new TUS upload is verified.
## 2026-10-30 — production outage: uncommitted pgAdmin DDL held lock (resolved)
- Symptom: Directus UI/API slow, 500s, health timeouts, container unhealthy. `directus.log` showed
repeated `KnexTimeoutError: Timeout acquiring a connection`, `Can't connect to the database`, and
60s request aborts.
- Root cause: pgAdmin backend **PID 674331** (`pgAdmin 4 - CONN:4204762`) was `idle in transaction`
holding a granted `AccessExclusiveLock` on `public.directus_files`. Its last query was the
`ALTER TABLE ... ALTER COLUMN uploaded_on SET DEFAULT ...`; it was never committed. All 80 Directus
pool connections (4 workers × 20) were active waiting on `Lock/relation` for `directus_files`, so the
API pool was exhausted.
- A `commit;` from a different pgAdmin connection (PID 674244) does not commit another session's
transaction. DDL must be run with auto-commit on or `BEGIN; ...; COMMIT;` in the same session.
- Recovery: terminated PID 674331 (`pg_terminate_backend`). Its uncommitted ALTER rolled back; lock
released; Directus pool drained; `/server/health` returned `{"status":"ok"}` and container health
returned `healthy`.
- Re-applied default safely with `lock_timeout=3s`:
`DEFAULT (now() AT TIME ZONE 'utc')`. Verified in `information_schema.columns`.
- Applied backfill in one transaction with `lock_timeout=5s`, `statement_timeout=30s`:
`UPDATE directus_files SET uploaded_on = created_on WHERE uploaded_on IS NULL;` → **UPDATE 2836**,
COMMIT. Verified `null_dates = 0`, `dated = 39,350`, zero `Lock` waiters afterwards.
- No code/config deploy involved. Raw SQL did not touch `modified_on`/`modified_by`.
- Remaining verification: upload one TUS file and confirm non-null `uploaded_on` + `tus_id = null`;
then confirm the folder list shows it at top with `-uploaded_on`.
- Verified on 2026-10-30 after recovery:
- latest 5 TUS uploads all have non-null `uploaded_on`, `tus_id = null`, correct `folder`
- `uploaded_on - created_on` ≈ 0–0.24s (default is applied at INSERT, finish is immediate)
- `show timezone` = `Etc/UTC`, so `(now() AT TIME ZONE 'utc')` cast to `timestamptz` is correct
- zero rows `state = 'idle in transaction'`; no lock waiters
- DB/TUS acceptance for F12 is done. Only remaining check: browser Files view shows a fresh TUS file
at the top of the unfiltered folder with `sort=-uploaded_on` and `page=1`.
- F12 passes. Browser shows fresh files at top when the cards sort control sends `-uploaded_on`.
- Apparent "inverted sorting" is a Directus cards UI icon bug, not data. v11.17.4
`app/src/layouts/cards/components/header.vue` flips the `sort` glyph for descending, while tabular
flips it for ascending. Query semantics are correct: `sort[]=uploaded_on` = ascending (oldest first),
`sort[]=-uploaded_on` = descending (newest first). Selecting a field resets direction to ascending.
Workaround: toggle until DevTools shows `-uploaded_on`, ignore glyph; or use tabular.
## 2026-10-30 — F11 persistent failure journal + bounded retry + /failures + module table
Applicant: coder
Feature: F11 — Persistent failure journal, bounded retry, failure API and module view
Branch: feat/F11 @ de25e50 (base goal/image-previews @ 7951f82)
RUN: 2026-10-30T06:26:28Z | coder | /home/rawman/Projects/DirectusExtensions/directus-extension-generate-image-previews | feat/F11 @ de25e50 | npx tsc --noEmit; npm test; npm run build | specs=0 | exit=0 | pass=55 | log=.opencode/tmp/runs/f11-test.log
Summary:
- Added `src/lib/failures.ts`: append-only `<state-dir>/failures.jsonl`, replay of the latest record per
`(file_id,preset_key,format)` (key = `completedKeyOf`), corrupt-line tolerance, atomic compaction
(unresolved + last 1,000 recovered), and `listFailures` with status filter + limit/offset.
- Added `src/lib/retry.ts`: MAX_ATTEMPTS=3, 5m/15m backoff, `nextRetryAt`, `dueRetries`,
`journalFailure`/`markRecovered`, and `runDueRetries` (enabled/window/lock gated, targeted bounded
batch default 50, grouped by file, full preset + acceptFormat via `warmVariant`, completion + recovered
on success, attempt+1 + next retry on failure, compaction under `run.lock`, never throws).
- Hooked failure journaling into the upload warm hook, `POST /generate`, and the backfill; successes mark
prior failures recovered. Journal writes never alter responses or hook behavior.
- Hook scheduler tick now runs due retries first, then the existing backfill.
- Added admin-only `GET /image-previews/failures?status=&limit=&offset=` with one batched
`directus_files` metadata query (`filename_download`, `type`, `folder`) and ISO/human timestamps.
- Module: “Failed previews” table + Refresh (no toasts); API reference includes `/failures`; polling
interval unchanged.
- Types added; README/AGENTS/project SKILL state-file docs updated.
- Tests: `failures.test.ts` (append/replay/latest-state/corrupt/idempotent/compaction/list) and
`retry.test.ts` (backoff/exhaustion/due selection/bounded batch/grouping/missing preset/recovered/
failure/disabled).
Files changed:
- src/lib/failures.ts (new), src/lib/failures.test.ts (new)
- src/lib/retry.ts (new), src/lib/retry.test.ts (new)
- src/lib/types.ts, src/lib/warmer.ts, src/lib/backfill.ts
- src/hook/index.ts, src/endpoint/index.ts, src/module/module.vue
- README.md, AGENTS.md, .opencode/skills/directus-image-previews/SKILL.md, .gitignore
- .opencode/progress.txt
Commands:
- `npx tsc --noEmit` → exit 0, 0 errors (`.opencode/tmp/runs/f11-tsc.log`)
- `npm test` → exit 0, 55/55 pass (`.opencode/tmp/runs/f11-test.log`)
- `npm run build` → exit 0, `dist/app.js` + `dist/api.js` emitted (`.opencode/tmp/runs/f11-build.log`)
Manual verification steps:
1. Cause a failure (unsupported preset format or missing file) from upload, `POST /generate`, and backfill;
confirm one line per failure in `<state-dir>/failures.jsonl` and a warn log with `fileId`,
`preset_key`, `format`, `attempt`, `next_retry_at`, `error`.
2. Inside the maintenance window with no backfill active, wait for the next 15-minute tick; confirm a due
pair is retried, success writes `state.jsonl` + a `recovered` event and logs info, failure writes
`attempt+1` with 15m backoff (or exhausted at 3).
3. `GET /image-previews/failures?status=failed&limit=50&offset=0` as admin: records include file metadata,
attempts, first/last failure, next retry raw+ISO/human, status, error, ms; non-admin gets 403.
4. Module “Failed previews” table renders and Refresh reloads; no toast/notification appears.
Not verified locally:
- Browser rendering of the new module table (no live Directus/admin session in this run).
- Live PM2 `schedule('*/15 * * * *')` tick and maintenance-window retry execution.
- Concurrent PM2 append behavior and compaction race with upload-hook appends.
## 2026-10-30 — F11 review fixes
- Head: `feat/F11` @ `e10819c`.
- Replay now derives attempts from `event.attempt`; fail→recover→fail starts a fresh cycle (attempt 1) and duplicate failed events are idempotent.
- Journal writer lock (`failures.lock`, 20×10 ms, 30 s stale): appends serialize, compaction returns `false` if it cannot lock, and append still falls back to writing.
- Backfill loads the journal once, skips unresolved failures per pair, and caps journal attempts at `MAX_ATTEMPTS`; recovered pairs still append `recovered`.
- Added `/failures` shape/validation tests, retry lock/window tests, fail→recover→fail, attempt cap, and 50-append compaction race tests.
## 2026-10-30 — F11 second review fixes
- Implementation head: `feat/F11` @ `7f46493` (progress commit follows).
- Journal writes now require the `failures.lock` directory (`owner.json` with random token + pid) and never append unlocked. Blocking acquisition defaults to 10 s / 10 ms poll; timeout drops the event with `image-previews: could not journal failure` and `fileId`/`preset_key`/`format`.
- Stale locks are broken only when the owner pid is dead (`process.kill(pid, 0)` → `ESRCH`); mtime staleness removed. Release only removes the lock when `owner.json.token` still matches.
- `compactFailures` uses the same lock across read→write temp→rename, returns `false` when it cannot acquire, and uses a pid+token-unique temp filename.
- Tests now cover live-owner append refusal, dead-pid recovery, foreign-token release, live-lock compaction refusal, temporarily-held-lock 50-append race, and `planFileTasks` total/resume/skip counting.
- `npx tsc --noEmit` 0 errors; `npm test` 69/69 pass; `npm run build` emits both bundles. Logs: `.opencode/tmp/runs/f11-fix2-{tsc,test,build}.log`.
- Remaining manual checks: live PM2 scheduler retry tick, browser module rendering, and a real multi-worker compaction/append run.
## 2026-10-30 — F11 final lock hardening
- Implementation head: `feat/F11` @ `bc20c6a` (progress commit follows).
- Lock is now an atomic `failures.lock` file written with `flag:'wx'`; owner carries `token`, `pid`, `bootId`, `startTime`, `ts`. Identity readers use `/proc/sys/kernel/random/boot_id` and `/proc/<pid>/stat` field 22, with injectable readers for tests.
- Dead-lock reclaim runs only under `failures.lock.reclaim` (mkdir guard). Inside the guard it re-stats/re-reads the current owner and unlinks only a dead owner or an unreadable owner older than the 2 s grace. Live owners are never broken; stale guards older than 30 s are removed.
- Release unlinks only when the current `owner.token` matches; no unlink on mismatch or read error.
- Tests added: concurrent reclaim of a dead lock, missing/corrupt owner grace, PID reuse by bootId and startTime, guard-held reclaim and stale-guard recovery. Existing live-owner timeout, foreign-token release, 50-append race, and compaction-blocked tests stay green.
- `npx tsc --noEmit` 0 errors; `npm test` 76/76 pass; `npm run build` emits both bundles. Logs: `.opencode/tmp/runs/f11-lock-{test,build}.log` and `f11-lock-final-*`.
- Remaining manual checks: live PM2 multi-worker append/compaction, live scheduler retry tick, browser module rendering.
## 2026-10-30 — F11 reclaim-guard lifecycle fix
- Implementation head: `feat/F11` @ `cd973ca` (progress commit follows).
- Stale guard removal is now atomic: `rename(guardPath, guardPath.stale.<staleKey>)` with a marker inside the quarantined dir; losers retry instead of deleting a successor guard. Guard writes `owner.json` `{token,pid,ts}` and its `finally` only removes the guard when the token matches.
- Added injectable `now` clock. Tests: two waiters on one stale guard (both events survive), foreign-token guard release keeps the successor, plus existing lock/race/compaction tests.
- `npx tsc --noEmit` 0; `npm test` 78/78 pass; `npm run build` emits both bundles. Logs: `.opencode/tmp/runs/f11-guard2-{test,build}.log` and `f11-guard-final-*`.
- Remaining manual checks: live PM2 multi-worker append/compaction, live scheduler retry tick, browser module rendering.
## 2026-10-30 — F11 stale-guard snapshot verification
- Implementation head: `feat/F11` @ `4bd69dd` (progress commit follows).
- Stale-guard branch now snapshots `observedMtime` + `observedOwner` before any re-read, quarantines with `guardPath.stale.<pid>.<uuid>`, then verifies the moved directory's owner token (or both null) and mtime against the snapshot. Mismatch restores the successor guard and retries; restore failure returns false without running the action.
- Added deterministic successor-replacement test via guard hooks; both appends survive and the successor guard is restored.
- `npx tsc --noEmit` 0; `npm test` 79/79 pass; `npm run build` emits both bundles. Logs: `.opencode/tmp/runs/f11-guard3-{test,build}.log` and `f11-guard3-final-*`.
- Remaining manual checks: live PM2 multi-worker append/compaction, live scheduler retry tick, browser module rendering.
## 2026-10-30 — Manager: F11 accepted, F12 host fix recorded
- Decision: squash-merged `feat/F11` into `goal/image-previews` after reviewer APPROVE (79/79 tests, tsc clean, build OK) and flipped F11 passes=true. User waived the manual browser/live-PM2 checks; recorded as unverified in the F11 note.
- Context: three review rounds hardened replay attempts, compaction-vs-append locking, exhausted-retry handling, and reclaim-guard identity. Residual fs-lock scheduling windows are documented and accepted.
- Outcome: accepted. F12 passes=true came from the user's own host verification (2,836 null rows backfilled, column default set) and was committed unchanged. Extension repo still owns no DB code for F12.
- Learning: journal locking is the hard part, not the retry logic — plan the lock protocol before writing the journal. Browser + live-cluster checks remain the only gaps; deploy requires a human.

1
.opencode/skills/directus-image-previews/SKILL.md
View file

@ -65,7 +65,6 @@ cache files:
settings.json effective config settings.json effective config
run.json progress (running/done/total/generated/hits/failed/rate/eta/paused/cancelled/error) run.json progress (running/done/total/generated/hits/failed/rate/eta/paused/cancelled/error)
state.jsonl completed {file_id,preset_key,format} → resume state.jsonl completed {file_id,preset_key,format} → resume
failures.jsonl append-only failure events; replay = latest per task; compacted under run.lock
run.lock cluster-wide run lock run.lock cluster-wide run lock
cancel.flag cancellation request cancel.flag cancellation request
``` ```

2
AGENTS.md
View file

@ -94,8 +94,6 @@ pacing).
(default `/directus/uploads/.image-previews`; override `IMAGE_PREVIEWS_STATE_DIR`). Do NOT default to (default `/directus/uploads/.image-previews`; override `IMAGE_PREVIEWS_STATE_DIR`). Do NOT default to
`/directus/extensions/...` — that path is a root-owned volume and the API runs as `node` → EACCES. `/directus/extensions/...` — that path is a root-owned volume and the API runs as `node` → EACCES.
- `state.jsonl` = append-only completed `{file_id,preset_key,format}` → resume. - `state.jsonl` = append-only completed `{file_id,preset_key,format}` → resume.
- `failures.jsonl` = append-only failure events; replay derives the latest state per
`(file_id,preset_key,format)`. Compacted under `run.lock` (unresolved + last 1,000 recovered).
- All state is **file-backed** (PM2 cluster, 8 workers — never in-memory). - All state is **file-backed** (PM2 cluster, 8 workers — never in-memory).
## Nightly scheduling ## Nightly scheduling

18
README.md
View file

@ -105,8 +105,6 @@ All require a Directus admin session or admin token. Human-readable fields accom
- `POST /image-previews/generate` — one file (`{fileId, keys?, formats?, dryRun?}`) - `POST /image-previews/generate` — one file (`{fileId, keys?, formats?, dryRun?}`)
- `GET /image-previews/resolve-plan?fileId=&key=` — predicted cache filename, no write - `GET /image-previews/resolve-plan?fileId=&key=` — predicted cache filename, no write
- `GET /image-previews/ops?limit=` — recent operations with timing + error messages - `GET /image-previews/ops?limit=` — recent operations with timing + error messages
- `GET /image-previews/failures?status=&limit=&offset=` — persisted failures with file metadata, attempts,
timestamps, next retry and error (`status` = `failed`, `recovered`, or `retry`; limit max 200)
The admin module includes a full API reference with curl examples. The admin module includes a full API reference with curl examples.
@ -117,24 +115,10 @@ The extension schedules itself **inside Directus** — no external cron, no toke
- every 15 minutes it checks whether it is inside the maintenance window; - every 15 minutes it checks whether it is inside the maintenance window;
- if enabled, inside the window, and no run is active, it starts/resumes the backfill; - if enabled, inside the window, and no run is active, it starts/resumes the backfill;
- the backfill **pauses when the window closes** and resumes the next night (it is resumable); - the backfill **pauses when the window closes** and resumes the next night (it is resumable);
- once all tasks are done, it stays idle until the next day; - once all tasks are done, it stays idle until the next day.
- before starting the backfill it runs any due failure retries (targeted pairs only, bounded batch).
`Run now` in the admin module ignores the window. `Run now` in the admin module ignores the window.
## Failures and retries
Every failed `(file, preset, format)` task from the upload hook, `POST /generate`, and the backfill is
appended to `<state-dir>/failures.jsonl`. Replay derives the latest state per task, so details survive
restarts and PM2 workers. The journal is append-only; it is compacted under `run.lock` to keep unresolved
tasks plus the last 1,000 recovered ones.
Retry policy: max 3 attempts (initial + 2 retries), backoff 5 minutes then 15 minutes. Due retries run
inside the maintenance window only when no backfill is active, under the same `run.lock`, and process only
the targeted file/preset/format pairs (default batch 50). A success appends the normal `state.jsonl`
completion and marks the task `recovered`; an exhausted task stays `failed` with its last error and is not
retried again. Failures are logged at warn with ids/metadata only; recoveries at info.
## Safe testing & deployment ## Safe testing & deployment
Never test against a production instance. Never test against a production instance.

4
package-lock.json generated
View file

@ -1,12 +1,12 @@
{ {
"name": "directus-extension-generate-image-previews", "name": "directus-extension-generate-image-previews",
"version": "0.2.0", "version": "0.1.0",
"lockfileVersion": 3, "lockfileVersion": 3,
"requires": true, "requires": true,
"packages": { "packages": {
"": { "": {
"name": "directus-extension-generate-image-previews", "name": "directus-extension-generate-image-previews",
"version": "0.2.0", "version": "0.1.0",
"dependencies": { "dependencies": {
"object-hash": "^3.0.0" "object-hash": "^3.0.0"
}, },

2
package.json
View file

@ -15,7 +15,7 @@
"url": "https://code.at.locus.im/Rawman/directus-extension-generate-image-previews/issues" "url": "https://code.at.locus.im/Rawman/directus-extension-generate-image-previews/issues"
}, },
"icon": "extension", "icon": "extension",
"version": "0.2.0", "version": "0.1.0",
"keywords": [ "keywords": [
"directus", "directus",
"directus-extension", "directus-extension",

70
src/endpoint/index.ts
View file

@ -2,15 +2,12 @@ import { defineEndpoint } from '@directus/extensions-sdk';
import type { Request, Response, NextFunction } from 'express'; import type { Request, Response, NextFunction } from 'express';
import { runBackfill, requestCancel } from '../lib/backfill'; import { runBackfill, requestCancel } from '../lib/backfill';
import { computePlan, getOps, recordOp } from '../lib/diagnostics'; import { computePlan, getOps, recordOp } from '../lib/diagnostics';
import { enrichFailures, listFailures, parseFailureQuery, replayFailureMap } from '../lib/failures';
import { humanBytes, humanDuration, humanMs, toIso } from '../lib/format'; import { humanBytes, humanDuration, humanMs, toIso } from '../lib/format';
import { buildRegistry, effectiveFormatsForPreset, formatsForPreset, loadCustomPresets } from '../lib/presets'; import { buildRegistry, effectiveFormatsForPreset, formatsForPreset, loadCustomPresets } from '../lib/presets';
import { journalFailure, markRecovered } from '../lib/retry';
import { mergeSettings, readSettings, writeSettings } from '../lib/settings'; import { mergeSettings, readSettings, writeSettings } from '../lib/settings';
import { isLocked, readRun } from '../lib/state'; import { isLocked, readRun } from '../lib/state';
import { previewExists, storageUsage } from '../lib/storage'; import { previewExists, storageUsage } from '../lib/storage';
import { SYSTEM_PRESETS } from '../lib/system-presets'; import { SYSTEM_PRESETS } from '../lib/system-presets';
import type { FailureFileMeta, FailureListQuery } from '../lib/types';
import { isInWindow, nextWindow } from '../lib/window'; import { isInWindow, nextWindow } from '../lib/window';
import { warmFile } from '../lib/warmer'; import { warmFile } from '../lib/warmer';
@ -198,44 +195,6 @@ export default defineEndpoint({
}), }),
); );
router.get(
'/failures',
asyncHandler(async (req, res) => {
let query: FailureListQuery;
try {
query = parseFailureQuery({
status: req.query.status,
limit: req.query.limit,
offset: req.query.offset,
});
} catch (error) {
res.status(400).json({ error: error instanceof Error ? error.message : String(error) });
return;
}
const result = await listFailures(env, query);
const ids = [ ...new Set(result.records.map((record) => record.file_id)) ];
const files = new Map<string, FailureFileMeta>();
if (ids.length > 0) {
const schema = await getSchema();
const items = new services['ItemsService']('directus_files', { schema });
const rows = await items.readByQuery({
filter: { id: { _in: ids } },
fields: [ 'id', 'filename_download', 'type', 'folder' ],
limit: ids.length,
});
for (const row of Array.isArray(rows) ? rows : []) {
files.set(String(row.id), row as FailureFileMeta);
}
}
res.json({ ...result, records: enrichFailures(result.records, files) });
}),
);
router.post( router.post(
'/generate', '/generate',
asyncHandler(async (req, res) => { asyncHandler(async (req, res) => {
@ -294,35 +253,6 @@ export default defineEndpoint({
}), }),
}); });
const failureRecords = await replayFailureMap(env);
for (const result of summary.results) {
if (result.ok) {
await markRecovered(
env,
{
file_id: fileId,
preset_key: result.key,
format: result.format,
ms: result.ms,
},
{ records: failureRecords, logger },
);
} else {
await journalFailure(
env,
{
file_id: fileId,
preset_key: result.key,
format: result.format,
error: result.error,
ms: result.ms,
},
{ records: failureRecords, logger },
);
}
}
res.json({ res.json({
fileId, fileId,
ms: Date.now() - started, ms: Date.now() - started,

37
src/hook/index.ts
View file

@ -1,9 +1,7 @@
import { defineHook } from '@directus/extensions-sdk'; import { defineHook } from '@directus/extensions-sdk';
import { runBackfill } from '../lib/backfill'; import { runBackfill } from '../lib/backfill';
import { recordOp } from '../lib/diagnostics'; import { recordOp } from '../lib/diagnostics';
import { replayFailureMap } from '../lib/failures';
import { buildRegistry, isImageType, loadCustomPresets } from '../lib/presets'; import { buildRegistry, isImageType, loadCustomPresets } from '../lib/presets';
import { journalFailure, markRecovered, runDueRetries } from '../lib/retry';
import { readSettings } from '../lib/settings'; import { readSettings } from '../lib/settings';
import { readRun } from '../lib/state'; import { readRun } from '../lib/state';
import { warmFile } from '../lib/warmer'; import { warmFile } from '../lib/warmer';
@ -62,35 +60,6 @@ export default defineHook(({ action, schedule }, { services, database, getSchema
}), }),
}); });
const failureRecords = await replayFailureMap(env);
for (const result of summary.results) {
if (result.ok) {
await markRecovered(
env,
{
file_id: fileId,
preset_key: result.key,
format: result.format,
ms: result.ms,
},
{ records: failureRecords, logger },
);
} else {
await journalFailure(
env,
{
file_id: fileId,
preset_key: result.key,
format: result.format,
error: result.error,
ms: result.ms,
},
{ records: failureRecords, logger },
);
}
}
if (settings.debug) logger.info({ fileId, summary }, 'image-previews: warmed upload'); if (settings.debug) logger.info({ fileId, summary }, 'image-previews: warmed upload');
} }
@ -104,12 +73,6 @@ export default defineHook(({ action, schedule }, { services, database, getSchema
if (run?.running) return; if (run?.running) return;
const retryOutcome = await runDueRetries({ services, database, getSchema, env, logger }, {});
if (retryOutcome.status === 'completed' && retryOutcome.processed > 0) {
logger.info(retryOutcome, 'image-previews: due retries processed');
}
const finishedToday = run?.finishedAt const finishedToday = run?.finishedAt
? new Date(run.finishedAt).toDateString() === new Date().toDateString() ? new Date(run.finishedAt).toDateString() === new Date().toDateString()
: false; : false;

104
src/lib/backfill.ts
View file

@ -1,14 +1,5 @@
import { computePlan, recordOp } from './diagnostics'; import { computePlan, recordOp } from './diagnostics';
import { replayFailureMap } from './failures'; import { buildRegistry, effectiveFormatsForPreset, isImageType, loadCustomPresets, tasksPerFile } from './presets';
import {
buildRegistry,
effectiveFormat,
formatsForPreset,
isImageType,
loadCustomPresets,
tasksPerFile,
} from './presets';
import { journalFailure, markRecovered } from './retry';
import { readSettings } from './settings'; import { readSettings } from './settings';
import { import {
acquireLock, acquireLock,
@ -22,8 +13,8 @@ import {
requestCancel, requestCancel,
writeRun, writeRun,
} from './state'; } from './state';
import type { FailureRecord, Preset, RunState } from './types'; import type { Preset, RunState } from './types';
import { errorMessage, warmTasks, type WarmTaskSpec } from './warmer'; import { errorMessage, warmFile } from './warmer';
import { isInWindow } from './window'; import { isInWindow } from './window';
import { listFileNames } from './storage'; import { listFileNames } from './storage';
@ -48,45 +39,6 @@ export interface BackfillOutcome {
const PAGE_SIZE = 100; const PAGE_SIZE = 100;
export interface FileTaskPlan {
tasks: WarmTaskSpec[];
completed: number;
skipped: number;
}
export function planFileTasks(
fileId: string,
presets: Preset[],
formats: string[],
completed: Set<string>,
failures: Map<string, FailureRecord>,
): FileTaskPlan {
const tasks: WarmTaskSpec[] = [];
let completedCount = 0;
let skipped = 0;
for (const preset of presets) {
for (const variant of formatsForPreset(preset, formats)) {
const format = effectiveFormat(preset, variant);
const key = completedKeyOf(fileId, preset.key, format);
if (completed.has(key)) {
completedCount += 1;
continue;
}
if (failures.get(key)?.status === 'failed') {
skipped += 1;
continue;
}
tasks.push({ preset, format: variant });
}
}
return { tasks, completed: completedCount, skipped };
}
let inProcessRunning = false; let inProcessRunning = false;
async function countImageFiles(database: any): Promise<number> { async function countImageFiles(database: any): Promise<number> {
@ -143,7 +95,6 @@ export async function runBackfill(deps: BackfillDeps, options: BackfillOptions =
await writeRun(run, deps.env); await writeRun(run, deps.env);
const completed = await loadCompleted(deps.env); const completed = await loadCompleted(deps.env);
const failureRecords = await replayFailureMap(deps.env);
let existingFiles = new Set<string>(); let existingFiles = new Set<string>();
try { try {
@ -181,20 +132,24 @@ export async function runBackfill(deps: BackfillDeps, options: BackfillOptions =
if (!isImageType(file.type)) continue; if (!isImageType(file.type)) continue;
const planned = planFileTasks(file.id, presets, formats, completed, failureRecords); const pending = presets.filter((preset) =>
const tasks = planned.tasks; effectiveFormatsForPreset(preset, formats).some(
(format) => !completed.has(completedKeyOf(file.id, preset.key, format)),
),
);
if (planned.completed > 0) { const pendingTasks = pending.reduce(
run.hits += planned.completed; (total, preset) => total + effectiveFormatsForPreset(preset, formats).length,
run.done += planned.completed; 0,
);
const alreadyDone = perFile - pendingTasks;
if (alreadyDone > 0) {
run.hits += alreadyDone;
run.done += alreadyDone;
} }
if (planned.skipped > 0) { if (pending.length === 0) {
run.failed += planned.skipped;
run.done += planned.skipped;
}
if (tasks.length === 0) {
await writeRun(run, deps.env); await writeRun(run, deps.env);
continue; continue;
} }
@ -218,7 +173,7 @@ export async function runBackfill(deps: BackfillDeps, options: BackfillOptions =
} }
}; };
const summary = await warmTasks(assets, file.id, tasks, { const summary = await warmFile(assets, file.id, pending, {
concurrency: settings.concurrency, concurrency: settings.concurrency,
formats, formats,
isCached, isCached,
@ -244,32 +199,11 @@ export async function runBackfill(deps: BackfillDeps, options: BackfillOptions =
{ file_id: file.id, preset_key: result.key, format: result.format, ts: Date.now() }, { file_id: file.id, preset_key: result.key, format: result.format, ts: Date.now() },
deps.env, deps.env,
); );
await markRecovered(
deps.env,
{
file_id: file.id,
preset_key: result.key,
format: result.format,
ms: result.ms,
},
{ records: failureRecords, logger: deps.logger },
);
if (result.cached) run.hits += 1; if (result.cached) run.hits += 1;
else run.generated += 1; else run.generated += 1;
} else { } else {
run.failed += 1; run.failed += 1;
await journalFailure(
deps.env,
{
file_id: file.id,
preset_key: result.key,
format: result.format,
error: result.error,
ms: result.ms,
},
{ records: failureRecords, logger: deps.logger },
);
} }
run.done += 1; run.done += 1;

649
src/lib/failures.test.ts
View file

@ -1,649 +0,0 @@
import assert from 'node:assert/strict';
import { spawn } from 'node:child_process';
import { appendFile, mkdir, mkdtemp, readFile, readdir, rename, rm, utimes, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { test } from 'node:test';
import {
acquireFailureLock,
appendFailure,
compactFailures,
enrichFailures,
failuresFilePath,
failuresLockPath,
failuresReclaimPath,
listFailures,
parseFailureQuery,
releaseReclaimGuard,
replayFailures,
} from './failures';
import { ensureStateDir } from './state';
import type { FailureEvent, FailureRecord } from './types';
async function withTempEnv(run: (env: Record<string, string>) => Promise<void>): Promise<void> {
const dir = await mkdtemp(join(tmpdir(), 'image-previews-failures-'));
try {
await run({ IMAGE_PREVIEWS_STATE_DIR: dir });
} finally {
await rm(dir, { recursive: true, force: true });
}
}
async function deadPid(): Promise<number> {
const child = spawn(process.execPath, [ '-e', '' ], { stdio: 'ignore' });
const pid = child.pid;
if (!pid) throw new Error('child process has no pid');
await new Promise<void>((resolve, reject) => {
child.once('exit', () => resolve());
child.once('error', reject);
});
return pid;
}
async function sleep(ms: number): Promise<void> {
await new Promise((resolve) => setTimeout(resolve, ms));
}
function event(overrides: Partial<FailureEvent> & Pick<FailureEvent, 'type' | 'ts' | 'file_id'>): FailureEvent {
return {
type: overrides.type,
ts: overrides.ts,
file_id: overrides.file_id,
preset_key: overrides.preset_key ?? 'system-medium-cover',
format: overrides.format === undefined ? 'avif' : overrides.format,
attempt: overrides.attempt ?? 1,
next_retry_at: overrides.next_retry_at,
error: overrides.error,
ms: overrides.ms,
};
}
async function waitForGuardOwner(env: Record<string, string>, token: string): Promise<void> {
const guard = failuresReclaimPath(env);
const deadline = Date.now() + 1000;
while (Date.now() < deadline) {
try {
const owner = JSON.parse(await readFile(`${guard}/owner.json`, 'utf8')) as { token?: string };
if (owner.token === token) return;
} catch {
await sleep(5);
continue;
}
await sleep(5);
}
throw new Error(`guard owner ${token} was not restored`);
}
async function writeLock(env: Record<string, string>, owner: Record<string, unknown> = {}): Promise<void> {
await writeFile(
failuresLockPath(env),
JSON.stringify({
token: 'test-owner',
pid: process.pid,
bootId: null,
startTime: null,
ts: Date.now(),
...owner,
}),
'utf8',
);
}
test('appendFailure + replayFailures derive the latest state per task', async () => {
await withTempEnv(async (env) => {
await appendFailure(
event({ type: 'failed', ts: 1000, file_id: 'f1', attempt: 1, next_retry_at: 2000, error: 'boom', ms: 10 }),
env,
);
await appendFailure(
event({ type: 'failed', ts: 3000, file_id: 'f1', attempt: 2, next_retry_at: null, error: 'boom2', ms: 20 }),
env,
);
await appendFailure(event({ type: 'recovered', ts: 4000, file_id: 'f1', attempt: 3, ms: 5 }), env);
await appendFailure(event({ type: 'failed', ts: 5000, file_id: 'f2', format: null }), env);
const records = await replayFailures(env);
assert.equal(records.length, 2);
const recovered = records.find((record) => record.file_id === 'f1');
assert.equal(recovered?.status, 'recovered');
assert.equal(recovered?.attempts, 3);
assert.equal(recovered?.first_failed_at, 1000);
assert.equal(recovered?.last_failed_at, 3000);
assert.equal(recovered?.next_retry_at, null);
assert.equal(recovered?.error, 'boom2');
assert.equal(recovered?.ms, 5);
const failed = records.find((record) => record.file_id === 'f2');
assert.equal(failed?.status, 'failed');
assert.equal(failed?.attempts, 1);
assert.equal(failed?.format, null);
});
});
test('replayFailures tolerates corrupt lines and is idempotent', async () => {
await withTempEnv(async (env) => {
await ensureStateDir(env);
await appendFile(failuresFilePath(env), 'not-json\n{"type":"nope"}\n', 'utf8');
await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env);
const first = await replayFailures(env);
const second = await replayFailures(env);
assert.equal(first.length, 1);
assert.deepEqual(first, second);
});
});
test('compactFailures keeps unresolved rows and the last 1,000 recovered rows', async () => {
await withTempEnv(async (env) => {
await ensureStateDir(env);
const lines: string[] = [];
for (let index = 0; index < 1002; index += 1) {
lines.push(
JSON.stringify(
event({
type: 'recovered',
ts: index + 1,
file_id: `recovered-${index}`,
attempt: 2,
}),
),
);
}
lines.push(JSON.stringify(event({ type: 'failed', ts: 5000, file_id: 'unresolved-a', next_retry_at: null })));
lines.push(JSON.stringify(event({ type: 'failed', ts: 5001, file_id: 'unresolved-b', next_retry_at: 6000 })));
await appendFile(failuresFilePath(env), `${lines.join('\n')}\n`, 'utf8');
const stats = await compactFailures(env);
const records = await replayFailures(env);
assert.notEqual(stats, false);
if (stats === false) throw new Error('compaction did not acquire the lock');
assert.equal(stats.kept, 1002);
assert.equal(stats.dropped, 2);
assert.equal(records.length, 1002);
assert.ok(records.some((record) => record.file_id === 'unresolved-a' && record.status === 'failed'));
assert.ok(records.some((record) => record.file_id === 'unresolved-b' && record.status === 'failed'));
assert.ok(!records.some((record) => record.file_id === 'recovered-0'));
assert.ok(!records.some((record) => record.file_id === 'recovered-1'));
assert.ok(records.some((record) => record.file_id === 'recovered-2'));
assert.ok(records.some((record) => record.file_id === 'recovered-1001'));
});
});
test('listFailures filters by derived status and paginates', async () => {
await withTempEnv(async (env) => {
await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'exhausted', next_retry_at: null, error: 'a' }), env);
await appendFailure(event({ type: 'failed', ts: 2000, file_id: 'pending', next_retry_at: 3000, error: 'b' }), env);
await appendFailure(event({ type: 'recovered', ts: 3000, file_id: 'done', attempt: 2 }), env);
const exhausted = await listFailures(env, { status: 'failed', limit: 50, offset: 0 });
assert.equal(exhausted.total, 1);
assert.equal(exhausted.records[0]?.file_id, 'exhausted');
assert.equal(exhausted.records[0]?.status, 'failed');
const pending = await listFailures(env, { status: 'retry', limit: 50, offset: 0 });
assert.equal(pending.total, 1);
assert.equal(pending.records[0]?.file_id, 'pending');
assert.equal(pending.records[0]?.status, 'retry');
const recovered = await listFailures(env, { status: 'recovered', limit: 50, offset: 0 });
assert.equal(recovered.total, 1);
assert.equal(recovered.records[0]?.file_id, 'done');
const all = await listFailures(env, { limit: 2, offset: 0 });
assert.equal(all.total, 3);
assert.equal(all.limit, 2);
assert.equal(all.offset, 0);
assert.equal(all.records.length, 2);
assert.equal(all.records[0]?.file_id, 'done');
const page = await listFailures(env, { limit: 1, offset: 1 });
assert.equal(page.total, 3);
assert.equal(page.records.length, 1);
assert.equal(page.records[0]?.file_id, 'pending');
const clamped = await listFailures(env, { limit: 999, offset: 0 });
assert.equal(clamped.limit, 200);
});
});
test('replay resets a new failure cycle and ignores duplicate failed events', async () => {
await withTempEnv(async (env) => {
await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1', attempt: 1, next_retry_at: 2000 }), env);
await appendFailure(event({ type: 'recovered', ts: 2000, file_id: 'f1', attempt: 2 }), env);
await appendFailure(
event({ type: 'failed', ts: 3000, file_id: 'f1', attempt: 1, next_retry_at: 4000, error: 'again' }),
env,
);
await appendFailure(
event({ type: 'failed', ts: 3000, file_id: 'f1', attempt: 1, next_retry_at: 4000, error: 'again' }),
env,
);
const records = await replayFailures(env);
const record = records.find((candidate) => candidate.file_id === 'f1');
assert.equal(record?.attempts, 1);
assert.equal(record?.status, 'failed');
assert.equal(record?.first_failed_at, 3000);
assert.equal(record?.last_failed_at, 3000);
assert.equal(record?.next_retry_at, 4000);
});
});
test('compactFailures returns false and leaves the file untouched when a live owner holds the lock', async () => {
await withTempEnv(async (env) => {
await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env);
const held = await acquireFailureLock(env);
const before = await readFile(failuresFilePath(env), 'utf8');
try {
const result = await compactFailures(env, { timeoutMs: 30, waitMs: 5 });
assert.equal(result, false);
assert.equal(await readFile(failuresFilePath(env), 'utf8'), before);
} finally {
await held.release();
}
});
});
test('appendFailure returns false and leaves the file unchanged when a live owner holds the lock', async () => {
await withTempEnv(async (env) => {
await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env);
const held = await acquireFailureLock(env);
const before = await readFile(failuresFilePath(env), 'utf8');
const warnings: Array<Record<string, unknown>> = [];
try {
const result = await appendFailure(
event({ type: 'failed', ts: 2000, file_id: 'f2' }),
env,
{
timeoutMs: 30,
waitMs: 5,
logger: { warn: (payload: Record<string, unknown>) => warnings.push(payload) },
},
);
assert.equal(result, false);
assert.equal(await readFile(failuresFilePath(env), 'utf8'), before);
assert.equal(warnings[0]?.['fileId'], 'f2');
assert.equal(warnings[0]?.['preset_key'], 'system-medium-cover');
assert.equal(warnings[0]?.['format'], 'avif');
} finally {
await held.release();
}
});
});
test('appendFailure breaks a dead-pid lock and appends', async () => {
await withTempEnv(async (env) => {
const pid = await deadPid();
await writeLock(env, { token: 'dead-owner', pid });
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
timeoutMs: 500,
waitMs: 5,
});
assert.equal(result, true);
assert.equal((await replayFailures(env)).length, 1);
});
});
test('release with a foreign token does not remove the lock', async () => {
await withTempEnv(async (env) => {
const held = await acquireFailureLock(env);
await writeLock(env, { token: 'foreign-owner' });
await held.release();
const owner = JSON.parse(await readFile(failuresLockPath(env), 'utf8')) as {
token?: string;
};
assert.equal(owner.token, 'foreign-owner');
await rm(failuresLockPath(env), { force: true });
});
});
test('missing owner younger than grace is not reclaimed', async () => {
await withTempEnv(async (env) => {
await writeFile(failuresLockPath(env), '', 'utf8');
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
timeoutMs: 30,
waitMs: 5,
graceMs: 100,
});
assert.equal(result, false);
assert.equal(await readFile(failuresLockPath(env), 'utf8'), '');
});
});
test('missing owner older than grace is reclaimed', async () => {
await withTempEnv(async (env) => {
await writeFile(failuresLockPath(env), '', 'utf8');
await utimes(failuresLockPath(env), new Date(0), new Date(0));
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
timeoutMs: 500,
waitMs: 5,
graceMs: 10,
});
assert.equal(result, true);
assert.equal((await replayFailures(env)).length, 1);
});
});
test('corrupt owner older than grace is treated like a missing owner', async () => {
await withTempEnv(async (env) => {
await writeFile(failuresLockPath(env), '{not-json', 'utf8');
await utimes(failuresLockPath(env), new Date(0), new Date(0));
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
timeoutMs: 500,
waitMs: 5,
graceMs: 10,
});
assert.equal(result, true);
assert.equal((await replayFailures(env)).length, 1);
});
});
test('PID reuse with a foreign bootId is reclaimed', async () => {
await withTempEnv(async (env) => {
await writeLock(env, { token: 'reused-boot', bootId: 'foreign-boot' });
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
timeoutMs: 500,
waitMs: 5,
readBootId: async () => 'current-boot',
readStartTime: async () => 'current-start',
});
assert.equal(result, true);
assert.equal((await replayFailures(env)).length, 1);
});
});
test('PID reuse with a foreign startTime is reclaimed', async () => {
await withTempEnv(async (env) => {
await writeLock(env, { token: 'reused-start', startTime: 'foreign-start' });
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
timeoutMs: 500,
waitMs: 5,
readBootId: async () => null,
readStartTime: async () => 'current-start',
});
assert.equal(result, true);
assert.equal((await replayFailures(env)).length, 1);
});
});
test('reclaim waits for the guard and a stale guard is removed', async () => {
await withTempEnv(async (env) => {
const pid = await deadPid();
await writeLock(env, { token: 'dead-owner', pid });
const guard = failuresReclaimPath(env);
await mkdir(guard);
const blocked = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
timeoutMs: 30,
waitMs: 5,
graceMs: 0,
guardStaleMs: 30_000,
});
assert.equal(blocked, false);
assert.match(await readFile(failuresLockPath(env), 'utf8'), /dead-owner/);
const recovered = await appendFailure(event({ type: 'failed', ts: 2000, file_id: 'f2' }), env, {
timeoutMs: 500,
waitMs: 5,
graceMs: 0,
guardStaleMs: 30_000,
now: () => Date.now() + 31_000,
});
assert.equal(recovered, true);
assert.equal((await replayFailures(env)).length, 1);
const names = await readdir(env['IMAGE_PREVIEWS_STATE_DIR'] as string);
assert.ok(!names.some((name) => name.startsWith('failures.lock.reclaim')));
});
});
test('two waiters over the same stale guard reclaim once and keep both events', async () => {
await withTempEnv(async (env) => {
const pid = await deadPid();
await writeLock(env, { token: 'dead-owner', pid });
const guard = failuresReclaimPath(env);
await mkdir(guard);
await utimes(guard, new Date(0), new Date(0));
const options = { timeoutMs: 2000, waitMs: 5, graceMs: 0, guardStaleMs: 30_000 };
const results = await Promise.all([
appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, options),
appendFailure(event({ type: 'failed', ts: 2000, file_id: 'f2' }), env, options),
]);
assert.deepEqual(results, [ true, true ]);
assert.equal((await replayFailures(env)).length, 2);
const names = await readdir(env['IMAGE_PREVIEWS_STATE_DIR'] as string);
assert.ok(!names.some((name) => name.startsWith('failures.lock.reclaim')));
});
});
test('a successor guard replaced between observation and rename is restored', async () => {
await withTempEnv(async (env) => {
const pid = await deadPid();
await writeLock(env, { token: 'dead-owner', pid });
const guard = failuresReclaimPath(env);
await mkdir(guard);
await writeFile(
`${guard}/owner.json`,
JSON.stringify({ token: 'stale-guard', pid: process.pid, ts: 0 }),
'utf8',
);
await utimes(guard, new Date(0), new Date(0));
let observedResolve: () => void = () => {};
const observed = new Promise<void>((resolve) => {
observedResolve = resolve;
});
let resumeResolve: () => void = () => {};
const resume = new Promise<void>((resolve) => {
resumeResolve = resolve;
});
let rejectedResolve: () => void = () => {};
const rejected = new Promise<void>((resolve) => {
rejectedResolve = resolve;
});
let releaseResolve: () => void = () => {};
const release = new Promise<void>((resolve) => {
releaseResolve = resolve;
});
const first = appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
timeoutMs: 2000,
waitMs: 5,
graceMs: 0,
guardStaleMs: 30_000,
onStaleGuardObserved: async () => {
observedResolve();
await resume;
},
onGuardRejected: async () => {
rejectedResolve();
await release;
},
});
await observed;
const replaced = `${guard}.replaced`;
await rename(guard, replaced);
await mkdir(guard);
await writeFile(
`${guard}/owner.json`,
JSON.stringify({ token: 'successor-guard', pid: process.pid, ts: Date.now() }),
'utf8',
);
resumeResolve();
await rejected;
await waitForGuardOwner(env, 'successor-guard');
const second = appendFailure(event({ type: 'failed', ts: 2000, file_id: 'f2' }), env, {
timeoutMs: 2000,
waitMs: 5,
graceMs: 0,
guardStaleMs: 30_000,
});
await rm(guard, { recursive: true, force: true });
releaseResolve();
const results = await Promise.all([ first, second ]);
assert.deepEqual(results, [ true, true ]);
assert.equal((await replayFailures(env)).length, 2);
await rm(replaced, { recursive: true, force: true });
});
});
test('foreign-token guard release does not remove a successor guard', async () => {
await withTempEnv(async (env) => {
const guard = failuresReclaimPath(env);
await mkdir(guard);
await writeFile(
`${guard}/owner.json`,
JSON.stringify({ token: 'successor-guard', pid: process.pid, ts: Date.now() }),
'utf8',
);
await releaseReclaimGuard(env, 'old-guard');
const owner = JSON.parse(await readFile(`${guard}/owner.json`, 'utf8')) as { token?: string };
assert.equal(owner.token, 'successor-guard');
await releaseReclaimGuard(env, 'successor-guard');
await assert.rejects(readFile(`${guard}/owner.json`, 'utf8'));
});
});
test('concurrent reclaim of a dead lock loses no event', async () => {
await withTempEnv(async (env) => {
const pid = await deadPid();
await writeLock(env, { token: 'dead-owner', pid });
const results = await Promise.all([
appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
timeoutMs: 1000,
waitMs: 5,
}),
appendFailure(event({ type: 'failed', ts: 2000, file_id: 'f2' }), env, {
timeoutMs: 1000,
waitMs: 5,
}),
]);
assert.deepEqual(results, [ true, true ]);
assert.equal((await replayFailures(env)).length, 2);
});
});
test('50 appends racing one compaction with a temporarily held lock keep every key', async () => {
await withTempEnv(async (env) => {
const held = await acquireFailureLock(env);
const appends = Array.from({ length: 50 }, (_, index) =>
appendFailure(event({ type: 'failed', ts: 1000 + index, file_id: `race-${index}`, next_retry_at: null }), env),
);
const compaction = compactFailures(env);
await sleep(50);
await held.release();
await Promise.all([ compaction, ...appends ]);
const records = await replayFailures(env);
assert.equal(records.length, 50);
for (let index = 0; index < 50; index += 1) {
assert.ok(records.some((record) => record.file_id === `race-${index}`));
}
const names = await readdir(env['IMAGE_PREVIEWS_STATE_DIR'] as string);
assert.ok(!names.some((name) => name.endsWith('.tmp')));
});
});
test('parseFailureQuery validates status and clamps limit/offset', () => {
assert.deepEqual(parseFailureQuery({ status: 'failed', limit: '999', offset: '-3' }), {
status: 'failed',
limit: 200,
offset: 0,
});
assert.deepEqual(parseFailureQuery({}), { status: null, limit: 50, offset: 0 });
assert.deepEqual(parseFailureQuery({ status: 'retry', limit: '0', offset: '2' }), {
status: 'retry',
limit: 1,
offset: 2,
});
assert.throws(() => parseFailureQuery({ status: 'bogus' }), /status must be failed, recovered or retry/);
});
test('enrichFailures adds file metadata, timestamps and human durations', () => {
const records: FailureRecord[] = [
{
file_id: 'f1',
preset_key: 'system-medium-cover',
format: 'avif',
attempts: 2,
status: 'failed',
first_failed_at: 1000,
last_failed_at: 2000,
next_retry_at: 3000,
error: 'boom',
ms: 12,
},
];
const files = new Map([
[ 'f1', { filename_download: 'photo.jpg', type: 'image/jpeg', folder: 'folder-1' } ],
]);
const enriched = enrichFailures(records, files);
const record = enriched[0];
assert.deepEqual(record?.file, {
filename_download: 'photo.jpg',
type: 'image/jpeg',
folder: 'folder-1',
});
assert.equal(record?.attempts, 2);
assert.equal(record?.first_failed_at, 1000);
assert.equal(record?.last_failed_at, 2000);
assert.equal(record?.next_retry_at, 3000);
assert.equal(record?.status, 'failed');
assert.equal(record?.error, 'boom');
assert.equal(record?.first_failed_at_iso, new Date(1000).toISOString());
assert.equal(record?.last_failed_at_iso, new Date(2000).toISOString());
assert.equal(record?.next_retry_at_iso, new Date(3000).toISOString());
assert.equal(record?.next_retry_at_human, new Date(3000).toLocaleString());
assert.equal(record?.ms_human, '12 ms');
});

584
src/lib/failures.ts
View file

@ -1,584 +0,0 @@
import { randomUUID } from 'node:crypto';
import { appendFile, mkdir, readFile, rename, rm, stat, unlink, writeFile } from 'node:fs/promises';
import { humanMs, toIso } from './format';
import type { EnvLike } from './settings';
import { completedKeyOf, ensureStateDir, stateFilePath } from './state';
import type {
FailureEnrichedRecord,
FailureEvent,
FailureFileMeta,
FailureListQuery,
FailureListResult,
FailureRecord,
FailureStatus,
} from './types';
const FAILURES_FILE = 'failures.jsonl';
const FAILURES_LOCK = 'failures.lock';
const RECOVERED_KEEP = 1000;
const LOCK_TIMEOUT_MS = 10_000;
const LOCK_WAIT_MS = 10;
const LOCK_GRACE_MS = 2_000;
const GUARD_STALE_MS = 30_000;
const FAILURE_STATUSES: FailureStatus[] = [ 'failed', 'recovered', 'retry' ];
export interface FailureLockOptions {
timeoutMs?: number;
waitMs?: number;
graceMs?: number;
guardStaleMs?: number;
now?: () => number;
onStaleGuardObserved?: () => Promise<void>;
onGuardRejected?: () => Promise<void>;
logger?: any;
readBootId?: () => Promise<string | null>;
readStartTime?: (pid: number) => Promise<string | null>;
}
export interface FailureLock {
acquired: boolean;
token: string;
release: () => Promise<void>;
}
interface LockOwner {
token: string;
pid: number;
bootId: string | null;
startTime: string | null;
ts: number;
}
export function failuresFilePath(env: EnvLike): string {
return stateFilePath(env, FAILURES_FILE);
}
export function failuresLockPath(env: EnvLike): string {
return stateFilePath(env, FAILURES_LOCK);
}
export function failuresReclaimPath(env: EnvLike): string {
return `${failuresLockPath(env)}.reclaim`;
}
async function sleep(ms: number): Promise<void> {
await new Promise((resolve) => setTimeout(resolve, ms));
}
async function fileMtime(path: string): Promise<number | null> {
try {
return (await stat(path)).mtimeMs;
} catch {
return null;
}
}
async function defaultReadBootId(): Promise<string | null> {
try {
const value = (await readFile('/proc/sys/kernel/random/boot_id', 'utf8')).trim();
return value.length > 0 ? value : null;
} catch {
return null;
}
}
async function defaultReadStartTime(pid: number): Promise<string | null> {
try {
const raw = await readFile(`/proc/${pid}/stat`, 'utf8');
const close = raw.lastIndexOf(')');
if (close < 0) return null;
const fields = raw.slice(close + 1).trim().split(/\s+/);
return fields[19] ?? null;
} catch {
return null;
}
}
async function readLockOwner(path: string): Promise<LockOwner | null> {
try {
const parsed = JSON.parse(await readFile(path, 'utf8')) as Record<string, unknown>;
const token = parsed['token'];
const pid = parsed['pid'];
if (typeof token !== 'string' || typeof pid !== 'number') return null;
const bootId = typeof parsed['bootId'] === 'string' ? parsed['bootId'] : null;
const startTime = typeof parsed['startTime'] === 'string' ? parsed['startTime'] : null;
const ts = typeof parsed['ts'] === 'number' ? parsed['ts'] : 0;
return { token, pid, bootId, startTime, ts };
} catch {
return null;
}
}
function isOwnerAlive(owner: LockOwner, bootId: string | null, startTime: string | null): boolean {
if (!Number.isInteger(owner.pid) || owner.pid <= 0) return false;
try {
process.kill(owner.pid, 0);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ESRCH') return false;
}
if (owner.bootId && bootId && owner.bootId !== bootId) return false;
if (owner.startTime && startTime && owner.startTime !== startTime) return false;
return true;
}
async function unlinkLock(path: string): Promise<boolean> {
try {
await unlink(path);
return true;
} catch {
return false;
}
}
async function releaseFailureLock(env: EnvLike, token: string): Promise<void> {
const owner = await readLockOwner(failuresLockPath(env));
if (!owner || owner.token !== token) return;
await unlinkLock(failuresLockPath(env));
}
export async function releaseReclaimGuard(env: EnvLike, token: string): Promise<void> {
const guardPath = failuresReclaimPath(env);
const owner = await readLockOwner(`${guardPath}/owner.json`);
if (!owner || owner.token !== token) return;
try {
await rm(guardPath, { recursive: true, force: true });
} catch {
return;
}
}
async function withReclaimGuard(
env: EnvLike,
options: FailureLockOptions,
action: () => Promise<boolean>,
): Promise<boolean> {
const guardPath = failuresReclaimPath(env);
const timeoutMs = options.timeoutMs ?? LOCK_TIMEOUT_MS;
const waitMs = options.waitMs ?? LOCK_WAIT_MS;
const guardStaleMs = options.guardStaleMs ?? GUARD_STALE_MS;
const now = options.now ?? (() => Date.now());
const token = randomUUID();
const deadline = now() + timeoutMs;
const quarantined: string[] = [];
let owned = false;
try {
while (true) {
try {
await mkdir(guardPath);
try {
await writeFile(
`${guardPath}/owner.json`,
JSON.stringify({ token, pid: process.pid, ts: now() }),
'utf8',
);
} catch (error) {
await rm(guardPath, { recursive: true, force: true });
throw error;
}
owned = true;
break;
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'EEXIST') return false;
const observedMtime = await fileMtime(guardPath);
const observedOwner =
observedMtime === null ? null : await readLockOwner(`${guardPath}/owner.json`);
if (observedMtime !== null && now() - observedMtime > guardStaleMs) {
await options.onStaleGuardObserved?.();
const stalePath = `${guardPath}.stale.${process.pid}.${randomUUID()}`;
try {
await rename(guardPath, stalePath);
} catch {
if (now() >= deadline) return false;
await sleep(waitMs);
continue;
}
const movedMtime = await fileMtime(stalePath);
const movedOwner = await readLockOwner(`${stalePath}/owner.json`);
const ownerMatches =
(observedOwner?.token ?? null) === (movedOwner?.token ?? null);
const mtimeMatches =
observedMtime !== null && movedMtime !== null && movedMtime === observedMtime;
if (!ownerMatches || !mtimeMatches) {
try {
await rename(stalePath, guardPath);
} catch {
quarantined.push(stalePath);
return false;
}
await options.onGuardRejected?.();
if (now() >= deadline) return false;
await sleep(waitMs);
continue;
}
quarantined.push(stalePath);
continue;
}
if (now() >= deadline) return false;
await sleep(waitMs);
}
}
return await action();
} finally {
if (owned) await releaseReclaimGuard(env, token);
for (const path of quarantined) {
try {
await rm(path, { recursive: true, force: true });
} catch {
continue;
}
}
}
}
async function reclaimLock(env: EnvLike, options: FailureLockOptions): Promise<boolean> {
const lockPath = failuresLockPath(env);
const graceMs = options.graceMs ?? LOCK_GRACE_MS;
return withReclaimGuard(env, options, async () => {
const mtime = await fileMtime(lockPath);
if (mtime === null) return false;
const owner = await readLockOwner(lockPath);
if (!owner) {
if (Date.now() - mtime <= graceMs) return false;
return unlinkLock(lockPath);
}
const bootId = await (options.readBootId ?? defaultReadBootId)();
const startTime = await (options.readStartTime ?? defaultReadStartTime)(owner.pid);
if (isOwnerAlive(owner, bootId, startTime)) return false;
return unlinkLock(lockPath);
});
}
export async function acquireFailureLock(
env: EnvLike,
options: FailureLockOptions = {},
): Promise<FailureLock> {
await ensureStateDir(env);
const timeoutMs = options.timeoutMs ?? LOCK_TIMEOUT_MS;
const waitMs = options.waitMs ?? LOCK_WAIT_MS;
const graceMs = options.graceMs ?? LOCK_GRACE_MS;
const token = randomUUID();
const bootId = await (options.readBootId ?? defaultReadBootId)();
const startTime = await (options.readStartTime ?? defaultReadStartTime)(process.pid);
const owner: LockOwner = { token, pid: process.pid, bootId, startTime, ts: Date.now() };
const deadline = Date.now() + timeoutMs;
while (true) {
try {
await writeFile(failuresLockPath(env), JSON.stringify(owner), { flag: 'wx' });
return {
acquired: true,
token,
release: async () => {
await releaseFailureLock(env, token);
},
};
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'EEXIST') {
return { acquired: false, token, release: async () => {} };
}
const existing = await readLockOwner(failuresLockPath(env));
if (existing) {
const existingStartTime = await (options.readStartTime ?? defaultReadStartTime)(existing.pid);
if (!isOwnerAlive(existing, bootId, existingStartTime)) {
const reclaimed = await reclaimLock(env, options);
if (Date.now() >= deadline) return { acquired: false, token, release: async () => {} };
if (!reclaimed) await sleep(waitMs);
continue;
}
} else {
const mtime = await fileMtime(failuresLockPath(env));
if (mtime !== null && Date.now() - mtime > graceMs) {
const reclaimed = await reclaimLock(env, options);
if (Date.now() >= deadline) return { acquired: false, token, release: async () => {} };
if (!reclaimed) await sleep(waitMs);
continue;
}
}
if (Date.now() >= deadline) return { acquired: false, token, release: async () => {} };
await sleep(waitMs);
}
}
}
export async function appendFailure(
event: FailureEvent,
env: EnvLike,
options: FailureLockOptions = {},
): Promise<boolean> {
await ensureStateDir(env);
const lock = await acquireFailureLock(env, options);
if (!lock.acquired) {
options.logger?.warn?.(
{ fileId: event.file_id, preset_key: event.preset_key, format: event.format },
'image-previews: could not journal failure',
);
return false;
}
try {
await appendFile(failuresFilePath(env), `${JSON.stringify(event)}\n`, 'utf8');
return true;
} finally {
await lock.release();
}
}
function isFailureEvent(value: unknown): value is FailureEvent {
if (!value || typeof value !== 'object') return false;
const event = value as Record<string, unknown>;
const type = event['type'];
if (type !== 'failed' && type !== 'retry' && type !== 'recovered') return false;
if (typeof event['ts'] !== 'number' || !Number.isFinite(event['ts'])) return false;
if (typeof event['file_id'] !== 'string' || event['file_id'].length === 0) return false;
if (typeof event['preset_key'] !== 'string' || event['preset_key'].length === 0) return false;
if (event['format'] !== null && typeof event['format'] !== 'string') return false;
if (typeof event['attempt'] !== 'number' || !Number.isFinite(event['attempt'])) return false;
return true;
}
export async function readFailureEvents(env: EnvLike): Promise<FailureEvent[]> {
const events: FailureEvent[] = [];
try {
const raw = await readFile(failuresFilePath(env), 'utf8');
for (const line of raw.split('\n')) {
if (!line.trim()) continue;
try {
const parsed = JSON.parse(line) as unknown;
if (isFailureEvent(parsed)) events.push(parsed);
} catch {
continue;
}
}
} catch {
return events;
}
return events;
}
function eventKey(event: FailureEvent): string {
return completedKeyOf(event.file_id, event.preset_key, event.format);
}
export function replayFailureEvents(events: FailureEvent[]): FailureRecord[] {
const records = new Map<string, FailureRecord>();
for (const event of events) {
const key = eventKey(event);
const previous = records.get(key);
const startsCycle = previous === undefined || previous.status === 'recovered';
const firstFailedAt = startsCycle ? event.ts : previous.first_failed_at;
const lastFailedAt = startsCycle ? event.ts : previous.last_failed_at;
const nextRetryAt = event.next_retry_at ?? null;
const error = event.error ?? previous?.error ?? null;
const ms = event.ms ?? previous?.ms ?? null;
if (event.type === 'failed') {
records.set(key, {
file_id: event.file_id,
preset_key: event.preset_key,
format: event.format,
attempts: event.attempt,
status: 'failed',
first_failed_at: firstFailedAt,
last_failed_at: event.ts,
next_retry_at: nextRetryAt,
error,
ms,
});
} else if (event.type === 'retry') {
records.set(key, {
file_id: event.file_id,
preset_key: event.preset_key,
format: event.format,
attempts: event.attempt,
status: previous?.status === 'recovered' ? 'failed' : (previous?.status ?? 'failed'),
first_failed_at: firstFailedAt,
last_failed_at: lastFailedAt,
next_retry_at: nextRetryAt,
error,
ms,
});
} else {
records.set(key, {
file_id: event.file_id,
preset_key: event.preset_key,
format: event.format,
attempts: event.attempt,
status: 'recovered',
first_failed_at: firstFailedAt,
last_failed_at: lastFailedAt,
next_retry_at: null,
error,
ms,
});
}
}
return [ ...records.values() ];
}
export async function replayFailures(env: EnvLike): Promise<FailureRecord[]> {
return replayFailureEvents(await readFailureEvents(env));
}
export async function replayFailureMap(env: EnvLike): Promise<Map<string, FailureRecord>> {
const records = await replayFailures(env);
return new Map(records.map((record) => [ completedKeyOf(record.file_id, record.preset_key, record.format), record ]));
}
export function failureStatusOf(record: FailureRecord): FailureRecord['status'] {
if (record.status === 'recovered') return 'recovered';
return record.next_retry_at !== null ? 'retry' : 'failed';
}
function withDerivedStatus(record: FailureRecord): FailureRecord {
const status = failureStatusOf(record);
return status === record.status ? record : { ...record, status };
}
export async function listFailures(env: EnvLike, query: FailureListQuery): Promise<FailureListResult> {
const limit = Math.max(1, Math.min(200, Math.trunc(query.limit) || 50));
const offset = Math.max(0, Math.trunc(query.offset) || 0);
const records = (await replayFailures(env))
.map(withDerivedStatus)
.sort((a, b) => b.last_failed_at - a.last_failed_at || eventKeyLike(a).localeCompare(eventKeyLike(b)));
const filtered = query.status ? records.filter((record) => record.status === query.status) : records;
return {
records: filtered.slice(offset, offset + limit),
total: filtered.length,
limit,
offset,
};
}
function eventKeyLike(record: FailureRecord): string {
return completedKeyOf(record.file_id, record.preset_key, record.format);
}
export function parseFailureQuery(input: { status?: unknown; limit?: unknown; offset?: unknown }): FailureListQuery {
const rawStatus = typeof input.status === 'string' ? input.status : null;
if (rawStatus && !FAILURE_STATUSES.includes(rawStatus as FailureStatus)) {
throw new Error('status must be failed, recovered or retry');
}
const limitRaw = Number(input.limit);
const offsetRaw = Number(input.offset);
return {
status: (rawStatus as FailureStatus | null) ?? null,
limit: Number.isFinite(limitRaw) ? Math.min(200, Math.max(1, Math.trunc(limitRaw))) : 50,
offset: Number.isFinite(offsetRaw) ? Math.max(0, Math.trunc(offsetRaw)) : 0,
};
}
export function enrichFailures(
records: FailureRecord[],
files: Map<string, FailureFileMeta>,
): FailureEnrichedRecord[] {
return records.map((record) => {
const file = files.get(record.file_id);
return {
...record,
file: {
filename_download: file?.filename_download ?? null,
type: file?.type ?? null,
folder: file?.folder ?? null,
},
first_failed_at_iso: toIso(record.first_failed_at),
first_failed_at_human: new Date(record.first_failed_at).toLocaleString(),
last_failed_at_iso: toIso(record.last_failed_at),
last_failed_at_human: new Date(record.last_failed_at).toLocaleString(),
next_retry_at_iso: toIso(record.next_retry_at),
next_retry_at_human: record.next_retry_at ? new Date(record.next_retry_at).toLocaleString() : null,
ms_human: record.ms === null ? null : humanMs(record.ms),
};
});
}
export async function compactFailures(
env: EnvLike,
options: FailureLockOptions = {},
): Promise<false | { kept: number; dropped: number }> {
const lock = await acquireFailureLock(env, options);
if (!lock.acquired) return false;
try {
const events = await readFailureEvents(env);
const records = replayFailureEvents(events);
const unresolved = records.filter((record) => record.status !== 'recovered');
const recovered = records
.filter((record) => record.status === 'recovered')
.sort((a, b) => b.last_failed_at - a.last_failed_at)
.slice(0, RECOVERED_KEEP);
const keep = new Set([ ...unresolved, ...recovered ].map(eventKeyLike));
const keptEvents = events.filter((event) => keep.has(eventKey(event)));
const content = keptEvents.map((event) => JSON.stringify(event)).join('\n');
const target = failuresFilePath(env);
await ensureStateDir(env);
const temp = `${target}.${process.pid}.${lock.token}.tmp`;
await writeFile(temp, keptEvents.length > 0 ? `${content}\n` : '', 'utf8');
await rename(temp, target);
return {
kept: keep.size,
dropped: Math.max(0, records.length - keep.size),
};
} finally {
await lock.release();
}
}

588
src/lib/retry.test.ts
View file

@ -1,588 +0,0 @@
import assert from 'node:assert/strict';
import { mkdtemp, readFile, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { test } from 'node:test';
import { planFileTasks } from './backfill';
import { appendFailure, failuresFilePath, replayFailures } from './failures';
import {
MAX_ATTEMPTS,
dueRetries,
journalFailure,
nextRetryAt,
runDueRetries,
type RetryDeps,
} from './retry';
import { envDefaults, writeSettings } from './settings';
import { tasksPerFile } from './presets';
import { acquireLock, appendCompleted, completedKeyOf, isLocked, loadCompleted, releaseLock } from './state';
import type { FailureRecord, Preset } from './types';
async function withTempEnv(run: (env: Record<string, string>) => Promise<void>): Promise<void> {
const dir = await mkdtemp(join(tmpdir(), 'image-previews-retry-'));
try {
await run({ IMAGE_PREVIEWS_STATE_DIR: dir });
} finally {
await rm(dir, { recursive: true, force: true });
}
}
interface DepsHarness {
deps: RetryDeps;
getAssetCalls: Array<{ id: string; transformation: Record<string, unknown> }>;
readOneCalls: number;
warnings: Array<Record<string, unknown>>;
infos: Array<Record<string, unknown>>;
}
function makeDeps(
env: Record<string, string>,
files: Record<string, { id: string; type: string }>,
options: { fail?: boolean } = {},
): DepsHarness {
const getAssetCalls: Array<{ id: string; transformation: Record<string, unknown> }> = [];
const warnings: Array<Record<string, unknown>> = [];
const infos: Array<Record<string, unknown>> = [];
const harness = {
getAssetCalls,
readOneCalls: 0,
warnings,
infos,
} as DepsHarness;
class SettingsService {
constructor(_options: Record<string, unknown>) {}
async readSingleton(): Promise<Record<string, unknown>> {
return { storage_asset_presets: [] };
}
}
class ItemsService {
constructor(_options: Record<string, unknown>) {}
async readOne(id: string): Promise<{ id: string; type: string }> {
harness.readOneCalls += 1;
const file = files[id];
if (!file) throw new Error('not found');
return file;
}
}
class AssetsService {
constructor(_options: Record<string, unknown>) {}
async getAsset(id: string, transformation: Record<string, unknown>): Promise<{ stream: { destroy(): void } }> {
getAssetCalls.push({ id, transformation });
if (options.fail) throw new Error('boom');
return { stream: { destroy() {} } };
}
}
harness.deps = {
services: { SettingsService, ItemsService, AssetsService },
database: null,
getSchema: async () => ({}),
env,
logger: {
warn: (payload: Record<string, unknown>) => warnings.push(payload),
info: (payload: Record<string, unknown>) => infos.push(payload),
error: () => {},
},
};
return harness;
}
function failedRecord(overrides: Partial<FailureRecord> & Pick<FailureRecord, 'file_id'>): FailureRecord {
return {
file_id: overrides.file_id,
preset_key: overrides.preset_key ?? 'system-medium-cover',
format: overrides.format === undefined ? 'avif' : overrides.format,
attempts: overrides.attempts ?? 1,
status: overrides.status ?? 'failed',
first_failed_at: overrides.first_failed_at ?? 1000,
last_failed_at: overrides.last_failed_at ?? 1000,
next_retry_at: overrides.next_retry_at === undefined ? 2000 : overrides.next_retry_at,
error: overrides.error ?? null,
ms: overrides.ms ?? null,
};
}
test('nextRetryAt uses 5 then 15 minute backoff and stops after 3 attempts', () => {
const now = 1_000_000;
assert.equal(nextRetryAt(1, now), now + 5 * 60 * 1000);
assert.equal(nextRetryAt(2, now), now + 15 * 60 * 1000);
assert.equal(nextRetryAt(3, now), null);
assert.equal(nextRetryAt(4, now), null);
assert.equal(nextRetryAt(0, now), now + 5 * 60 * 1000);
assert.equal(MAX_ATTEMPTS, 3);
});
test('dueRetries selects only retryable failures whose next retry is due', () => {
const now = 10_000;
const records = [
failedRecord({ file_id: 'due', next_retry_at: now - 1 }),
failedRecord({ file_id: 'future', next_retry_at: now + 1 }),
failedRecord({ file_id: 'exhausted', attempts: 3, next_retry_at: now - 1 }),
failedRecord({ file_id: 'no-schedule', next_retry_at: null }),
failedRecord({ file_id: 'recovered', status: 'recovered', next_retry_at: null }),
];
assert.deepEqual(
dueRetries(records, now).map((record) => record.file_id),
[ 'due' ],
);
});
test('runDueRetries recovers a due pair and appends completion + recovered event', async () => {
await withTempEnv(async (env) => {
const now = Date.now();
await appendFailure(
{
type: 'failed',
ts: now - 1000,
file_id: 'f1',
preset_key: 'system-medium-cover',
format: 'avif',
attempt: 1,
next_retry_at: now - 1,
error: 'boom',
ms: 10,
},
env,
);
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
assert.equal(outcome.status, 'completed');
assert.equal(outcome.processed, 1);
assert.equal(outcome.recovered, 1);
assert.equal(outcome.failed, 0);
assert.equal(harness.getAssetCalls.length, 1);
const transformation = harness.getAssetCalls[0]?.transformation;
assert.equal((transformation?.['transformationParams'] as { key?: string } | undefined)?.key, 'system-medium-cover');
assert.equal(transformation?.['acceptFormat'], 'avif');
const completed = await loadCompleted(env);
assert.ok(completed.has(completedKeyOf('f1', 'system-medium-cover', 'avif')));
const records = await replayFailures(env);
const record = records.find((candidate) => candidate.file_id === 'f1');
assert.equal(record?.status, 'recovered');
assert.equal(record?.attempts, 2);
assert.equal(record?.next_retry_at, null);
assert.equal(await isLocked(env), false);
});
});
test('runDueRetries bumps the attempt and schedules 15 minutes on failure', async () => {
await withTempEnv(async (env) => {
const now = Date.now();
await appendFailure(
{
type: 'failed',
ts: now - 1000,
file_id: 'f1',
preset_key: 'system-medium-cover',
format: 'avif',
attempt: 1,
next_retry_at: now - 1,
error: 'first',
ms: 10,
},
env,
);
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } }, { fail: true });
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
assert.equal(outcome.status, 'completed');
assert.equal(outcome.processed, 1);
assert.equal(outcome.recovered, 0);
assert.equal(outcome.failed, 1);
const records = await replayFailures(env);
const record = records.find((candidate) => candidate.file_id === 'f1');
assert.equal(record?.status, 'failed');
assert.equal(record?.attempts, 2);
assert.equal(record?.error, 'boom');
assert.ok(record?.next_retry_at !== null);
assert.ok((record?.next_retry_at ?? 0) >= now + 14 * 60 * 1000);
assert.ok((record?.next_retry_at ?? 0) <= now + 16 * 60 * 1000);
const completed = await loadCompleted(env);
assert.equal(completed.size, 0);
const warning = harness.warnings.find((entry) => entry['fileId'] === 'f1');
assert.equal(warning?.['preset_key'], 'system-medium-cover');
assert.equal(warning?.['format'], 'avif');
assert.equal(warning?.['attempt'], 2);
assert.equal(warning?.['error'], 'boom');
assert.equal(warning?.['next_retry_at'], record?.next_retry_at);
assert.equal(await isLocked(env), false);
});
});
test('runDueRetries is bounded and groups pairs by file', async () => {
await withTempEnv(async (env) => {
const now = Date.now();
for (let index = 0; index < 5; index += 1) {
await appendFailure(
{
type: 'failed',
ts: now - 1000,
file_id: `f${index}`,
preset_key: 'system-medium-cover',
format: 'avif',
attempt: 1,
next_retry_at: now - 1,
error: 'boom',
},
env,
);
}
const harness = makeDeps(
env,
{
f0: { id: 'f0', type: 'image/jpeg' },
f1: { id: 'f1', type: 'image/jpeg' },
f2: { id: 'f2', type: 'image/jpeg' },
f3: { id: 'f3', type: 'image/jpeg' },
f4: { id: 'f4', type: 'image/jpeg' },
},
);
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true, limit: 2 });
assert.equal(outcome.processed, 2);
assert.equal(outcome.recovered, 2);
assert.equal(harness.getAssetCalls.length, 2);
assert.equal((await replayFailures(env)).filter((record) => record.status === 'recovered').length, 2);
});
});
test('runDueRetries reads each file once for multiple due formats', async () => {
await withTempEnv(async (env) => {
const now = Date.now();
for (const format of [ 'avif', 'webp' ]) {
await appendFailure(
{
type: 'failed',
ts: now - 1000,
file_id: 'f1',
preset_key: 'system-medium-cover',
format,
attempt: 1,
next_retry_at: now - 1,
error: 'boom',
},
env,
);
}
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
assert.equal(outcome.processed, 2);
assert.equal(harness.readOneCalls, 1);
assert.equal(harness.getAssetCalls.length, 2);
});
});
test('runDueRetries marks a missing preset non-retryable without calling assets', async () => {
await withTempEnv(async (env) => {
const now = Date.now();
await appendFailure(
{
type: 'failed',
ts: now - 1000,
file_id: 'f1',
preset_key: 'gone',
format: 'avif',
attempt: 1,
next_retry_at: now - 1,
error: 'boom',
},
env,
);
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
assert.equal(outcome.processed, 1);
assert.equal(outcome.failed, 1);
assert.equal(harness.getAssetCalls.length, 0);
const record = (await replayFailures(env)).find((candidate) => candidate.file_id === 'f1');
assert.equal(record?.status, 'failed');
assert.equal(record?.attempts, MAX_ATTEMPTS);
assert.equal(record?.next_retry_at, null);
assert.match(record?.error ?? '', /unknown preset/);
});
});
test('runDueRetries is skipped when disabled', async () => {
await withTempEnv(async (env) => {
await writeSettings({ ...envDefaults(env), enabled: false }, env);
const harness = makeDeps(env, {});
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
assert.equal(outcome.status, 'skipped');
assert.equal(outcome.reason, 'disabled');
assert.equal(outcome.processed, 0);
});
});
test('runDueRetries does not run or write while run.lock is held', async () => {
await withTempEnv(async (env) => {
const now = Date.now();
await appendFailure(
{
type: 'failed',
ts: now - 1000,
file_id: 'f1',
preset_key: 'system-medium-cover',
format: 'avif',
attempt: 1,
next_retry_at: now - 1,
error: 'boom',
},
env,
);
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
assert.equal(await acquireLock(env), true);
const before = await readFile(failuresFilePath(env), 'utf8');
try {
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
assert.equal(outcome.status, 'skipped');
assert.equal(outcome.reason, 'already-running');
assert.equal(outcome.processed, 0);
assert.equal(harness.getAssetCalls.length, 0);
assert.equal(await readFile(failuresFilePath(env), 'utf8'), before);
assert.equal((await loadCompleted(env)).size, 0);
} finally {
await releaseLock(env);
}
});
});
test('runDueRetries is skipped outside the maintenance window', async () => {
await withTempEnv(async (env) => {
const now = new Date();
const current = now.getHours() * 60 + now.getMinutes();
const start = (current + 60) % 1440;
const end = (start + 1) % 1440;
const time = (minutes: number): string =>
`${String(Math.floor(minutes / 60)).padStart(2, '0')}:${String(minutes % 60).padStart(2, '0')}`;
await writeSettings(
{ ...envDefaults(env), windowStart: time(start), windowEnd: time(end) },
env,
);
await appendFailure(
{
type: 'failed',
ts: Date.now() - 1000,
file_id: 'f1',
preset_key: 'system-medium-cover',
format: 'avif',
attempt: 1,
next_retry_at: Date.now() - 1,
error: 'boom',
},
env,
);
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
const outcome = await runDueRetries(harness.deps, {});
assert.equal(outcome.status, 'skipped');
assert.equal(outcome.reason, 'outside-window');
assert.equal(harness.getAssetCalls.length, 0);
assert.equal((await loadCompleted(env)).size, 0);
});
});
test('fail -> recover -> fail starts a new retry cycle', async () => {
await withTempEnv(async (env) => {
const now = Date.now();
await appendFailure(
{
type: 'failed',
ts: now - 10_000,
file_id: 'f1',
preset_key: 'system-medium-cover',
format: 'avif',
attempt: 1,
next_retry_at: now - 9_000,
error: 'first',
},
env,
);
await appendFailure(
{
type: 'recovered',
ts: now - 8_000,
file_id: 'f1',
preset_key: 'system-medium-cover',
format: 'avif',
attempt: 2,
},
env,
);
await appendFailure(
{
type: 'failed',
ts: now - 1_000,
file_id: 'f1',
preset_key: 'system-medium-cover',
format: 'avif',
attempt: 1,
next_retry_at: now - 1,
error: 'again',
},
env,
);
const due = dueRetries(await replayFailures(env), now);
assert.equal(due.length, 1);
assert.equal(due[0]?.attempts, 1);
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
assert.equal(outcome.processed, 1);
assert.equal(outcome.recovered, 1);
const record = (await replayFailures(env)).find((candidate) => candidate.file_id === 'f1');
assert.equal(record?.status, 'recovered');
assert.equal(record?.attempts, 2);
});
});
test('journalFailure never writes an attempt above MAX_ATTEMPTS', async () => {
await withTempEnv(async (env) => {
const now = Date.now();
await appendFailure(
{
type: 'failed',
ts: now - 1000,
file_id: 'f1',
preset_key: 'system-medium-cover',
format: 'avif',
attempt: MAX_ATTEMPTS,
next_retry_at: null,
error: 'terminal',
},
env,
);
const record = await journalFailure(env, {
file_id: 'f1',
preset_key: 'system-medium-cover',
format: 'avif',
error: 'again',
});
assert.equal(record?.attempts, MAX_ATTEMPTS);
assert.equal(record?.next_retry_at, null);
assert.equal((await replayFailures(env))[0]?.attempts, MAX_ATTEMPTS);
});
});
test('planFileTasks skips unresolved failures and warms recovered ones', () => {
const presets: Preset[] = [ { key: 'system-medium-cover', format: 'auto' } ];
const key = completedKeyOf('f1', 'system-medium-cover', 'avif');
const failed = new Map<string, FailureRecord>([
[ key, failedRecord({ file_id: 'f1', attempts: MAX_ATTEMPTS, next_retry_at: null }) ],
]);
const blocked = planFileTasks('f1', presets, [ 'avif' ], new Set(), failed);
assert.equal(blocked.tasks.length, 0);
assert.equal(blocked.skipped, 1);
assert.equal(blocked.completed, 0);
const recovered = new Map<string, FailureRecord>([
[ key, failedRecord({ file_id: 'f1', status: 'recovered', next_retry_at: null }) ],
]);
const allowed = planFileTasks('f1', presets, [ 'avif' ], new Set(), recovered);
assert.equal(allowed.tasks.length, 1);
assert.equal(allowed.skipped, 0);
});
test('planFileTasks keeps totals stable, resumes state.jsonl, and skips unresolved pairs', async () => {
await withTempEnv(async (env) => {
const presets: Preset[] = [
{ key: 'system-medium-cover', format: 'auto' },
{ key: 'custom-fixed', format: 'webp' },
];
const formats = [ 'avif', 'webp' ];
const total = tasksPerFile(presets, formats);
await appendCompleted(
{ file_id: 'f1', preset_key: 'system-medium-cover', format: 'avif', ts: 1 },
env,
);
const completed = await loadCompleted(env);
const failures = new Map<string, FailureRecord>([
[
completedKeyOf('f1', 'system-medium-cover', 'webp'),
failedRecord({ file_id: 'f1', format: 'webp', attempts: 1, next_retry_at: Date.now() + 1000 }),
],
[
completedKeyOf('f1', 'custom-fixed', 'webp'),
failedRecord({
file_id: 'f1',
preset_key: 'custom-fixed',
format: 'webp',
attempts: MAX_ATTEMPTS,
next_retry_at: null,
}),
],
]);
const plan = planFileTasks('f1', presets, formats, completed, failures);
assert.equal(total, 3);
assert.equal(plan.completed, 1);
assert.equal(plan.skipped, 2);
assert.equal(plan.tasks.length, 0);
assert.equal(plan.completed + plan.skipped + plan.tasks.length, total);
const retryableOnly = new Map<string, FailureRecord>([
[
completedKeyOf('f1', 'custom-fixed', 'webp'),
failedRecord({
file_id: 'f1',
preset_key: 'custom-fixed',
format: 'webp',
attempts: MAX_ATTEMPTS,
next_retry_at: null,
}),
],
]);
const mixed = planFileTasks('f1', presets, formats, completed, retryableOnly);
assert.equal(mixed.completed, 1);
assert.equal(mixed.skipped, 1);
assert.equal(mixed.tasks.length, 1);
assert.equal(mixed.completed + mixed.tasks.length + mixed.skipped, total);
});
});

417
src/lib/retry.ts
View file

@ -1,417 +0,0 @@
import { recordOp } from './diagnostics';
import { appendFailure, compactFailures, replayFailureMap } from './failures';
import { buildRegistry, isImageType, loadCustomPresets } from './presets';
import { readSettings } from './settings';
import { acquireLock, appendCompleted, completedKeyOf, readRun, releaseLock } from './state';
import type { FailureRecord, Preset } from './types';
import { errorMessage, warmVariant } from './warmer';
import { isInWindow } from './window';
export const MAX_ATTEMPTS = 3;
export const RETRY_BACKOFF_MS = [ 5 * 60 * 1000, 15 * 60 * 1000 ];
export const DEFAULT_RETRY_BATCH = 50;
export function nextRetryAt(attempt: number, now = Date.now()): number | null {
if (!Number.isFinite(attempt) || attempt >= MAX_ATTEMPTS) return null;
const index = Math.max(0, Math.min(Math.trunc(attempt) - 1, RETRY_BACKOFF_MS.length - 1));
return now + (RETRY_BACKOFF_MS[index] as number);
}
export function dueRetries(records: FailureRecord[], now = Date.now()): FailureRecord[] {
return records.filter(
(record) =>
record.status === 'failed' &&
record.attempts < MAX_ATTEMPTS &&
record.next_retry_at !== null &&
record.next_retry_at <= now,
);
}
export interface FailureInput {
file_id: string;
preset_key: string;
format: string | null;
error?: string;
ms?: number;
ts?: number;
}
interface JournalOptions {
records?: Map<string, FailureRecord>;
logger?: any;
}
export async function journalFailure(
env: Record<string, any>,
input: FailureInput,
options: JournalOptions = {},
): Promise<FailureRecord | null> {
try {
const records = options.records ?? (await replayFailureMap(env));
const key = completedKeyOf(input.file_id, input.preset_key, input.format);
const previous = records.get(key);
const ts = input.ts ?? Date.now();
const attempt =
previous && previous.status === 'failed' ? Math.min(MAX_ATTEMPTS, previous.attempts + 1) : 1;
const next = nextRetryAt(attempt, ts);
await appendFailure(
{
type: 'failed',
ts,
file_id: input.file_id,
preset_key: input.preset_key,
format: input.format,
attempt,
next_retry_at: next,
error: input.error,
ms: input.ms,
},
env,
{ logger: options.logger },
);
const record: FailureRecord = {
file_id: input.file_id,
preset_key: input.preset_key,
format: input.format,
attempts: attempt,
status: 'failed',
first_failed_at: previous && previous.status === 'failed' ? previous.first_failed_at : ts,
last_failed_at: ts,
next_retry_at: next,
error: input.error ?? null,
ms: input.ms ?? null,
};
records.set(key, record);
options.logger?.warn?.(
{
fileId: input.file_id,
preset_key: input.preset_key,
format: input.format,
attempt,
next_retry_at: next,
error: input.error,
},
'image-previews: preview failed',
);
return record;
} catch (error) {
options.logger?.error?.(error, 'image-previews: could not append failure journal');
return null;
}
}
export async function markRecovered(
env: Record<string, any>,
input: FailureInput,
options: JournalOptions = {},
): Promise<void> {
try {
const records = options.records ?? (await replayFailureMap(env));
const key = completedKeyOf(input.file_id, input.preset_key, input.format);
const previous = records.get(key);
if (!previous || previous.status === 'recovered') return;
const ts = input.ts ?? Date.now();
const attempt = Math.min(MAX_ATTEMPTS, previous.attempts + 1);
await appendFailure(
{
type: 'recovered',
ts,
file_id: input.file_id,
preset_key: input.preset_key,
format: input.format,
attempt,
next_retry_at: null,
ms: input.ms,
},
env,
{ logger: options.logger },
);
records.set(key, {
...previous,
attempts: attempt,
status: 'recovered',
next_retry_at: null,
ms: input.ms ?? previous.ms,
});
options.logger?.info?.(
{
fileId: input.file_id,
preset_key: input.preset_key,
format: input.format,
attempt,
ms: input.ms,
},
'image-previews: preview recovered',
);
} catch (error) {
options.logger?.error?.(error, 'image-previews: could not append recovery journal');
}
}
export interface RetryDeps {
services: Record<string, any>;
database: any;
getSchema: () => Promise<any>;
env: Record<string, any>;
logger: any;
}
export interface RetryTarget {
file_id: string;
preset_key: string;
format: string | null;
}
export interface RetryOptions {
now?: number;
ignoreWindow?: boolean;
limit?: number;
targets?: RetryTarget[];
}
export interface RetryOutcome {
status: 'completed' | 'skipped' | 'failed';
reason?: string;
processed: number;
recovered: number;
failed: number;
}
async function markNonRetryable(
env: Record<string, any>,
record: FailureRecord,
error: string,
records: Map<string, FailureRecord>,
logger: any,
): Promise<void> {
const ts = Date.now();
try {
await appendFailure(
{
type: 'retry',
ts,
file_id: record.file_id,
preset_key: record.preset_key,
format: record.format,
attempt: MAX_ATTEMPTS,
next_retry_at: null,
error,
},
env,
{ logger },
);
} catch (journalError) {
logger?.error?.(journalError, 'image-previews: could not append retry journal');
}
records.set(completedKeyOf(record.file_id, record.preset_key, record.format), {
...record,
attempts: MAX_ATTEMPTS,
status: 'failed',
next_retry_at: null,
error,
});
logger?.warn?.(
{
fileId: record.file_id,
preset_key: record.preset_key,
format: record.format,
attempt: MAX_ATTEMPTS,
next_retry_at: null,
error,
},
'image-previews: retry skipped',
);
}
export async function runDueRetries(deps: RetryDeps, options: RetryOptions = {}): Promise<RetryOutcome> {
const outcome: RetryOutcome = { status: 'skipped', processed: 0, recovered: 0, failed: 0 };
try {
const settings = await readSettings(deps.env);
if (!settings.enabled) return { ...outcome, reason: 'disabled' };
if (!options.ignoreWindow && !isInWindow(settings.windowStart, settings.windowEnd)) {
return { ...outcome, reason: 'outside-window' };
}
const run = await readRun(deps.env);
if (run?.running) return { ...outcome, reason: 'already-running' };
if (!(await acquireLock(deps.env))) return { ...outcome, reason: 'already-running' };
try {
const now = options.now ?? Date.now();
const schema = await deps.getSchema();
const custom = await loadCustomPresets(new deps.services['SettingsService']({ schema }));
const presets = buildRegistry(custom, null);
const presetByKey = new Map<string, Preset>(presets.map((preset) => [ preset.key, preset ]));
const records = await replayFailureMap(deps.env);
const targets = options.targets
? new Set(options.targets.map((target) => completedKeyOf(target.file_id, target.preset_key, target.format)))
: null;
const limit = Math.max(1, Math.min(200, Math.trunc(options.limit ?? DEFAULT_RETRY_BATCH)));
const candidates = dueRetries([ ...records.values() ], now)
.filter((record) => !targets || targets.has(completedKeyOf(record.file_id, record.preset_key, record.format)))
.sort(
(a, b) =>
(a.next_retry_at ?? 0) - (b.next_retry_at ?? 0) ||
a.last_failed_at - b.last_failed_at ||
completedKeyOf(a.file_id, a.preset_key, a.format).localeCompare(
completedKeyOf(b.file_id, b.preset_key, b.format),
),
)
.slice(0, limit);
const grouped = new Map<string, FailureRecord[]>();
for (const record of candidates) {
const group = grouped.get(record.file_id);
if (group) group.push(record);
else grouped.set(record.file_id, [ record ]);
}
const items = new deps.services['ItemsService']('directus_files', { schema });
const assets = new deps.services['AssetsService']({ schema });
for (const [ fileId, group ] of grouped) {
let file: { id?: string; type?: string | null } | null = null;
try {
file = await items.readOne(fileId, { fields: [ 'id', 'type' ] });
} catch {
file = null;
}
if (!file || !isImageType(file.type)) {
for (const record of group) {
await markNonRetryable(deps.env, record, 'file not found or not an image', records, deps.logger);
outcome.processed += 1;
outcome.failed += 1;
}
continue;
}
for (const record of group) {
const preset = presetByKey.get(record.preset_key);
if (!preset) {
await markNonRetryable(deps.env, record, `unknown preset: ${record.preset_key}`, records, deps.logger);
outcome.processed += 1;
outcome.failed += 1;
continue;
}
const variant = preset.format === 'auto' ? record.format : null;
if (preset.format === 'auto' && !variant) {
await markNonRetryable(deps.env, record, 'missing format', records, deps.logger);
outcome.processed += 1;
outcome.failed += 1;
continue;
}
const result = await warmVariant(assets, fileId, preset, variant);
const ts = Date.now();
recordOp({
ts,
fileId,
key: result.key,
format: result.format,
ok: result.ok,
ms: result.ms,
error: result.error,
});
if (result.ok) {
await appendCompleted(
{ file_id: fileId, preset_key: record.preset_key, format: result.format, ts },
deps.env,
);
await markRecovered(
deps.env,
{
file_id: fileId,
preset_key: record.preset_key,
format: result.format,
ms: result.ms,
ts,
},
{ records, logger: deps.logger },
);
outcome.recovered += 1;
} else {
const attempt = Math.min(MAX_ATTEMPTS, record.attempts + 1);
const next = nextRetryAt(attempt, ts);
try {
await appendFailure(
{
type: 'failed',
ts,
file_id: fileId,
preset_key: record.preset_key,
format: result.format,
attempt,
next_retry_at: next,
error: result.error,
ms: result.ms,
},
deps.env,
{ logger: deps.logger },
);
} catch (journalError) {
deps.logger?.error?.(journalError, 'image-previews: could not append retry journal');
}
records.set(completedKeyOf(fileId, record.preset_key, record.format), {
...record,
attempts: attempt,
status: 'failed',
last_failed_at: ts,
next_retry_at: next,
error: result.error ?? null,
ms: result.ms,
});
deps.logger?.warn?.(
{
fileId,
preset_key: record.preset_key,
format: result.format,
attempt,
next_retry_at: next,
error: result.error,
},
'image-previews: retry failed',
);
outcome.failed += 1;
}
outcome.processed += 1;
}
}
await compactFailures(deps.env);
return { ...outcome, status: 'completed' };
} finally {
await releaseLock(deps.env);
}
} catch (error) {
deps.logger?.error?.(error, 'image-previews: retry run failed');
return { ...outcome, status: 'failed', reason: errorMessage(error) };
}
}

63
src/lib/types.ts
View file

@ -35,69 +35,6 @@ export interface CompletedRecord {
ts: number; ts: number;
} }
export type FailureEventType = 'failed' | 'retry' | 'recovered';
export type FailureStatus = 'failed' | 'recovered' | 'retry';
export interface FailureEvent {
type: FailureEventType;
ts: number;
file_id: string;
preset_key: string;
format: string | null;
attempt: number;
next_retry_at?: number | null;
error?: string;
ms?: number;
}
export interface FailureRecord {
file_id: string;
preset_key: string;
format: string | null;
attempts: number;
status: FailureStatus;
first_failed_at: number;
last_failed_at: number;
next_retry_at: number | null;
error: string | null;
ms: number | null;
}
export interface FailureListQuery {
status?: FailureStatus | null;
limit: number;
offset: number;
}
export interface FailureListResult {
records: FailureRecord[];
total: number;
limit: number;
offset: number;
}
export interface FailureFileMeta {
filename_download?: string | null;
type?: string | null;
folder?: string | null;
}
export interface FailureEnrichedRecord extends FailureRecord {
file: {
filename_download: string | null;
type: string | null;
folder: string | null;
};
first_failed_at_iso: string | null;
first_failed_at_human: string;
last_failed_at_iso: string | null;
last_failed_at_human: string;
next_retry_at_iso: string | null;
next_retry_at_human: string | null;
ms_human: string | null;
}
export interface RunState { export interface RunState {
running: boolean; running: boolean;
startedAt: number | null; startedAt: number | null;

41
src/lib/warmer.ts
View file

@ -17,7 +17,7 @@ export interface WarmOptions {
onResult?: (result: WarmResultItem) => void; onResult?: (result: WarmResultItem) => void;
} }
export interface WarmTaskSpec { interface WarmTask {
preset: Preset; preset: Preset;
format: string | null; format: string | null;
} }
@ -43,41 +43,13 @@ function releaseStream(stream: unknown): void {
} }
} }
export async function warmVariant(
assets: AssetsLike,
fileId: string,
preset: Preset,
format: string | null,
options: { isCached?: (preset: Preset, format: string | null) => boolean | Promise<boolean> } = {},
): Promise<WarmResultItem> {
const transformation: Record<string, unknown> = { transformationParams: preset };
if (format) transformation['acceptFormat'] = format;
const started = Date.now();
const outputFormat = effectiveFormat(preset, format);
const cached = options.isCached ? await options.isCached(preset, format) : false;
if (cached) {
return { key: preset.key, format: outputFormat, ok: true, ms: Date.now() - started, cached: true };
}
try {
const asset = await assets.getAsset(fileId, transformation);
releaseStream(asset?.stream);
return { key: preset.key, format: outputFormat, ok: true, ms: Date.now() - started, cached: false };
} catch (error) {
return { key: preset.key, format: outputFormat, ok: false, ms: Date.now() - started, error: errorMessage(error) };
}
}
export async function warmFile( export async function warmFile(
assets: AssetsLike, assets: AssetsLike,
fileId: string, fileId: string,
presets: Preset[], presets: Preset[],
options: WarmOptions, options: WarmOptions,
): Promise<WarmSummary> { ): Promise<WarmSummary> {
const tasks: WarmTaskSpec[] = []; const tasks: WarmTask[] = [];
for (const preset of presets) { for (const preset of presets) {
for (const format of formatsForPreset(preset, options.formats)) { for (const format of formatsForPreset(preset, options.formats)) {
@ -85,15 +57,6 @@ export async function warmFile(
} }
} }
return warmTasks(assets, fileId, tasks, options);
}
export async function warmTasks(
assets: AssetsLike,
fileId: string,
tasks: WarmTaskSpec[],
options: WarmOptions,
): Promise<WarmSummary> {
const results: WarmResultItem[] = new Array(tasks.length); const results: WarmResultItem[] = new Array(tasks.length);
let cursor = 0; let cursor = 0;

58
src/module/module.vue
View file

@ -195,34 +195,6 @@
</table> </table>
</section> </section>
<section class="ip__card">
<header class="ip__head">
<h2>Failed previews</h2>
<v-button @click="loadFailures">Refresh</v-button>
</header>
<p class="ip__hint">
Persisted failure journal. Retries run automatically inside the maintenance window with no
active backfill: up to 3 attempts, then 5 and 15 minute backoff.
</p>
<table class="ip__table">
<thead>
<tr><th>file</th><th>preset</th><th>format</th><th>attempts</th><th>last error</th><th>next retry</th><th>status</th></tr>
</thead>
<tbody>
<tr v-for="failure in failures" :key="failure.file_id + failure.preset_key + (failure.format ?? '')">
<td class="ip__mono">{{ failure.file?.filename_download ?? failure.file_id.slice(0, 8) }}</td>
<td>{{ failure.preset_key }}</td>
<td>{{ failure.format ?? 'original' }}</td>
<td>{{ failure.attempts }}</td>
<td class="ip__error">{{ failure.error ?? '' }}</td>
<td>{{ failure.next_retry_at_human ?? '—' }}</td>
<td>{{ failure.status }}</td>
</tr>
</tbody>
</table>
<p v-if="failures.length === 0" class="ip__hint">No failures recorded.</p>
</section>
<section class="ip__card"> <section class="ip__card">
<header class="ip__head"><h2>API reference</h2></header> <header class="ip__head"><h2>API reference</h2></header>
<p class="ip__hint"> <p class="ip__hint">
@ -278,23 +250,11 @@ interface PresetInfo {
enabled: boolean; enabled: boolean;
} }
interface FailureRecord {
file_id: string;
preset_key: string;
format: string | null;
attempts: number;
status: string;
error: string | null;
next_retry_at_human: string | null;
file: { filename_download: string | null; type: string | null; folder: string | null } | null;
}
const base = '/image-previews'; const base = '/image-previews';
const status = ref<Status | null>(null); const status = ref<Status | null>(null);
const config = ref<Record<string, any> | null>(null); const config = ref<Record<string, any> | null>(null);
const disk = ref<{ previews: Bucket; originals: Bucket; total: Bucket } | null>(null); const disk = ref<{ previews: Bucket; originals: Bucket; total: Bucket } | null>(null);
const ops = ref<any[]>([]); const ops = ref<any[]>([]);
const failures = ref<FailureRecord[]>([]);
const presets = ref<PresetInfo[]>([]); const presets = ref<PresetInfo[]>([]);
const shadowedKeys = ref<string[]>([]); const shadowedKeys = ref<string[]>([]);
const fileId = ref(''); const fileId = ref('');
@ -391,13 +351,6 @@ const endpoints = [
description: 'In-memory ring buffer of the last 200 generation attempts with timing and error messages.', description: 'In-memory ring buffer of the last 200 generation attempts with timing and error messages.',
example: 'curl -H "Authorization: Bearer $TOKEN" "$BASE/image-previews/ops?limit=50"', example: 'curl -H "Authorization: Bearer $TOKEN" "$BASE/image-previews/ops?limit=50"',
}, },
{
method: 'GET',
path: '/failures',
summary: 'persisted failures',
description: 'Failure journal records with file metadata, attempts, timestamps, next retry and status. Filter with status=failed|recovered|retry, limit (max 200) and offset.',
example: 'curl -H "Authorization: Bearer $TOKEN" "$BASE/image-previews/failures?status=failed&limit=50&offset=0"',
},
]; ];
async function api(path: string, init?: RequestInit) { async function api(path: string, init?: RequestInit) {
@ -436,15 +389,6 @@ async function loadPresets() {
shadowedKeys.value = result.shadowedKeys ?? []; shadowedKeys.value = result.shadowedKeys ?? [];
} }
async function loadFailures() {
try {
const result = await api('/failures?limit=100');
failures.value = result.records ?? [];
} catch {
/* ignore polling errors */
}
}
async function saveConfig() { async function saveConfig() {
if (!config.value) return; if (!config.value) return;
@ -512,7 +456,7 @@ async function generate(dryRun: boolean) {
} }
onMounted(async () => { onMounted(async () => {
await Promise.all([ loadStatus(), loadConfig(), loadDisk(), loadPresets(), loadFailures() ]); await Promise.all([ loadStatus(), loadConfig(), loadDisk(), loadPresets() ]);
timer = setInterval(loadStatus, 3000); timer = setInterval(loadStatus, 3000);
}); });