From ec3e0c16e07495079b0b63ddc6b8cab3c604ad9e Mon Sep 17 00:00:00 2001 From: Rawman Grey Date: Fri, 30 Oct 2026 10:22:27 +0300 Subject: [PATCH 1/3] docs: record F12 uploaded_on verification (host fix) --- .opencode/feature_list.json | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/.opencode/feature_list.json b/.opencode/feature_list.json index 5c3b508..321511d 100644 --- a/.opencode/feature_list.json +++ b/.opencode/feature_list.json @@ -238,12 +238,14 @@ "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": false, + "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. `now() at time zone 'utc'` matches Directus's UTC writes for the tz-less timestamp column.", + "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." ] } From 871182d6d5ced6ea6fbfcc3c29d5530898010541 Mon Sep 17 00:00:00 2001 From: Rawman Grey Date: Fri, 30 Oct 2026 10:22:29 +0300 Subject: [PATCH 2/3] feat: persist failure journal and bounded retries (F11) --- .gitignore | 1 + .opencode/progress.txt | 159 +++++ .../skills/directus-image-previews/SKILL.md | 1 + AGENTS.md | 2 + README.md | 18 +- src/endpoint/index.ts | 70 ++ src/hook/index.ts | 37 + src/lib/backfill.ts | 104 ++- src/lib/failures.test.ts | 649 ++++++++++++++++++ src/lib/failures.ts | 584 ++++++++++++++++ src/lib/retry.test.ts | 588 ++++++++++++++++ src/lib/retry.ts | 417 +++++++++++ src/lib/types.ts | 63 ++ src/lib/warmer.ts | 41 +- src/module/module.vue | 58 +- 15 files changed, 2769 insertions(+), 23 deletions(-) create mode 100644 src/lib/failures.test.ts create mode 100644 src/lib/failures.ts create mode 100644 src/lib/retry.test.ts create mode 100644 src/lib/retry.ts diff --git a/.gitignore b/.gitignore index 21df11a..1b870ed 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ node_modules dist dev/extension dev/.smoke.env +.opencode/tmp/ diff --git a/.opencode/progress.txt b/.opencode/progress.txt index ba18bbe..3f70b97 100644 --- a/.opencode/progress.txt +++ b/.opencode/progress.txt @@ -146,3 +146,162 @@ 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. diff --git a/.opencode/skills/directus-image-previews/SKILL.md b/.opencode/skills/directus-image-previews/SKILL.md index 1e804d8..0bc00f6 100644 --- a/.opencode/skills/directus-image-previews/SKILL.md +++ b/.opencode/skills/directus-image-previews/SKILL.md @@ -65,6 +65,7 @@ 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 2486bdb..56815ec 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -94,6 +94,8 @@ 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 0ab3c68..b0c1b5a 100644 --- a/README.md +++ b/README.md @@ -105,6 +105,8 @@ 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. @@ -115,10 +117,24 @@ 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. +- once all tasks are done, it stays idle until the next day; +- before starting the backfill it runs any due failure retries (targeted pairs only, bounded batch). `Run now` in the admin module ignores the window. +## 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/src/endpoint/index.ts b/src/endpoint/index.ts index 7a5e117..c2dab8b 100644 --- a/src/endpoint/index.ts +++ b/src/endpoint/index.ts @@ -2,12 +2,15 @@ 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'; @@ -195,6 +198,44 @@ 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) => { @@ -253,6 +294,35 @@ 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 021ac3c..f92ce67 100644 --- a/src/hook/index.ts +++ b/src/hook/index.ts @@ -1,7 +1,9 @@ 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'; @@ -60,6 +62,35 @@ 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'); } @@ -73,6 +104,12 @@ 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 cd32525..e400605 100644 --- a/src/lib/backfill.ts +++ b/src/lib/backfill.ts @@ -1,5 +1,14 @@ import { computePlan, recordOp } from './diagnostics'; -import { buildRegistry, effectiveFormatsForPreset, isImageType, loadCustomPresets, tasksPerFile } from './presets'; +import { replayFailureMap } from './failures'; +import { + buildRegistry, + effectiveFormat, + formatsForPreset, + isImageType, + loadCustomPresets, + tasksPerFile, +} from './presets'; +import { journalFailure, markRecovered } from './retry'; import { readSettings } from './settings'; import { acquireLock, @@ -13,8 +22,8 @@ import { requestCancel, writeRun, } from './state'; -import type { Preset, RunState } from './types'; -import { errorMessage, warmFile } from './warmer'; +import type { FailureRecord, Preset, RunState } from './types'; +import { errorMessage, warmTasks, type WarmTaskSpec } from './warmer'; import { isInWindow } from './window'; import { listFileNames } from './storage'; @@ -39,6 +48,45 @@ 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 { @@ -95,6 +143,7 @@ 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 { @@ -132,24 +181,20 @@ export async function runBackfill(deps: BackfillDeps, options: BackfillOptions = if (!isImageType(file.type)) continue; - const pending = presets.filter((preset) => - effectiveFormatsForPreset(preset, formats).some( - (format) => !completed.has(completedKeyOf(file.id, preset.key, format)), - ), - ); + const planned = planFileTasks(file.id, presets, formats, completed, failureRecords); + const tasks = planned.tasks; - 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.completed > 0) { + run.hits += planned.completed; + run.done += planned.completed; } - if (pending.length === 0) { + if (planned.skipped > 0) { + run.failed += planned.skipped; + run.done += planned.skipped; + } + + if (tasks.length === 0) { await writeRun(run, deps.env); continue; } @@ -173,7 +218,7 @@ export async function runBackfill(deps: BackfillDeps, options: BackfillOptions = } }; - const summary = await warmFile(assets, file.id, pending, { + const summary = await warmTasks(assets, file.id, tasks, { concurrency: settings.concurrency, formats, isCached, @@ -199,11 +244,32 @@ 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 new file mode 100644 index 0000000..37f4e4f --- /dev/null +++ b/src/lib/failures.test.ts @@ -0,0 +1,649 @@ +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 new file mode 100644 index 0000000..4eae9af --- /dev/null +++ b/src/lib/failures.ts @@ -0,0 +1,584 @@ +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 new file mode 100644 index 0000000..9f2c2eb --- /dev/null +++ b/src/lib/retry.test.ts @@ -0,0 +1,588 @@ +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 new file mode 100644 index 0000000..863d2b2 --- /dev/null +++ b/src/lib/retry.ts @@ -0,0 +1,417 @@ +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 c5c232d..d6f5355 100644 --- a/src/lib/types.ts +++ b/src/lib/types.ts @@ -35,6 +35,69 @@ 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 6ea8fae..5a56853 100644 --- a/src/lib/warmer.ts +++ b/src/lib/warmer.ts @@ -17,7 +17,7 @@ export interface WarmOptions { onResult?: (result: WarmResultItem) => void; } -interface WarmTask { +export interface WarmTaskSpec { preset: Preset; format: string | null; } @@ -43,13 +43,41 @@ 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: WarmTask[] = []; + const tasks: WarmTaskSpec[] = []; for (const preset of presets) { for (const format of formatsForPreset(preset, options.formats)) { @@ -57,6 +85,15 @@ 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 f400a6e..b700028 100644 --- a/src/module/module.vue +++ b/src/module/module.vue @@ -195,6 +195,34 @@ +
+
+

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

