diff --git a/.gitignore b/.gitignore index 1b870ed..21df11a 100644 --- a/.gitignore +++ b/.gitignore @@ -3,4 +3,3 @@ node_modules dist dev/extension dev/.smoke.env -.opencode/tmp/ diff --git a/.opencode/feature_list.json b/.opencode/feature_list.json index 81b4930..541f494 100644 --- a/.opencode/feature_list.json +++ b/.opencode/feature_list.json @@ -201,54 +201,6 @@ "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." ] - }, - { - "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 `/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." - ] } ] } diff --git a/.opencode/goal.md b/.opencode/goal.md index 3ecf86a..bcc22dc 100644 --- a/.opencode/goal.md +++ b/.opencode/goal.md @@ -3,7 +3,6 @@ Status: SCAFFOLDED — opened 2026-10-25. Standalone extension project (independent of any host application). 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). 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. - 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) - **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 backfill.ts enumerate directus_files (image/*), window-aware, resumable 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 disk.ts sum sizes of __ files (driver.list + stat) window.ts maintenance-window logic (is-now-in-window, next-window) diff --git a/.opencode/progress.txt b/.opencode/progress.txt index 27dc46e..ba18bbe 100644 --- a/.opencode/progress.txt +++ b/.opencode/progress.txt @@ -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. - UI now shows last-run errors and `paused (window closed)`. - 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 `/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 `/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//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.)` 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..`, 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. diff --git a/.opencode/skills/directus-image-previews/SKILL.md b/.opencode/skills/directus-image-previews/SKILL.md index 0bc00f6..1e804d8 100644 --- a/.opencode/skills/directus-image-previews/SKILL.md +++ b/.opencode/skills/directus-image-previews/SKILL.md @@ -65,7 +65,6 @@ cache files: settings.json effective config run.json progress (running/done/total/generated/hits/failed/rate/eta/paused/cancelled/error) 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 cancel.flag cancellation request ``` diff --git a/AGENTS.md b/AGENTS.md index 56815ec..2486bdb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -94,8 +94,6 @@ pacing). (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. - `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). ## Nightly scheduling diff --git a/README.md b/README.md index b0c1b5a..0ab3c68 100644 --- a/README.md +++ b/README.md @@ -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?}`) - `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/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. @@ -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; - 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); -- 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). +- once all tasks are done, it stays idle until the next day. `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 `/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 Never test against a production instance. diff --git a/package-lock.json b/package-lock.json index 688bbc8..72ba10a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "directus-extension-generate-image-previews", - "version": "0.2.0", + "version": "0.1.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "directus-extension-generate-image-previews", - "version": "0.2.0", + "version": "0.1.0", "dependencies": { "object-hash": "^3.0.0" }, diff --git a/package.json b/package.json index ced424d..bd73575 100644 --- a/package.json +++ b/package.json @@ -15,7 +15,7 @@ "url": "https://code.at.locus.im/Rawman/directus-extension-generate-image-previews/issues" }, "icon": "extension", - "version": "0.2.0", + "version": "0.1.0", "keywords": [ "directus", "directus-extension", diff --git a/src/endpoint/index.ts b/src/endpoint/index.ts index c2dab8b..7a5e117 100644 --- a/src/endpoint/index.ts +++ b/src/endpoint/index.ts @@ -2,15 +2,12 @@ import { defineEndpoint } from '@directus/extensions-sdk'; import type { Request, Response, NextFunction } from 'express'; import { runBackfill, requestCancel } from '../lib/backfill'; import { computePlan, getOps, recordOp } from '../lib/diagnostics'; -import { enrichFailures, listFailures, parseFailureQuery, replayFailureMap } from '../lib/failures'; import { humanBytes, humanDuration, humanMs, toIso } from '../lib/format'; import { buildRegistry, effectiveFormatsForPreset, formatsForPreset, loadCustomPresets } from '../lib/presets'; -import { journalFailure, markRecovered } from '../lib/retry'; import { mergeSettings, readSettings, writeSettings } from '../lib/settings'; import { isLocked, readRun } from '../lib/state'; import { previewExists, storageUsage } from '../lib/storage'; import { SYSTEM_PRESETS } from '../lib/system-presets'; -import type { FailureFileMeta, FailureListQuery } from '../lib/types'; import { isInWindow, nextWindow } from '../lib/window'; 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(); - - 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( '/generate', 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({ fileId, ms: Date.now() - started, diff --git a/src/hook/index.ts b/src/hook/index.ts index f92ce67..021ac3c 100644 --- a/src/hook/index.ts +++ b/src/hook/index.ts @@ -1,9 +1,7 @@ import { defineHook } from '@directus/extensions-sdk'; import { runBackfill } from '../lib/backfill'; import { recordOp } from '../lib/diagnostics'; -import { replayFailureMap } from '../lib/failures'; import { buildRegistry, isImageType, loadCustomPresets } from '../lib/presets'; -import { journalFailure, markRecovered, runDueRetries } from '../lib/retry'; import { readSettings } from '../lib/settings'; import { readRun } from '../lib/state'; 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'); } @@ -104,12 +73,6 @@ export default defineHook(({ action, schedule }, { services, database, getSchema 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 ? new Date(run.finishedAt).toDateString() === new Date().toDateString() : false; diff --git a/src/lib/backfill.ts b/src/lib/backfill.ts index e400605..cd32525 100644 --- a/src/lib/backfill.ts +++ b/src/lib/backfill.ts @@ -1,14 +1,5 @@ import { computePlan, recordOp } from './diagnostics'; -import { replayFailureMap } from './failures'; -import { - buildRegistry, - effectiveFormat, - formatsForPreset, - isImageType, - loadCustomPresets, - tasksPerFile, -} from './presets'; -import { journalFailure, markRecovered } from './retry'; +import { buildRegistry, effectiveFormatsForPreset, isImageType, loadCustomPresets, tasksPerFile } from './presets'; import { readSettings } from './settings'; import { acquireLock, @@ -22,8 +13,8 @@ import { requestCancel, writeRun, } from './state'; -import type { FailureRecord, Preset, RunState } from './types'; -import { errorMessage, warmTasks, type WarmTaskSpec } from './warmer'; +import type { Preset, RunState } from './types'; +import { errorMessage, warmFile } from './warmer'; import { isInWindow } from './window'; import { listFileNames } from './storage'; @@ -48,45 +39,6 @@ export interface BackfillOutcome { const PAGE_SIZE = 100; -export interface FileTaskPlan { - tasks: WarmTaskSpec[]; - completed: number; - skipped: number; -} - -export function planFileTasks( - fileId: string, - presets: Preset[], - formats: string[], - completed: Set, - failures: Map, -): 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; async function countImageFiles(database: any): Promise { @@ -143,7 +95,6 @@ export async function runBackfill(deps: BackfillDeps, options: BackfillOptions = await writeRun(run, deps.env); const completed = await loadCompleted(deps.env); - const failureRecords = await replayFailureMap(deps.env); let existingFiles = new Set(); try { @@ -181,20 +132,24 @@ export async function runBackfill(deps: BackfillDeps, options: BackfillOptions = if (!isImageType(file.type)) continue; - const planned = planFileTasks(file.id, presets, formats, completed, failureRecords); - const tasks = planned.tasks; + const pending = presets.filter((preset) => + effectiveFormatsForPreset(preset, formats).some( + (format) => !completed.has(completedKeyOf(file.id, preset.key, format)), + ), + ); - if (planned.completed > 0) { - run.hits += planned.completed; - run.done += planned.completed; + const pendingTasks = pending.reduce( + (total, preset) => total + effectiveFormatsForPreset(preset, formats).length, + 0, + ); + const alreadyDone = perFile - pendingTasks; + + if (alreadyDone > 0) { + run.hits += alreadyDone; + run.done += alreadyDone; } - if (planned.skipped > 0) { - run.failed += planned.skipped; - run.done += planned.skipped; - } - - if (tasks.length === 0) { + if (pending.length === 0) { await writeRun(run, deps.env); 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, formats, 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() }, 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; else run.generated += 1; } else { 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; diff --git a/src/lib/failures.test.ts b/src/lib/failures.test.ts deleted file mode 100644 index 37f4e4f..0000000 --- a/src/lib/failures.test.ts +++ /dev/null @@ -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) => Promise): Promise { - 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 { - 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((resolve, reject) => { - child.once('exit', () => resolve()); - child.once('error', reject); - }); - - return pid; -} - -async function sleep(ms: number): Promise { - await new Promise((resolve) => setTimeout(resolve, ms)); -} - -function event(overrides: Partial & Pick): 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, token: string): Promise { - 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, owner: Record = {}): Promise { - 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> = []; - - try { - const result = await appendFailure( - event({ type: 'failed', ts: 2000, file_id: 'f2' }), - env, - { - timeoutMs: 30, - waitMs: 5, - logger: { warn: (payload: Record) => 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((resolve) => { - observedResolve = resolve; - }); - let resumeResolve: () => void = () => {}; - const resume = new Promise((resolve) => { - resumeResolve = resolve; - }); - let rejectedResolve: () => void = () => {}; - const rejected = new Promise((resolve) => { - rejectedResolve = resolve; - }); - let releaseResolve: () => void = () => {}; - const release = new Promise((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'); -}); diff --git a/src/lib/failures.ts b/src/lib/failures.ts deleted file mode 100644 index 4eae9af..0000000 --- a/src/lib/failures.ts +++ /dev/null @@ -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; - onGuardRejected?: () => Promise; - logger?: any; - readBootId?: () => Promise; - readStartTime?: (pid: number) => Promise; -} - -export interface FailureLock { - acquired: boolean; - token: string; - release: () => Promise; -} - -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 { - await new Promise((resolve) => setTimeout(resolve, ms)); -} - -async function fileMtime(path: string): Promise { - try { - return (await stat(path)).mtimeMs; - } catch { - return null; - } -} - -async function defaultReadBootId(): Promise { - 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 { - 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 { - try { - const parsed = JSON.parse(await readFile(path, 'utf8')) as Record; - 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 { - try { - await unlink(path); - return true; - } catch { - return false; - } -} - -async function releaseFailureLock(env: EnvLike, token: string): Promise { - 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 { - 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, -): Promise { - 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 { - 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 { - 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 { - 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; - 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 { - 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(); - - 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 { - return replayFailureEvents(await readFailureEvents(env)); -} - -export async function replayFailureMap(env: EnvLike): Promise> { - 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 { - 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, -): 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 { - 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(); - } -} diff --git a/src/lib/retry.test.ts b/src/lib/retry.test.ts deleted file mode 100644 index 9f2c2eb..0000000 --- a/src/lib/retry.test.ts +++ /dev/null @@ -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) => Promise): Promise { - 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 }>; - readOneCalls: number; - warnings: Array>; - infos: Array>; -} - -function makeDeps( - env: Record, - files: Record, - options: { fail?: boolean } = {}, -): DepsHarness { - const getAssetCalls: Array<{ id: string; transformation: Record }> = []; - const warnings: Array> = []; - const infos: Array> = []; - const harness = { - getAssetCalls, - readOneCalls: 0, - warnings, - infos, - } as DepsHarness; - - class SettingsService { - constructor(_options: Record) {} - - async readSingleton(): Promise> { - return { storage_asset_presets: [] }; - } - } - - class ItemsService { - constructor(_options: Record) {} - - 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) {} - - async getAsset(id: string, transformation: Record): 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) => warnings.push(payload), - info: (payload: Record) => infos.push(payload), - error: () => {}, - }, - }; - - return harness; -} - -function failedRecord(overrides: Partial & Pick): 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([ - [ 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([ - [ 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([ - [ - 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([ - [ - 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); - }); -}); diff --git a/src/lib/retry.ts b/src/lib/retry.ts deleted file mode 100644 index 863d2b2..0000000 --- a/src/lib/retry.ts +++ /dev/null @@ -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; - logger?: any; -} - -export async function journalFailure( - env: Record, - input: FailureInput, - options: JournalOptions = {}, -): Promise { - 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, - input: FailureInput, - options: JournalOptions = {}, -): Promise { - 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; - database: any; - getSchema: () => Promise; - env: Record; - 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, - record: FailureRecord, - error: string, - records: Map, - logger: any, -): Promise { - 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 { - 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(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(); - - 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) }; - } -} diff --git a/src/lib/types.ts b/src/lib/types.ts index d6f5355..c5c232d 100644 --- a/src/lib/types.ts +++ b/src/lib/types.ts @@ -35,69 +35,6 @@ export interface CompletedRecord { 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 { running: boolean; startedAt: number | null; diff --git a/src/lib/warmer.ts b/src/lib/warmer.ts index 5a56853..6ea8fae 100644 --- a/src/lib/warmer.ts +++ b/src/lib/warmer.ts @@ -17,7 +17,7 @@ export interface WarmOptions { onResult?: (result: WarmResultItem) => void; } -export interface WarmTaskSpec { +interface WarmTask { preset: Preset; 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 } = {}, -): Promise { - const transformation: Record = { 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( assets: AssetsLike, fileId: string, presets: Preset[], options: WarmOptions, ): Promise { - const tasks: WarmTaskSpec[] = []; + const tasks: WarmTask[] = []; for (const preset of presets) { 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 { const results: WarmResultItem[] = new Array(tasks.length); let cursor = 0; diff --git a/src/module/module.vue b/src/module/module.vue index b700028..f400a6e 100644 --- a/src/module/module.vue +++ b/src/module/module.vue @@ -195,34 +195,6 @@ -
-
-

Failed previews

- Refresh -
-

- Persisted failure journal. Retries run automatically inside the maintenance window with no - active backfill: up to 3 attempts, then 5 and 15 minute backoff. -

- - - - - - - - - - - - - - - -
filepresetformatattemptslast errornext retrystatus
{{ failure.file?.filename_download ?? failure.file_id.slice(0, 8) }}{{ failure.preset_key }}{{ failure.format ?? 'original' }}{{ failure.attempts }}{{ failure.error ?? '' }}{{ failure.next_retry_at_human ?? '—' }}{{ failure.status }}
-

No failures recorded.

-
-

API reference

@@ -278,23 +250,11 @@ interface PresetInfo { 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 status = ref(null); const config = ref | null>(null); const disk = ref<{ previews: Bucket; originals: Bucket; total: Bucket } | null>(null); const ops = ref([]); -const failures = ref([]); const presets = ref([]); const shadowedKeys = 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.', 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) { @@ -436,15 +389,6 @@ async function loadPresets() { 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() { if (!config.value) return; @@ -512,7 +456,7 @@ async function generate(dryRun: boolean) { } onMounted(async () => { - await Promise.all([ loadStatus(), loadConfig(), loadDisk(), loadPresets(), loadFailures() ]); + await Promise.all([ loadStatus(), loadConfig(), loadDisk(), loadPresets() ]); timer = setInterval(loadStatus, 3000); });