@@ -250,11 +278,23 @@ 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(''); @@ -351,6 +391,13 @@ 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) { @@ -389,6 +436,15 @@ 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; @@ -456,7 +512,7 @@ async function generate(dryRun: boolean) { } onMounted(async () => { - await Promise.all([ loadStatus(), loadConfig(), loadDisk(), loadPresets() ]); + await Promise.all([ loadStatus(), loadConfig(), loadDisk(), loadPresets(), loadFailures() ]); timer = setInterval(loadStatus, 3000); }); From 3d9ed8de516afa98aa8927acadf337ffda9d5ee7 Mon Sep 17 00:00:00 2001 From: Rawman Grey Date: Fri, 30 Oct 2026 10:22:34 +0300 Subject: [PATCH 3/3] docs: mark F11 accepted and log manager decision --- .opencode/feature_list.json | 5 +++-- .opencode/progress.txt | 7 +++++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/.opencode/feature_list.json b/.opencode/feature_list.json index 321511d..81b4930 100644 --- a/.opencode/feature_list.json +++ b/.opencode/feature_list.json @@ -216,7 +216,7 @@ "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": false, + "passes": true, "blocked_by": [ "F08" ], @@ -224,7 +224,8 @@ "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." + "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." ] }, { diff --git a/.opencode/progress.txt b/.opencode/progress.txt index 3f70b97..27dc46e 100644 --- a/.opencode/progress.txt +++ b/.opencode/progress.txt @@ -305,3 +305,10 @@ Not verified locally: - 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.