feat: persist failure journal and bounded retries (F11)
This commit is contained in:
parent
ec3e0c16e0
commit
871182d6d5
15 changed files with 2769 additions and 23 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -3,3 +3,4 @@ node_modules
|
|||
dist
|
||||
dev/extension
|
||||
dev/.smoke.env
|
||||
.opencode/tmp/
|
||||
|
|
|
|||
|
|
@ -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 `<state-dir>/failures.jsonl`, replay of the latest record per
|
||||
`(file_id,preset_key,format)` (key = `completedKeyOf`), corrupt-line tolerance, atomic compaction
|
||||
(unresolved + last 1,000 recovered), and `listFailures` with status filter + limit/offset.
|
||||
- Added `src/lib/retry.ts`: MAX_ATTEMPTS=3, 5m/15m backoff, `nextRetryAt`, `dueRetries`,
|
||||
`journalFailure`/`markRecovered`, and `runDueRetries` (enabled/window/lock gated, targeted bounded
|
||||
batch default 50, grouped by file, full preset + acceptFormat via `warmVariant`, completion + recovered
|
||||
on success, attempt+1 + next retry on failure, compaction under `run.lock`, never throws).
|
||||
- Hooked failure journaling into the upload warm hook, `POST /generate`, and the backfill; successes mark
|
||||
prior failures recovered. Journal writes never alter responses or hook behavior.
|
||||
- Hook scheduler tick now runs due retries first, then the existing backfill.
|
||||
- Added admin-only `GET /image-previews/failures?status=&limit=&offset=` with one batched
|
||||
`directus_files` metadata query (`filename_download`, `type`, `folder`) and ISO/human timestamps.
|
||||
- Module: “Failed previews” table + Refresh (no toasts); API reference includes `/failures`; polling
|
||||
interval unchanged.
|
||||
- Types added; README/AGENTS/project SKILL state-file docs updated.
|
||||
- Tests: `failures.test.ts` (append/replay/latest-state/corrupt/idempotent/compaction/list) and
|
||||
`retry.test.ts` (backoff/exhaustion/due selection/bounded batch/grouping/missing preset/recovered/
|
||||
failure/disabled).
|
||||
|
||||
Files changed:
|
||||
- src/lib/failures.ts (new), src/lib/failures.test.ts (new)
|
||||
- src/lib/retry.ts (new), src/lib/retry.test.ts (new)
|
||||
- src/lib/types.ts, src/lib/warmer.ts, src/lib/backfill.ts
|
||||
- src/hook/index.ts, src/endpoint/index.ts, src/module/module.vue
|
||||
- README.md, AGENTS.md, .opencode/skills/directus-image-previews/SKILL.md, .gitignore
|
||||
- .opencode/progress.txt
|
||||
|
||||
Commands:
|
||||
- `npx tsc --noEmit` → exit 0, 0 errors (`.opencode/tmp/runs/f11-tsc.log`)
|
||||
- `npm test` → exit 0, 55/55 pass (`.opencode/tmp/runs/f11-test.log`)
|
||||
- `npm run build` → exit 0, `dist/app.js` + `dist/api.js` emitted (`.opencode/tmp/runs/f11-build.log`)
|
||||
|
||||
Manual verification steps:
|
||||
1. Cause a failure (unsupported preset format or missing file) from upload, `POST /generate`, and backfill;
|
||||
confirm one line per failure in `<state-dir>/failures.jsonl` and a warn log with `fileId`,
|
||||
`preset_key`, `format`, `attempt`, `next_retry_at`, `error`.
|
||||
2. Inside the maintenance window with no backfill active, wait for the next 15-minute tick; confirm a due
|
||||
pair is retried, success writes `state.jsonl` + a `recovered` event and logs info, failure writes
|
||||
`attempt+1` with 15m backoff (or exhausted at 3).
|
||||
3. `GET /image-previews/failures?status=failed&limit=50&offset=0` as admin: records include file metadata,
|
||||
attempts, first/last failure, next retry raw+ISO/human, status, error, ms; non-admin gets 403.
|
||||
4. Module “Failed previews” table renders and Refresh reloads; no toast/notification appears.
|
||||
|
||||
Not verified locally:
|
||||
- Browser rendering of the new module table (no live Directus/admin session in this run).
|
||||
- Live PM2 `schedule('*/15 * * * *')` tick and maintenance-window retry execution.
|
||||
- Concurrent PM2 append behavior and compaction race with upload-hook appends.
|
||||
|
||||
## 2026-10-30 — F11 review fixes
|
||||
|
||||
- Head: `feat/F11` @ `e10819c`.
|
||||
- Replay now derives attempts from `event.attempt`; fail→recover→fail starts a fresh cycle (attempt 1) and duplicate failed events are idempotent.
|
||||
- Journal writer lock (`failures.lock`, 20×10 ms, 30 s stale): appends serialize, compaction returns `false` if it cannot lock, and append still falls back to writing.
|
||||
- Backfill loads the journal once, skips unresolved failures per pair, and caps journal attempts at `MAX_ATTEMPTS`; recovered pairs still append `recovered`.
|
||||
- Added `/failures` shape/validation tests, retry lock/window tests, fail→recover→fail, attempt cap, and 50-append compaction race tests.
|
||||
|
||||
## 2026-10-30 — F11 second review fixes
|
||||
|
||||
- Implementation head: `feat/F11` @ `7f46493` (progress commit follows).
|
||||
- Journal writes now require the `failures.lock` directory (`owner.json` with random token + pid) and never append unlocked. Blocking acquisition defaults to 10 s / 10 ms poll; timeout drops the event with `image-previews: could not journal failure` and `fileId`/`preset_key`/`format`.
|
||||
- Stale locks are broken only when the owner pid is dead (`process.kill(pid, 0)` → `ESRCH`); mtime staleness removed. Release only removes the lock when `owner.json.token` still matches.
|
||||
- `compactFailures` uses the same lock across read→write temp→rename, returns `false` when it cannot acquire, and uses a pid+token-unique temp filename.
|
||||
- Tests now cover live-owner append refusal, dead-pid recovery, foreign-token release, live-lock compaction refusal, temporarily-held-lock 50-append race, and `planFileTasks` total/resume/skip counting.
|
||||
- `npx tsc --noEmit` 0 errors; `npm test` 69/69 pass; `npm run build` emits both bundles. Logs: `.opencode/tmp/runs/f11-fix2-{tsc,test,build}.log`.
|
||||
- Remaining manual checks: live PM2 scheduler retry tick, browser module rendering, and a real multi-worker compaction/append run.
|
||||
|
||||
## 2026-10-30 — F11 final lock hardening
|
||||
|
||||
- Implementation head: `feat/F11` @ `bc20c6a` (progress commit follows).
|
||||
- Lock is now an atomic `failures.lock` file written with `flag:'wx'`; owner carries `token`, `pid`, `bootId`, `startTime`, `ts`. Identity readers use `/proc/sys/kernel/random/boot_id` and `/proc/<pid>/stat` field 22, with injectable readers for tests.
|
||||
- Dead-lock reclaim runs only under `failures.lock.reclaim` (mkdir guard). Inside the guard it re-stats/re-reads the current owner and unlinks only a dead owner or an unreadable owner older than the 2 s grace. Live owners are never broken; stale guards older than 30 s are removed.
|
||||
- Release unlinks only when the current `owner.token` matches; no unlink on mismatch or read error.
|
||||
- Tests added: concurrent reclaim of a dead lock, missing/corrupt owner grace, PID reuse by bootId and startTime, guard-held reclaim and stale-guard recovery. Existing live-owner timeout, foreign-token release, 50-append race, and compaction-blocked tests stay green.
|
||||
- `npx tsc --noEmit` 0 errors; `npm test` 76/76 pass; `npm run build` emits both bundles. Logs: `.opencode/tmp/runs/f11-lock-{test,build}.log` and `f11-lock-final-*`.
|
||||
- Remaining manual checks: live PM2 multi-worker append/compaction, live scheduler retry tick, browser module rendering.
|
||||
|
||||
## 2026-10-30 — F11 reclaim-guard lifecycle fix
|
||||
|
||||
- Implementation head: `feat/F11` @ `cd973ca` (progress commit follows).
|
||||
- Stale guard removal is now atomic: `rename(guardPath, guardPath.stale.<staleKey>)` with a marker inside the quarantined dir; losers retry instead of deleting a successor guard. Guard writes `owner.json` `{token,pid,ts}` and its `finally` only removes the guard when the token matches.
|
||||
- Added injectable `now` clock. Tests: two waiters on one stale guard (both events survive), foreign-token guard release keeps the successor, plus existing lock/race/compaction tests.
|
||||
- `npx tsc --noEmit` 0; `npm test` 78/78 pass; `npm run build` emits both bundles. Logs: `.opencode/tmp/runs/f11-guard2-{test,build}.log` and `f11-guard-final-*`.
|
||||
- Remaining manual checks: live PM2 multi-worker append/compaction, live scheduler retry tick, browser module rendering.
|
||||
|
||||
## 2026-10-30 — F11 stale-guard snapshot verification
|
||||
|
||||
- Implementation head: `feat/F11` @ `4bd69dd` (progress commit follows).
|
||||
- Stale-guard branch now snapshots `observedMtime` + `observedOwner` before any re-read, quarantines with `guardPath.stale.<pid>.<uuid>`, then verifies the moved directory's owner token (or both null) and mtime against the snapshot. Mismatch restores the successor guard and retries; restore failure returns false without running the action.
|
||||
- Added deterministic successor-replacement test via guard hooks; both appends survive and the successor guard is restored.
|
||||
- `npx tsc --noEmit` 0; `npm test` 79/79 pass; `npm run build` emits both bundles. Logs: `.opencode/tmp/runs/f11-guard3-{test,build}.log` and `f11-guard3-final-*`.
|
||||
- Remaining manual checks: live PM2 multi-worker append/compaction, live scheduler retry tick, browser module rendering.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
```
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
18
README.md
18
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 `<state-dir>/failures.jsonl`. Replay derives the latest state per task, so details survive
|
||||
restarts and PM2 workers. The journal is append-only; it is compacted under `run.lock` to keep unresolved
|
||||
tasks plus the last 1,000 recovered ones.
|
||||
|
||||
Retry policy: max 3 attempts (initial + 2 retries), backoff 5 minutes then 15 minutes. Due retries run
|
||||
inside the maintenance window only when no backfill is active, under the same `run.lock`, and process only
|
||||
the targeted file/preset/format pairs (default batch 50). A success appends the normal `state.jsonl`
|
||||
completion and marks the task `recovered`; an exhausted task stays `failed` with its last error and is not
|
||||
retried again. Failures are logged at warn with ids/metadata only; recoveries at info.
|
||||
|
||||
## Safe testing & deployment
|
||||
|
||||
Never test against a production instance.
|
||||
|
|
|
|||
|
|
@ -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<string, FailureFileMeta>();
|
||||
|
||||
if (ids.length > 0) {
|
||||
const schema = await getSchema();
|
||||
const items = new services['ItemsService']('directus_files', { schema });
|
||||
const rows = await items.readByQuery({
|
||||
filter: { id: { _in: ids } },
|
||||
fields: [ 'id', 'filename_download', 'type', 'folder' ],
|
||||
limit: ids.length,
|
||||
});
|
||||
|
||||
for (const row of Array.isArray(rows) ? rows : []) {
|
||||
files.set(String(row.id), row as FailureFileMeta);
|
||||
}
|
||||
}
|
||||
|
||||
res.json({ ...result, records: enrichFailures(result.records, files) });
|
||||
}),
|
||||
);
|
||||
|
||||
router.post(
|
||||
'/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,
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<string>,
|
||||
failures: Map<string, FailureRecord>,
|
||||
): FileTaskPlan {
|
||||
const tasks: WarmTaskSpec[] = [];
|
||||
let completedCount = 0;
|
||||
let skipped = 0;
|
||||
|
||||
for (const preset of presets) {
|
||||
for (const variant of formatsForPreset(preset, formats)) {
|
||||
const format = effectiveFormat(preset, variant);
|
||||
const key = completedKeyOf(fileId, preset.key, format);
|
||||
|
||||
if (completed.has(key)) {
|
||||
completedCount += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (failures.get(key)?.status === 'failed') {
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
tasks.push({ preset, format: variant });
|
||||
}
|
||||
}
|
||||
|
||||
return { tasks, completed: completedCount, skipped };
|
||||
}
|
||||
|
||||
let inProcessRunning = false;
|
||||
|
||||
async function countImageFiles(database: any): Promise<number> {
|
||||
|
|
@ -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<string>();
|
||||
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;
|
||||
|
|
|
|||
649
src/lib/failures.test.ts
Normal file
649
src/lib/failures.test.ts
Normal file
|
|
@ -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<string, string>) => Promise<void>): Promise<void> {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'image-previews-failures-'));
|
||||
|
||||
try {
|
||||
await run({ IMAGE_PREVIEWS_STATE_DIR: dir });
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
async function deadPid(): Promise<number> {
|
||||
const child = spawn(process.execPath, [ '-e', '' ], { stdio: 'ignore' });
|
||||
const pid = child.pid;
|
||||
|
||||
if (!pid) throw new Error('child process has no pid');
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
child.once('exit', () => resolve());
|
||||
child.once('error', reject);
|
||||
});
|
||||
|
||||
return pid;
|
||||
}
|
||||
|
||||
async function sleep(ms: number): Promise<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
function event(overrides: Partial<FailureEvent> & Pick<FailureEvent, 'type' | 'ts' | 'file_id'>): FailureEvent {
|
||||
return {
|
||||
type: overrides.type,
|
||||
ts: overrides.ts,
|
||||
file_id: overrides.file_id,
|
||||
preset_key: overrides.preset_key ?? 'system-medium-cover',
|
||||
format: overrides.format === undefined ? 'avif' : overrides.format,
|
||||
attempt: overrides.attempt ?? 1,
|
||||
next_retry_at: overrides.next_retry_at,
|
||||
error: overrides.error,
|
||||
ms: overrides.ms,
|
||||
};
|
||||
}
|
||||
|
||||
async function waitForGuardOwner(env: Record<string, string>, token: string): Promise<void> {
|
||||
const guard = failuresReclaimPath(env);
|
||||
const deadline = Date.now() + 1000;
|
||||
|
||||
while (Date.now() < deadline) {
|
||||
try {
|
||||
const owner = JSON.parse(await readFile(`${guard}/owner.json`, 'utf8')) as { token?: string };
|
||||
if (owner.token === token) return;
|
||||
} catch {
|
||||
await sleep(5);
|
||||
continue;
|
||||
}
|
||||
|
||||
await sleep(5);
|
||||
}
|
||||
|
||||
throw new Error(`guard owner ${token} was not restored`);
|
||||
}
|
||||
|
||||
async function writeLock(env: Record<string, string>, owner: Record<string, unknown> = {}): Promise<void> {
|
||||
await writeFile(
|
||||
failuresLockPath(env),
|
||||
JSON.stringify({
|
||||
token: 'test-owner',
|
||||
pid: process.pid,
|
||||
bootId: null,
|
||||
startTime: null,
|
||||
ts: Date.now(),
|
||||
...owner,
|
||||
}),
|
||||
'utf8',
|
||||
);
|
||||
}
|
||||
|
||||
test('appendFailure + replayFailures derive the latest state per task', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await appendFailure(
|
||||
event({ type: 'failed', ts: 1000, file_id: 'f1', attempt: 1, next_retry_at: 2000, error: 'boom', ms: 10 }),
|
||||
env,
|
||||
);
|
||||
await appendFailure(
|
||||
event({ type: 'failed', ts: 3000, file_id: 'f1', attempt: 2, next_retry_at: null, error: 'boom2', ms: 20 }),
|
||||
env,
|
||||
);
|
||||
await appendFailure(event({ type: 'recovered', ts: 4000, file_id: 'f1', attempt: 3, ms: 5 }), env);
|
||||
await appendFailure(event({ type: 'failed', ts: 5000, file_id: 'f2', format: null }), env);
|
||||
|
||||
const records = await replayFailures(env);
|
||||
assert.equal(records.length, 2);
|
||||
|
||||
const recovered = records.find((record) => record.file_id === 'f1');
|
||||
assert.equal(recovered?.status, 'recovered');
|
||||
assert.equal(recovered?.attempts, 3);
|
||||
assert.equal(recovered?.first_failed_at, 1000);
|
||||
assert.equal(recovered?.last_failed_at, 3000);
|
||||
assert.equal(recovered?.next_retry_at, null);
|
||||
assert.equal(recovered?.error, 'boom2');
|
||||
assert.equal(recovered?.ms, 5);
|
||||
|
||||
const failed = records.find((record) => record.file_id === 'f2');
|
||||
assert.equal(failed?.status, 'failed');
|
||||
assert.equal(failed?.attempts, 1);
|
||||
assert.equal(failed?.format, null);
|
||||
});
|
||||
});
|
||||
|
||||
test('replayFailures tolerates corrupt lines and is idempotent', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await ensureStateDir(env);
|
||||
await appendFile(failuresFilePath(env), 'not-json\n{"type":"nope"}\n', 'utf8');
|
||||
await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env);
|
||||
|
||||
const first = await replayFailures(env);
|
||||
const second = await replayFailures(env);
|
||||
|
||||
assert.equal(first.length, 1);
|
||||
assert.deepEqual(first, second);
|
||||
});
|
||||
});
|
||||
|
||||
test('compactFailures keeps unresolved rows and the last 1,000 recovered rows', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await ensureStateDir(env);
|
||||
const lines: string[] = [];
|
||||
|
||||
for (let index = 0; index < 1002; index += 1) {
|
||||
lines.push(
|
||||
JSON.stringify(
|
||||
event({
|
||||
type: 'recovered',
|
||||
ts: index + 1,
|
||||
file_id: `recovered-${index}`,
|
||||
attempt: 2,
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
lines.push(JSON.stringify(event({ type: 'failed', ts: 5000, file_id: 'unresolved-a', next_retry_at: null })));
|
||||
lines.push(JSON.stringify(event({ type: 'failed', ts: 5001, file_id: 'unresolved-b', next_retry_at: 6000 })));
|
||||
await appendFile(failuresFilePath(env), `${lines.join('\n')}\n`, 'utf8');
|
||||
|
||||
const stats = await compactFailures(env);
|
||||
const records = await replayFailures(env);
|
||||
|
||||
assert.notEqual(stats, false);
|
||||
if (stats === false) throw new Error('compaction did not acquire the lock');
|
||||
assert.equal(stats.kept, 1002);
|
||||
assert.equal(stats.dropped, 2);
|
||||
assert.equal(records.length, 1002);
|
||||
assert.ok(records.some((record) => record.file_id === 'unresolved-a' && record.status === 'failed'));
|
||||
assert.ok(records.some((record) => record.file_id === 'unresolved-b' && record.status === 'failed'));
|
||||
assert.ok(!records.some((record) => record.file_id === 'recovered-0'));
|
||||
assert.ok(!records.some((record) => record.file_id === 'recovered-1'));
|
||||
assert.ok(records.some((record) => record.file_id === 'recovered-2'));
|
||||
assert.ok(records.some((record) => record.file_id === 'recovered-1001'));
|
||||
});
|
||||
});
|
||||
|
||||
test('listFailures filters by derived status and paginates', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'exhausted', next_retry_at: null, error: 'a' }), env);
|
||||
await appendFailure(event({ type: 'failed', ts: 2000, file_id: 'pending', next_retry_at: 3000, error: 'b' }), env);
|
||||
await appendFailure(event({ type: 'recovered', ts: 3000, file_id: 'done', attempt: 2 }), env);
|
||||
|
||||
const exhausted = await listFailures(env, { status: 'failed', limit: 50, offset: 0 });
|
||||
assert.equal(exhausted.total, 1);
|
||||
assert.equal(exhausted.records[0]?.file_id, 'exhausted');
|
||||
assert.equal(exhausted.records[0]?.status, 'failed');
|
||||
|
||||
const pending = await listFailures(env, { status: 'retry', limit: 50, offset: 0 });
|
||||
assert.equal(pending.total, 1);
|
||||
assert.equal(pending.records[0]?.file_id, 'pending');
|
||||
assert.equal(pending.records[0]?.status, 'retry');
|
||||
|
||||
const recovered = await listFailures(env, { status: 'recovered', limit: 50, offset: 0 });
|
||||
assert.equal(recovered.total, 1);
|
||||
assert.equal(recovered.records[0]?.file_id, 'done');
|
||||
|
||||
const all = await listFailures(env, { limit: 2, offset: 0 });
|
||||
assert.equal(all.total, 3);
|
||||
assert.equal(all.limit, 2);
|
||||
assert.equal(all.offset, 0);
|
||||
assert.equal(all.records.length, 2);
|
||||
assert.equal(all.records[0]?.file_id, 'done');
|
||||
|
||||
const page = await listFailures(env, { limit: 1, offset: 1 });
|
||||
assert.equal(page.total, 3);
|
||||
assert.equal(page.records.length, 1);
|
||||
assert.equal(page.records[0]?.file_id, 'pending');
|
||||
|
||||
const clamped = await listFailures(env, { limit: 999, offset: 0 });
|
||||
assert.equal(clamped.limit, 200);
|
||||
});
|
||||
});
|
||||
|
||||
test('replay resets a new failure cycle and ignores duplicate failed events', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1', attempt: 1, next_retry_at: 2000 }), env);
|
||||
await appendFailure(event({ type: 'recovered', ts: 2000, file_id: 'f1', attempt: 2 }), env);
|
||||
await appendFailure(
|
||||
event({ type: 'failed', ts: 3000, file_id: 'f1', attempt: 1, next_retry_at: 4000, error: 'again' }),
|
||||
env,
|
||||
);
|
||||
await appendFailure(
|
||||
event({ type: 'failed', ts: 3000, file_id: 'f1', attempt: 1, next_retry_at: 4000, error: 'again' }),
|
||||
env,
|
||||
);
|
||||
|
||||
const records = await replayFailures(env);
|
||||
const record = records.find((candidate) => candidate.file_id === 'f1');
|
||||
|
||||
assert.equal(record?.attempts, 1);
|
||||
assert.equal(record?.status, 'failed');
|
||||
assert.equal(record?.first_failed_at, 3000);
|
||||
assert.equal(record?.last_failed_at, 3000);
|
||||
assert.equal(record?.next_retry_at, 4000);
|
||||
});
|
||||
});
|
||||
|
||||
test('compactFailures returns false and leaves the file untouched when a live owner holds the lock', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env);
|
||||
const held = await acquireFailureLock(env);
|
||||
const before = await readFile(failuresFilePath(env), 'utf8');
|
||||
|
||||
try {
|
||||
const result = await compactFailures(env, { timeoutMs: 30, waitMs: 5 });
|
||||
assert.equal(result, false);
|
||||
assert.equal(await readFile(failuresFilePath(env), 'utf8'), before);
|
||||
} finally {
|
||||
await held.release();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
test('appendFailure returns false and leaves the file unchanged when a live owner holds the lock', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env);
|
||||
const held = await acquireFailureLock(env);
|
||||
const before = await readFile(failuresFilePath(env), 'utf8');
|
||||
const warnings: Array<Record<string, unknown>> = [];
|
||||
|
||||
try {
|
||||
const result = await appendFailure(
|
||||
event({ type: 'failed', ts: 2000, file_id: 'f2' }),
|
||||
env,
|
||||
{
|
||||
timeoutMs: 30,
|
||||
waitMs: 5,
|
||||
logger: { warn: (payload: Record<string, unknown>) => warnings.push(payload) },
|
||||
},
|
||||
);
|
||||
|
||||
assert.equal(result, false);
|
||||
assert.equal(await readFile(failuresFilePath(env), 'utf8'), before);
|
||||
assert.equal(warnings[0]?.['fileId'], 'f2');
|
||||
assert.equal(warnings[0]?.['preset_key'], 'system-medium-cover');
|
||||
assert.equal(warnings[0]?.['format'], 'avif');
|
||||
} finally {
|
||||
await held.release();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
test('appendFailure breaks a dead-pid lock and appends', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const pid = await deadPid();
|
||||
await writeLock(env, { token: 'dead-owner', pid });
|
||||
|
||||
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
|
||||
timeoutMs: 500,
|
||||
waitMs: 5,
|
||||
});
|
||||
|
||||
assert.equal(result, true);
|
||||
assert.equal((await replayFailures(env)).length, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test('release with a foreign token does not remove the lock', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const held = await acquireFailureLock(env);
|
||||
await writeLock(env, { token: 'foreign-owner' });
|
||||
|
||||
await held.release();
|
||||
|
||||
const owner = JSON.parse(await readFile(failuresLockPath(env), 'utf8')) as {
|
||||
token?: string;
|
||||
};
|
||||
assert.equal(owner.token, 'foreign-owner');
|
||||
await rm(failuresLockPath(env), { force: true });
|
||||
});
|
||||
});
|
||||
|
||||
test('missing owner younger than grace is not reclaimed', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await writeFile(failuresLockPath(env), '', 'utf8');
|
||||
|
||||
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
|
||||
timeoutMs: 30,
|
||||
waitMs: 5,
|
||||
graceMs: 100,
|
||||
});
|
||||
|
||||
assert.equal(result, false);
|
||||
assert.equal(await readFile(failuresLockPath(env), 'utf8'), '');
|
||||
});
|
||||
});
|
||||
|
||||
test('missing owner older than grace is reclaimed', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await writeFile(failuresLockPath(env), '', 'utf8');
|
||||
await utimes(failuresLockPath(env), new Date(0), new Date(0));
|
||||
|
||||
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
|
||||
timeoutMs: 500,
|
||||
waitMs: 5,
|
||||
graceMs: 10,
|
||||
});
|
||||
|
||||
assert.equal(result, true);
|
||||
assert.equal((await replayFailures(env)).length, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test('corrupt owner older than grace is treated like a missing owner', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await writeFile(failuresLockPath(env), '{not-json', 'utf8');
|
||||
await utimes(failuresLockPath(env), new Date(0), new Date(0));
|
||||
|
||||
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
|
||||
timeoutMs: 500,
|
||||
waitMs: 5,
|
||||
graceMs: 10,
|
||||
});
|
||||
|
||||
assert.equal(result, true);
|
||||
assert.equal((await replayFailures(env)).length, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test('PID reuse with a foreign bootId is reclaimed', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await writeLock(env, { token: 'reused-boot', bootId: 'foreign-boot' });
|
||||
|
||||
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
|
||||
timeoutMs: 500,
|
||||
waitMs: 5,
|
||||
readBootId: async () => 'current-boot',
|
||||
readStartTime: async () => 'current-start',
|
||||
});
|
||||
|
||||
assert.equal(result, true);
|
||||
assert.equal((await replayFailures(env)).length, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test('PID reuse with a foreign startTime is reclaimed', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await writeLock(env, { token: 'reused-start', startTime: 'foreign-start' });
|
||||
|
||||
const result = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
|
||||
timeoutMs: 500,
|
||||
waitMs: 5,
|
||||
readBootId: async () => null,
|
||||
readStartTime: async () => 'current-start',
|
||||
});
|
||||
|
||||
assert.equal(result, true);
|
||||
assert.equal((await replayFailures(env)).length, 1);
|
||||
});
|
||||
});
|
||||
|
||||
test('reclaim waits for the guard and a stale guard is removed', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const pid = await deadPid();
|
||||
await writeLock(env, { token: 'dead-owner', pid });
|
||||
const guard = failuresReclaimPath(env);
|
||||
await mkdir(guard);
|
||||
|
||||
const blocked = await appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
|
||||
timeoutMs: 30,
|
||||
waitMs: 5,
|
||||
graceMs: 0,
|
||||
guardStaleMs: 30_000,
|
||||
});
|
||||
|
||||
assert.equal(blocked, false);
|
||||
assert.match(await readFile(failuresLockPath(env), 'utf8'), /dead-owner/);
|
||||
|
||||
const recovered = await appendFailure(event({ type: 'failed', ts: 2000, file_id: 'f2' }), env, {
|
||||
timeoutMs: 500,
|
||||
waitMs: 5,
|
||||
graceMs: 0,
|
||||
guardStaleMs: 30_000,
|
||||
now: () => Date.now() + 31_000,
|
||||
});
|
||||
|
||||
assert.equal(recovered, true);
|
||||
assert.equal((await replayFailures(env)).length, 1);
|
||||
|
||||
const names = await readdir(env['IMAGE_PREVIEWS_STATE_DIR'] as string);
|
||||
assert.ok(!names.some((name) => name.startsWith('failures.lock.reclaim')));
|
||||
});
|
||||
});
|
||||
|
||||
test('two waiters over the same stale guard reclaim once and keep both events', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const pid = await deadPid();
|
||||
await writeLock(env, { token: 'dead-owner', pid });
|
||||
const guard = failuresReclaimPath(env);
|
||||
await mkdir(guard);
|
||||
await utimes(guard, new Date(0), new Date(0));
|
||||
|
||||
const options = { timeoutMs: 2000, waitMs: 5, graceMs: 0, guardStaleMs: 30_000 };
|
||||
const results = await Promise.all([
|
||||
appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, options),
|
||||
appendFailure(event({ type: 'failed', ts: 2000, file_id: 'f2' }), env, options),
|
||||
]);
|
||||
|
||||
assert.deepEqual(results, [ true, true ]);
|
||||
assert.equal((await replayFailures(env)).length, 2);
|
||||
|
||||
const names = await readdir(env['IMAGE_PREVIEWS_STATE_DIR'] as string);
|
||||
assert.ok(!names.some((name) => name.startsWith('failures.lock.reclaim')));
|
||||
});
|
||||
});
|
||||
|
||||
test('a successor guard replaced between observation and rename is restored', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const pid = await deadPid();
|
||||
await writeLock(env, { token: 'dead-owner', pid });
|
||||
const guard = failuresReclaimPath(env);
|
||||
await mkdir(guard);
|
||||
await writeFile(
|
||||
`${guard}/owner.json`,
|
||||
JSON.stringify({ token: 'stale-guard', pid: process.pid, ts: 0 }),
|
||||
'utf8',
|
||||
);
|
||||
await utimes(guard, new Date(0), new Date(0));
|
||||
|
||||
let observedResolve: () => void = () => {};
|
||||
const observed = new Promise<void>((resolve) => {
|
||||
observedResolve = resolve;
|
||||
});
|
||||
let resumeResolve: () => void = () => {};
|
||||
const resume = new Promise<void>((resolve) => {
|
||||
resumeResolve = resolve;
|
||||
});
|
||||
let rejectedResolve: () => void = () => {};
|
||||
const rejected = new Promise<void>((resolve) => {
|
||||
rejectedResolve = resolve;
|
||||
});
|
||||
let releaseResolve: () => void = () => {};
|
||||
const release = new Promise<void>((resolve) => {
|
||||
releaseResolve = resolve;
|
||||
});
|
||||
|
||||
const first = appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
|
||||
timeoutMs: 2000,
|
||||
waitMs: 5,
|
||||
graceMs: 0,
|
||||
guardStaleMs: 30_000,
|
||||
onStaleGuardObserved: async () => {
|
||||
observedResolve();
|
||||
await resume;
|
||||
},
|
||||
onGuardRejected: async () => {
|
||||
rejectedResolve();
|
||||
await release;
|
||||
},
|
||||
});
|
||||
|
||||
await observed;
|
||||
|
||||
const replaced = `${guard}.replaced`;
|
||||
await rename(guard, replaced);
|
||||
await mkdir(guard);
|
||||
await writeFile(
|
||||
`${guard}/owner.json`,
|
||||
JSON.stringify({ token: 'successor-guard', pid: process.pid, ts: Date.now() }),
|
||||
'utf8',
|
||||
);
|
||||
|
||||
resumeResolve();
|
||||
await rejected;
|
||||
await waitForGuardOwner(env, 'successor-guard');
|
||||
|
||||
const second = appendFailure(event({ type: 'failed', ts: 2000, file_id: 'f2' }), env, {
|
||||
timeoutMs: 2000,
|
||||
waitMs: 5,
|
||||
graceMs: 0,
|
||||
guardStaleMs: 30_000,
|
||||
});
|
||||
|
||||
await rm(guard, { recursive: true, force: true });
|
||||
releaseResolve();
|
||||
|
||||
const results = await Promise.all([ first, second ]);
|
||||
|
||||
assert.deepEqual(results, [ true, true ]);
|
||||
assert.equal((await replayFailures(env)).length, 2);
|
||||
await rm(replaced, { recursive: true, force: true });
|
||||
});
|
||||
});
|
||||
|
||||
test('foreign-token guard release does not remove a successor guard', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const guard = failuresReclaimPath(env);
|
||||
await mkdir(guard);
|
||||
await writeFile(
|
||||
`${guard}/owner.json`,
|
||||
JSON.stringify({ token: 'successor-guard', pid: process.pid, ts: Date.now() }),
|
||||
'utf8',
|
||||
);
|
||||
|
||||
await releaseReclaimGuard(env, 'old-guard');
|
||||
|
||||
const owner = JSON.parse(await readFile(`${guard}/owner.json`, 'utf8')) as { token?: string };
|
||||
assert.equal(owner.token, 'successor-guard');
|
||||
|
||||
await releaseReclaimGuard(env, 'successor-guard');
|
||||
await assert.rejects(readFile(`${guard}/owner.json`, 'utf8'));
|
||||
});
|
||||
});
|
||||
|
||||
test('concurrent reclaim of a dead lock loses no event', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const pid = await deadPid();
|
||||
await writeLock(env, { token: 'dead-owner', pid });
|
||||
|
||||
const results = await Promise.all([
|
||||
appendFailure(event({ type: 'failed', ts: 1000, file_id: 'f1' }), env, {
|
||||
timeoutMs: 1000,
|
||||
waitMs: 5,
|
||||
}),
|
||||
appendFailure(event({ type: 'failed', ts: 2000, file_id: 'f2' }), env, {
|
||||
timeoutMs: 1000,
|
||||
waitMs: 5,
|
||||
}),
|
||||
]);
|
||||
|
||||
assert.deepEqual(results, [ true, true ]);
|
||||
assert.equal((await replayFailures(env)).length, 2);
|
||||
});
|
||||
});
|
||||
|
||||
test('50 appends racing one compaction with a temporarily held lock keep every key', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const held = await acquireFailureLock(env);
|
||||
const appends = Array.from({ length: 50 }, (_, index) =>
|
||||
appendFailure(event({ type: 'failed', ts: 1000 + index, file_id: `race-${index}`, next_retry_at: null }), env),
|
||||
);
|
||||
const compaction = compactFailures(env);
|
||||
|
||||
await sleep(50);
|
||||
await held.release();
|
||||
await Promise.all([ compaction, ...appends ]);
|
||||
|
||||
const records = await replayFailures(env);
|
||||
assert.equal(records.length, 50);
|
||||
|
||||
for (let index = 0; index < 50; index += 1) {
|
||||
assert.ok(records.some((record) => record.file_id === `race-${index}`));
|
||||
}
|
||||
|
||||
const names = await readdir(env['IMAGE_PREVIEWS_STATE_DIR'] as string);
|
||||
assert.ok(!names.some((name) => name.endsWith('.tmp')));
|
||||
});
|
||||
});
|
||||
|
||||
test('parseFailureQuery validates status and clamps limit/offset', () => {
|
||||
assert.deepEqual(parseFailureQuery({ status: 'failed', limit: '999', offset: '-3' }), {
|
||||
status: 'failed',
|
||||
limit: 200,
|
||||
offset: 0,
|
||||
});
|
||||
assert.deepEqual(parseFailureQuery({}), { status: null, limit: 50, offset: 0 });
|
||||
assert.deepEqual(parseFailureQuery({ status: 'retry', limit: '0', offset: '2' }), {
|
||||
status: 'retry',
|
||||
limit: 1,
|
||||
offset: 2,
|
||||
});
|
||||
assert.throws(() => parseFailureQuery({ status: 'bogus' }), /status must be failed, recovered or retry/);
|
||||
});
|
||||
|
||||
test('enrichFailures adds file metadata, timestamps and human durations', () => {
|
||||
const records: FailureRecord[] = [
|
||||
{
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
attempts: 2,
|
||||
status: 'failed',
|
||||
first_failed_at: 1000,
|
||||
last_failed_at: 2000,
|
||||
next_retry_at: 3000,
|
||||
error: 'boom',
|
||||
ms: 12,
|
||||
},
|
||||
];
|
||||
const files = new Map([
|
||||
[ 'f1', { filename_download: 'photo.jpg', type: 'image/jpeg', folder: 'folder-1' } ],
|
||||
]);
|
||||
const enriched = enrichFailures(records, files);
|
||||
const record = enriched[0];
|
||||
|
||||
assert.deepEqual(record?.file, {
|
||||
filename_download: 'photo.jpg',
|
||||
type: 'image/jpeg',
|
||||
folder: 'folder-1',
|
||||
});
|
||||
assert.equal(record?.attempts, 2);
|
||||
assert.equal(record?.first_failed_at, 1000);
|
||||
assert.equal(record?.last_failed_at, 2000);
|
||||
assert.equal(record?.next_retry_at, 3000);
|
||||
assert.equal(record?.status, 'failed');
|
||||
assert.equal(record?.error, 'boom');
|
||||
assert.equal(record?.first_failed_at_iso, new Date(1000).toISOString());
|
||||
assert.equal(record?.last_failed_at_iso, new Date(2000).toISOString());
|
||||
assert.equal(record?.next_retry_at_iso, new Date(3000).toISOString());
|
||||
assert.equal(record?.next_retry_at_human, new Date(3000).toLocaleString());
|
||||
assert.equal(record?.ms_human, '12 ms');
|
||||
});
|
||||
584
src/lib/failures.ts
Normal file
584
src/lib/failures.ts
Normal file
|
|
@ -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<void>;
|
||||
onGuardRejected?: () => Promise<void>;
|
||||
logger?: any;
|
||||
readBootId?: () => Promise<string | null>;
|
||||
readStartTime?: (pid: number) => Promise<string | null>;
|
||||
}
|
||||
|
||||
export interface FailureLock {
|
||||
acquired: boolean;
|
||||
token: string;
|
||||
release: () => Promise<void>;
|
||||
}
|
||||
|
||||
interface LockOwner {
|
||||
token: string;
|
||||
pid: number;
|
||||
bootId: string | null;
|
||||
startTime: string | null;
|
||||
ts: number;
|
||||
}
|
||||
|
||||
export function failuresFilePath(env: EnvLike): string {
|
||||
return stateFilePath(env, FAILURES_FILE);
|
||||
}
|
||||
|
||||
export function failuresLockPath(env: EnvLike): string {
|
||||
return stateFilePath(env, FAILURES_LOCK);
|
||||
}
|
||||
|
||||
export function failuresReclaimPath(env: EnvLike): string {
|
||||
return `${failuresLockPath(env)}.reclaim`;
|
||||
}
|
||||
|
||||
async function sleep(ms: number): Promise<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
async function fileMtime(path: string): Promise<number | null> {
|
||||
try {
|
||||
return (await stat(path)).mtimeMs;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function defaultReadBootId(): Promise<string | null> {
|
||||
try {
|
||||
const value = (await readFile('/proc/sys/kernel/random/boot_id', 'utf8')).trim();
|
||||
return value.length > 0 ? value : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function defaultReadStartTime(pid: number): Promise<string | null> {
|
||||
try {
|
||||
const raw = await readFile(`/proc/${pid}/stat`, 'utf8');
|
||||
const close = raw.lastIndexOf(')');
|
||||
if (close < 0) return null;
|
||||
|
||||
const fields = raw.slice(close + 1).trim().split(/\s+/);
|
||||
return fields[19] ?? null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function readLockOwner(path: string): Promise<LockOwner | null> {
|
||||
try {
|
||||
const parsed = JSON.parse(await readFile(path, 'utf8')) as Record<string, unknown>;
|
||||
const token = parsed['token'];
|
||||
const pid = parsed['pid'];
|
||||
|
||||
if (typeof token !== 'string' || typeof pid !== 'number') return null;
|
||||
|
||||
const bootId = typeof parsed['bootId'] === 'string' ? parsed['bootId'] : null;
|
||||
const startTime = typeof parsed['startTime'] === 'string' ? parsed['startTime'] : null;
|
||||
const ts = typeof parsed['ts'] === 'number' ? parsed['ts'] : 0;
|
||||
|
||||
return { token, pid, bootId, startTime, ts };
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function isOwnerAlive(owner: LockOwner, bootId: string | null, startTime: string | null): boolean {
|
||||
if (!Number.isInteger(owner.pid) || owner.pid <= 0) return false;
|
||||
|
||||
try {
|
||||
process.kill(owner.pid, 0);
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === 'ESRCH') return false;
|
||||
}
|
||||
|
||||
if (owner.bootId && bootId && owner.bootId !== bootId) return false;
|
||||
if (owner.startTime && startTime && owner.startTime !== startTime) return false;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
async function unlinkLock(path: string): Promise<boolean> {
|
||||
try {
|
||||
await unlink(path);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function releaseFailureLock(env: EnvLike, token: string): Promise<void> {
|
||||
const owner = await readLockOwner(failuresLockPath(env));
|
||||
|
||||
if (!owner || owner.token !== token) return;
|
||||
|
||||
await unlinkLock(failuresLockPath(env));
|
||||
}
|
||||
|
||||
export async function releaseReclaimGuard(env: EnvLike, token: string): Promise<void> {
|
||||
const guardPath = failuresReclaimPath(env);
|
||||
const owner = await readLockOwner(`${guardPath}/owner.json`);
|
||||
|
||||
if (!owner || owner.token !== token) return;
|
||||
|
||||
try {
|
||||
await rm(guardPath, { recursive: true, force: true });
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
async function withReclaimGuard(
|
||||
env: EnvLike,
|
||||
options: FailureLockOptions,
|
||||
action: () => Promise<boolean>,
|
||||
): Promise<boolean> {
|
||||
const guardPath = failuresReclaimPath(env);
|
||||
const timeoutMs = options.timeoutMs ?? LOCK_TIMEOUT_MS;
|
||||
const waitMs = options.waitMs ?? LOCK_WAIT_MS;
|
||||
const guardStaleMs = options.guardStaleMs ?? GUARD_STALE_MS;
|
||||
const now = options.now ?? (() => Date.now());
|
||||
const token = randomUUID();
|
||||
const deadline = now() + timeoutMs;
|
||||
const quarantined: string[] = [];
|
||||
let owned = false;
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
try {
|
||||
await mkdir(guardPath);
|
||||
|
||||
try {
|
||||
await writeFile(
|
||||
`${guardPath}/owner.json`,
|
||||
JSON.stringify({ token, pid: process.pid, ts: now() }),
|
||||
'utf8',
|
||||
);
|
||||
} catch (error) {
|
||||
await rm(guardPath, { recursive: true, force: true });
|
||||
throw error;
|
||||
}
|
||||
|
||||
owned = true;
|
||||
break;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'EEXIST') return false;
|
||||
|
||||
const observedMtime = await fileMtime(guardPath);
|
||||
const observedOwner =
|
||||
observedMtime === null ? null : await readLockOwner(`${guardPath}/owner.json`);
|
||||
|
||||
if (observedMtime !== null && now() - observedMtime > guardStaleMs) {
|
||||
await options.onStaleGuardObserved?.();
|
||||
|
||||
const stalePath = `${guardPath}.stale.${process.pid}.${randomUUID()}`;
|
||||
|
||||
try {
|
||||
await rename(guardPath, stalePath);
|
||||
} catch {
|
||||
if (now() >= deadline) return false;
|
||||
await sleep(waitMs);
|
||||
continue;
|
||||
}
|
||||
|
||||
const movedMtime = await fileMtime(stalePath);
|
||||
const movedOwner = await readLockOwner(`${stalePath}/owner.json`);
|
||||
const ownerMatches =
|
||||
(observedOwner?.token ?? null) === (movedOwner?.token ?? null);
|
||||
const mtimeMatches =
|
||||
observedMtime !== null && movedMtime !== null && movedMtime === observedMtime;
|
||||
|
||||
if (!ownerMatches || !mtimeMatches) {
|
||||
try {
|
||||
await rename(stalePath, guardPath);
|
||||
} catch {
|
||||
quarantined.push(stalePath);
|
||||
return false;
|
||||
}
|
||||
|
||||
await options.onGuardRejected?.();
|
||||
|
||||
if (now() >= deadline) return false;
|
||||
|
||||
await sleep(waitMs);
|
||||
continue;
|
||||
}
|
||||
|
||||
quarantined.push(stalePath);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (now() >= deadline) return false;
|
||||
|
||||
await sleep(waitMs);
|
||||
}
|
||||
}
|
||||
|
||||
return await action();
|
||||
} finally {
|
||||
if (owned) await releaseReclaimGuard(env, token);
|
||||
|
||||
for (const path of quarantined) {
|
||||
try {
|
||||
await rm(path, { recursive: true, force: true });
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function reclaimLock(env: EnvLike, options: FailureLockOptions): Promise<boolean> {
|
||||
const lockPath = failuresLockPath(env);
|
||||
const graceMs = options.graceMs ?? LOCK_GRACE_MS;
|
||||
|
||||
return withReclaimGuard(env, options, async () => {
|
||||
const mtime = await fileMtime(lockPath);
|
||||
if (mtime === null) return false;
|
||||
|
||||
const owner = await readLockOwner(lockPath);
|
||||
|
||||
if (!owner) {
|
||||
if (Date.now() - mtime <= graceMs) return false;
|
||||
return unlinkLock(lockPath);
|
||||
}
|
||||
|
||||
const bootId = await (options.readBootId ?? defaultReadBootId)();
|
||||
const startTime = await (options.readStartTime ?? defaultReadStartTime)(owner.pid);
|
||||
|
||||
if (isOwnerAlive(owner, bootId, startTime)) return false;
|
||||
|
||||
return unlinkLock(lockPath);
|
||||
});
|
||||
}
|
||||
|
||||
export async function acquireFailureLock(
|
||||
env: EnvLike,
|
||||
options: FailureLockOptions = {},
|
||||
): Promise<FailureLock> {
|
||||
await ensureStateDir(env);
|
||||
|
||||
const timeoutMs = options.timeoutMs ?? LOCK_TIMEOUT_MS;
|
||||
const waitMs = options.waitMs ?? LOCK_WAIT_MS;
|
||||
const graceMs = options.graceMs ?? LOCK_GRACE_MS;
|
||||
const token = randomUUID();
|
||||
const bootId = await (options.readBootId ?? defaultReadBootId)();
|
||||
const startTime = await (options.readStartTime ?? defaultReadStartTime)(process.pid);
|
||||
const owner: LockOwner = { token, pid: process.pid, bootId, startTime, ts: Date.now() };
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
|
||||
while (true) {
|
||||
try {
|
||||
await writeFile(failuresLockPath(env), JSON.stringify(owner), { flag: 'wx' });
|
||||
return {
|
||||
acquired: true,
|
||||
token,
|
||||
release: async () => {
|
||||
await releaseFailureLock(env, token);
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'EEXIST') {
|
||||
return { acquired: false, token, release: async () => {} };
|
||||
}
|
||||
|
||||
const existing = await readLockOwner(failuresLockPath(env));
|
||||
|
||||
if (existing) {
|
||||
const existingStartTime = await (options.readStartTime ?? defaultReadStartTime)(existing.pid);
|
||||
|
||||
if (!isOwnerAlive(existing, bootId, existingStartTime)) {
|
||||
const reclaimed = await reclaimLock(env, options);
|
||||
|
||||
if (Date.now() >= deadline) return { acquired: false, token, release: async () => {} };
|
||||
if (!reclaimed) await sleep(waitMs);
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
const mtime = await fileMtime(failuresLockPath(env));
|
||||
|
||||
if (mtime !== null && Date.now() - mtime > graceMs) {
|
||||
const reclaimed = await reclaimLock(env, options);
|
||||
|
||||
if (Date.now() >= deadline) return { acquired: false, token, release: async () => {} };
|
||||
if (!reclaimed) await sleep(waitMs);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
if (Date.now() >= deadline) return { acquired: false, token, release: async () => {} };
|
||||
|
||||
await sleep(waitMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function appendFailure(
|
||||
event: FailureEvent,
|
||||
env: EnvLike,
|
||||
options: FailureLockOptions = {},
|
||||
): Promise<boolean> {
|
||||
await ensureStateDir(env);
|
||||
const lock = await acquireFailureLock(env, options);
|
||||
|
||||
if (!lock.acquired) {
|
||||
options.logger?.warn?.(
|
||||
{ fileId: event.file_id, preset_key: event.preset_key, format: event.format },
|
||||
'image-previews: could not journal failure',
|
||||
);
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
await appendFile(failuresFilePath(env), `${JSON.stringify(event)}\n`, 'utf8');
|
||||
return true;
|
||||
} finally {
|
||||
await lock.release();
|
||||
}
|
||||
}
|
||||
|
||||
function isFailureEvent(value: unknown): value is FailureEvent {
|
||||
if (!value || typeof value !== 'object') return false;
|
||||
|
||||
const event = value as Record<string, unknown>;
|
||||
const type = event['type'];
|
||||
|
||||
if (type !== 'failed' && type !== 'retry' && type !== 'recovered') return false;
|
||||
if (typeof event['ts'] !== 'number' || !Number.isFinite(event['ts'])) return false;
|
||||
if (typeof event['file_id'] !== 'string' || event['file_id'].length === 0) return false;
|
||||
if (typeof event['preset_key'] !== 'string' || event['preset_key'].length === 0) return false;
|
||||
if (event['format'] !== null && typeof event['format'] !== 'string') return false;
|
||||
if (typeof event['attempt'] !== 'number' || !Number.isFinite(event['attempt'])) return false;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
export async function readFailureEvents(env: EnvLike): Promise<FailureEvent[]> {
|
||||
const events: FailureEvent[] = [];
|
||||
|
||||
try {
|
||||
const raw = await readFile(failuresFilePath(env), 'utf8');
|
||||
|
||||
for (const line of raw.split('\n')) {
|
||||
if (!line.trim()) continue;
|
||||
|
||||
try {
|
||||
const parsed = JSON.parse(line) as unknown;
|
||||
if (isFailureEvent(parsed)) events.push(parsed);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return events;
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
function eventKey(event: FailureEvent): string {
|
||||
return completedKeyOf(event.file_id, event.preset_key, event.format);
|
||||
}
|
||||
|
||||
export function replayFailureEvents(events: FailureEvent[]): FailureRecord[] {
|
||||
const records = new Map<string, FailureRecord>();
|
||||
|
||||
for (const event of events) {
|
||||
const key = eventKey(event);
|
||||
const previous = records.get(key);
|
||||
const startsCycle = previous === undefined || previous.status === 'recovered';
|
||||
const firstFailedAt = startsCycle ? event.ts : previous.first_failed_at;
|
||||
const lastFailedAt = startsCycle ? event.ts : previous.last_failed_at;
|
||||
const nextRetryAt = event.next_retry_at ?? null;
|
||||
const error = event.error ?? previous?.error ?? null;
|
||||
const ms = event.ms ?? previous?.ms ?? null;
|
||||
|
||||
if (event.type === 'failed') {
|
||||
records.set(key, {
|
||||
file_id: event.file_id,
|
||||
preset_key: event.preset_key,
|
||||
format: event.format,
|
||||
attempts: event.attempt,
|
||||
status: 'failed',
|
||||
first_failed_at: firstFailedAt,
|
||||
last_failed_at: event.ts,
|
||||
next_retry_at: nextRetryAt,
|
||||
error,
|
||||
ms,
|
||||
});
|
||||
} else if (event.type === 'retry') {
|
||||
records.set(key, {
|
||||
file_id: event.file_id,
|
||||
preset_key: event.preset_key,
|
||||
format: event.format,
|
||||
attempts: event.attempt,
|
||||
status: previous?.status === 'recovered' ? 'failed' : (previous?.status ?? 'failed'),
|
||||
first_failed_at: firstFailedAt,
|
||||
last_failed_at: lastFailedAt,
|
||||
next_retry_at: nextRetryAt,
|
||||
error,
|
||||
ms,
|
||||
});
|
||||
} else {
|
||||
records.set(key, {
|
||||
file_id: event.file_id,
|
||||
preset_key: event.preset_key,
|
||||
format: event.format,
|
||||
attempts: event.attempt,
|
||||
status: 'recovered',
|
||||
first_failed_at: firstFailedAt,
|
||||
last_failed_at: lastFailedAt,
|
||||
next_retry_at: null,
|
||||
error,
|
||||
ms,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return [ ...records.values() ];
|
||||
}
|
||||
|
||||
export async function replayFailures(env: EnvLike): Promise<FailureRecord[]> {
|
||||
return replayFailureEvents(await readFailureEvents(env));
|
||||
}
|
||||
|
||||
export async function replayFailureMap(env: EnvLike): Promise<Map<string, FailureRecord>> {
|
||||
const records = await replayFailures(env);
|
||||
return new Map(records.map((record) => [ completedKeyOf(record.file_id, record.preset_key, record.format), record ]));
|
||||
}
|
||||
|
||||
export function failureStatusOf(record: FailureRecord): FailureRecord['status'] {
|
||||
if (record.status === 'recovered') return 'recovered';
|
||||
return record.next_retry_at !== null ? 'retry' : 'failed';
|
||||
}
|
||||
|
||||
function withDerivedStatus(record: FailureRecord): FailureRecord {
|
||||
const status = failureStatusOf(record);
|
||||
return status === record.status ? record : { ...record, status };
|
||||
}
|
||||
|
||||
export async function listFailures(env: EnvLike, query: FailureListQuery): Promise<FailureListResult> {
|
||||
const limit = Math.max(1, Math.min(200, Math.trunc(query.limit) || 50));
|
||||
const offset = Math.max(0, Math.trunc(query.offset) || 0);
|
||||
const records = (await replayFailures(env))
|
||||
.map(withDerivedStatus)
|
||||
.sort((a, b) => b.last_failed_at - a.last_failed_at || eventKeyLike(a).localeCompare(eventKeyLike(b)));
|
||||
const filtered = query.status ? records.filter((record) => record.status === query.status) : records;
|
||||
|
||||
return {
|
||||
records: filtered.slice(offset, offset + limit),
|
||||
total: filtered.length,
|
||||
limit,
|
||||
offset,
|
||||
};
|
||||
}
|
||||
|
||||
function eventKeyLike(record: FailureRecord): string {
|
||||
return completedKeyOf(record.file_id, record.preset_key, record.format);
|
||||
}
|
||||
|
||||
export function parseFailureQuery(input: { status?: unknown; limit?: unknown; offset?: unknown }): FailureListQuery {
|
||||
const rawStatus = typeof input.status === 'string' ? input.status : null;
|
||||
|
||||
if (rawStatus && !FAILURE_STATUSES.includes(rawStatus as FailureStatus)) {
|
||||
throw new Error('status must be failed, recovered or retry');
|
||||
}
|
||||
|
||||
const limitRaw = Number(input.limit);
|
||||
const offsetRaw = Number(input.offset);
|
||||
|
||||
return {
|
||||
status: (rawStatus as FailureStatus | null) ?? null,
|
||||
limit: Number.isFinite(limitRaw) ? Math.min(200, Math.max(1, Math.trunc(limitRaw))) : 50,
|
||||
offset: Number.isFinite(offsetRaw) ? Math.max(0, Math.trunc(offsetRaw)) : 0,
|
||||
};
|
||||
}
|
||||
|
||||
export function enrichFailures(
|
||||
records: FailureRecord[],
|
||||
files: Map<string, FailureFileMeta>,
|
||||
): FailureEnrichedRecord[] {
|
||||
return records.map((record) => {
|
||||
const file = files.get(record.file_id);
|
||||
|
||||
return {
|
||||
...record,
|
||||
file: {
|
||||
filename_download: file?.filename_download ?? null,
|
||||
type: file?.type ?? null,
|
||||
folder: file?.folder ?? null,
|
||||
},
|
||||
first_failed_at_iso: toIso(record.first_failed_at),
|
||||
first_failed_at_human: new Date(record.first_failed_at).toLocaleString(),
|
||||
last_failed_at_iso: toIso(record.last_failed_at),
|
||||
last_failed_at_human: new Date(record.last_failed_at).toLocaleString(),
|
||||
next_retry_at_iso: toIso(record.next_retry_at),
|
||||
next_retry_at_human: record.next_retry_at ? new Date(record.next_retry_at).toLocaleString() : null,
|
||||
ms_human: record.ms === null ? null : humanMs(record.ms),
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
export async function compactFailures(
|
||||
env: EnvLike,
|
||||
options: FailureLockOptions = {},
|
||||
): Promise<false | { kept: number; dropped: number }> {
|
||||
const lock = await acquireFailureLock(env, options);
|
||||
if (!lock.acquired) return false;
|
||||
|
||||
try {
|
||||
const events = await readFailureEvents(env);
|
||||
const records = replayFailureEvents(events);
|
||||
const unresolved = records.filter((record) => record.status !== 'recovered');
|
||||
const recovered = records
|
||||
.filter((record) => record.status === 'recovered')
|
||||
.sort((a, b) => b.last_failed_at - a.last_failed_at)
|
||||
.slice(0, RECOVERED_KEEP);
|
||||
const keep = new Set([ ...unresolved, ...recovered ].map(eventKeyLike));
|
||||
const keptEvents = events.filter((event) => keep.has(eventKey(event)));
|
||||
const content = keptEvents.map((event) => JSON.stringify(event)).join('\n');
|
||||
const target = failuresFilePath(env);
|
||||
|
||||
await ensureStateDir(env);
|
||||
const temp = `${target}.${process.pid}.${lock.token}.tmp`;
|
||||
await writeFile(temp, keptEvents.length > 0 ? `${content}\n` : '', 'utf8');
|
||||
await rename(temp, target);
|
||||
|
||||
return {
|
||||
kept: keep.size,
|
||||
dropped: Math.max(0, records.length - keep.size),
|
||||
};
|
||||
} finally {
|
||||
await lock.release();
|
||||
}
|
||||
}
|
||||
588
src/lib/retry.test.ts
Normal file
588
src/lib/retry.test.ts
Normal file
|
|
@ -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<string, string>) => Promise<void>): Promise<void> {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'image-previews-retry-'));
|
||||
|
||||
try {
|
||||
await run({ IMAGE_PREVIEWS_STATE_DIR: dir });
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
interface DepsHarness {
|
||||
deps: RetryDeps;
|
||||
getAssetCalls: Array<{ id: string; transformation: Record<string, unknown> }>;
|
||||
readOneCalls: number;
|
||||
warnings: Array<Record<string, unknown>>;
|
||||
infos: Array<Record<string, unknown>>;
|
||||
}
|
||||
|
||||
function makeDeps(
|
||||
env: Record<string, string>,
|
||||
files: Record<string, { id: string; type: string }>,
|
||||
options: { fail?: boolean } = {},
|
||||
): DepsHarness {
|
||||
const getAssetCalls: Array<{ id: string; transformation: Record<string, unknown> }> = [];
|
||||
const warnings: Array<Record<string, unknown>> = [];
|
||||
const infos: Array<Record<string, unknown>> = [];
|
||||
const harness = {
|
||||
getAssetCalls,
|
||||
readOneCalls: 0,
|
||||
warnings,
|
||||
infos,
|
||||
} as DepsHarness;
|
||||
|
||||
class SettingsService {
|
||||
constructor(_options: Record<string, unknown>) {}
|
||||
|
||||
async readSingleton(): Promise<Record<string, unknown>> {
|
||||
return { storage_asset_presets: [] };
|
||||
}
|
||||
}
|
||||
|
||||
class ItemsService {
|
||||
constructor(_options: Record<string, unknown>) {}
|
||||
|
||||
async readOne(id: string): Promise<{ id: string; type: string }> {
|
||||
harness.readOneCalls += 1;
|
||||
const file = files[id];
|
||||
if (!file) throw new Error('not found');
|
||||
return file;
|
||||
}
|
||||
}
|
||||
|
||||
class AssetsService {
|
||||
constructor(_options: Record<string, unknown>) {}
|
||||
|
||||
async getAsset(id: string, transformation: Record<string, unknown>): Promise<{ stream: { destroy(): void } }> {
|
||||
getAssetCalls.push({ id, transformation });
|
||||
if (options.fail) throw new Error('boom');
|
||||
return { stream: { destroy() {} } };
|
||||
}
|
||||
}
|
||||
|
||||
harness.deps = {
|
||||
services: { SettingsService, ItemsService, AssetsService },
|
||||
database: null,
|
||||
getSchema: async () => ({}),
|
||||
env,
|
||||
logger: {
|
||||
warn: (payload: Record<string, unknown>) => warnings.push(payload),
|
||||
info: (payload: Record<string, unknown>) => infos.push(payload),
|
||||
error: () => {},
|
||||
},
|
||||
};
|
||||
|
||||
return harness;
|
||||
}
|
||||
|
||||
function failedRecord(overrides: Partial<FailureRecord> & Pick<FailureRecord, 'file_id'>): FailureRecord {
|
||||
return {
|
||||
file_id: overrides.file_id,
|
||||
preset_key: overrides.preset_key ?? 'system-medium-cover',
|
||||
format: overrides.format === undefined ? 'avif' : overrides.format,
|
||||
attempts: overrides.attempts ?? 1,
|
||||
status: overrides.status ?? 'failed',
|
||||
first_failed_at: overrides.first_failed_at ?? 1000,
|
||||
last_failed_at: overrides.last_failed_at ?? 1000,
|
||||
next_retry_at: overrides.next_retry_at === undefined ? 2000 : overrides.next_retry_at,
|
||||
error: overrides.error ?? null,
|
||||
ms: overrides.ms ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
test('nextRetryAt uses 5 then 15 minute backoff and stops after 3 attempts', () => {
|
||||
const now = 1_000_000;
|
||||
|
||||
assert.equal(nextRetryAt(1, now), now + 5 * 60 * 1000);
|
||||
assert.equal(nextRetryAt(2, now), now + 15 * 60 * 1000);
|
||||
assert.equal(nextRetryAt(3, now), null);
|
||||
assert.equal(nextRetryAt(4, now), null);
|
||||
assert.equal(nextRetryAt(0, now), now + 5 * 60 * 1000);
|
||||
assert.equal(MAX_ATTEMPTS, 3);
|
||||
});
|
||||
|
||||
test('dueRetries selects only retryable failures whose next retry is due', () => {
|
||||
const now = 10_000;
|
||||
const records = [
|
||||
failedRecord({ file_id: 'due', next_retry_at: now - 1 }),
|
||||
failedRecord({ file_id: 'future', next_retry_at: now + 1 }),
|
||||
failedRecord({ file_id: 'exhausted', attempts: 3, next_retry_at: now - 1 }),
|
||||
failedRecord({ file_id: 'no-schedule', next_retry_at: null }),
|
||||
failedRecord({ file_id: 'recovered', status: 'recovered', next_retry_at: null }),
|
||||
];
|
||||
|
||||
assert.deepEqual(
|
||||
dueRetries(records, now).map((record) => record.file_id),
|
||||
[ 'due' ],
|
||||
);
|
||||
});
|
||||
|
||||
test('runDueRetries recovers a due pair and appends completion + recovered event', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const now = Date.now();
|
||||
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts: now - 1000,
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
attempt: 1,
|
||||
next_retry_at: now - 1,
|
||||
error: 'boom',
|
||||
ms: 10,
|
||||
},
|
||||
env,
|
||||
);
|
||||
|
||||
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
|
||||
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
|
||||
|
||||
assert.equal(outcome.status, 'completed');
|
||||
assert.equal(outcome.processed, 1);
|
||||
assert.equal(outcome.recovered, 1);
|
||||
assert.equal(outcome.failed, 0);
|
||||
assert.equal(harness.getAssetCalls.length, 1);
|
||||
|
||||
const transformation = harness.getAssetCalls[0]?.transformation;
|
||||
assert.equal((transformation?.['transformationParams'] as { key?: string } | undefined)?.key, 'system-medium-cover');
|
||||
assert.equal(transformation?.['acceptFormat'], 'avif');
|
||||
|
||||
const completed = await loadCompleted(env);
|
||||
assert.ok(completed.has(completedKeyOf('f1', 'system-medium-cover', 'avif')));
|
||||
|
||||
const records = await replayFailures(env);
|
||||
const record = records.find((candidate) => candidate.file_id === 'f1');
|
||||
assert.equal(record?.status, 'recovered');
|
||||
assert.equal(record?.attempts, 2);
|
||||
assert.equal(record?.next_retry_at, null);
|
||||
assert.equal(await isLocked(env), false);
|
||||
});
|
||||
});
|
||||
|
||||
test('runDueRetries bumps the attempt and schedules 15 minutes on failure', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const now = Date.now();
|
||||
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts: now - 1000,
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
attempt: 1,
|
||||
next_retry_at: now - 1,
|
||||
error: 'first',
|
||||
ms: 10,
|
||||
},
|
||||
env,
|
||||
);
|
||||
|
||||
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } }, { fail: true });
|
||||
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
|
||||
|
||||
assert.equal(outcome.status, 'completed');
|
||||
assert.equal(outcome.processed, 1);
|
||||
assert.equal(outcome.recovered, 0);
|
||||
assert.equal(outcome.failed, 1);
|
||||
|
||||
const records = await replayFailures(env);
|
||||
const record = records.find((candidate) => candidate.file_id === 'f1');
|
||||
assert.equal(record?.status, 'failed');
|
||||
assert.equal(record?.attempts, 2);
|
||||
assert.equal(record?.error, 'boom');
|
||||
assert.ok(record?.next_retry_at !== null);
|
||||
assert.ok((record?.next_retry_at ?? 0) >= now + 14 * 60 * 1000);
|
||||
assert.ok((record?.next_retry_at ?? 0) <= now + 16 * 60 * 1000);
|
||||
|
||||
const completed = await loadCompleted(env);
|
||||
assert.equal(completed.size, 0);
|
||||
|
||||
const warning = harness.warnings.find((entry) => entry['fileId'] === 'f1');
|
||||
assert.equal(warning?.['preset_key'], 'system-medium-cover');
|
||||
assert.equal(warning?.['format'], 'avif');
|
||||
assert.equal(warning?.['attempt'], 2);
|
||||
assert.equal(warning?.['error'], 'boom');
|
||||
assert.equal(warning?.['next_retry_at'], record?.next_retry_at);
|
||||
assert.equal(await isLocked(env), false);
|
||||
});
|
||||
});
|
||||
|
||||
test('runDueRetries is bounded and groups pairs by file', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const now = Date.now();
|
||||
|
||||
for (let index = 0; index < 5; index += 1) {
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts: now - 1000,
|
||||
file_id: `f${index}`,
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
attempt: 1,
|
||||
next_retry_at: now - 1,
|
||||
error: 'boom',
|
||||
},
|
||||
env,
|
||||
);
|
||||
}
|
||||
|
||||
const harness = makeDeps(
|
||||
env,
|
||||
{
|
||||
f0: { id: 'f0', type: 'image/jpeg' },
|
||||
f1: { id: 'f1', type: 'image/jpeg' },
|
||||
f2: { id: 'f2', type: 'image/jpeg' },
|
||||
f3: { id: 'f3', type: 'image/jpeg' },
|
||||
f4: { id: 'f4', type: 'image/jpeg' },
|
||||
},
|
||||
);
|
||||
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true, limit: 2 });
|
||||
|
||||
assert.equal(outcome.processed, 2);
|
||||
assert.equal(outcome.recovered, 2);
|
||||
assert.equal(harness.getAssetCalls.length, 2);
|
||||
assert.equal((await replayFailures(env)).filter((record) => record.status === 'recovered').length, 2);
|
||||
});
|
||||
});
|
||||
|
||||
test('runDueRetries reads each file once for multiple due formats', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const now = Date.now();
|
||||
|
||||
for (const format of [ 'avif', 'webp' ]) {
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts: now - 1000,
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format,
|
||||
attempt: 1,
|
||||
next_retry_at: now - 1,
|
||||
error: 'boom',
|
||||
},
|
||||
env,
|
||||
);
|
||||
}
|
||||
|
||||
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
|
||||
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
|
||||
|
||||
assert.equal(outcome.processed, 2);
|
||||
assert.equal(harness.readOneCalls, 1);
|
||||
assert.equal(harness.getAssetCalls.length, 2);
|
||||
});
|
||||
});
|
||||
|
||||
test('runDueRetries marks a missing preset non-retryable without calling assets', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const now = Date.now();
|
||||
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts: now - 1000,
|
||||
file_id: 'f1',
|
||||
preset_key: 'gone',
|
||||
format: 'avif',
|
||||
attempt: 1,
|
||||
next_retry_at: now - 1,
|
||||
error: 'boom',
|
||||
},
|
||||
env,
|
||||
);
|
||||
|
||||
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
|
||||
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
|
||||
|
||||
assert.equal(outcome.processed, 1);
|
||||
assert.equal(outcome.failed, 1);
|
||||
assert.equal(harness.getAssetCalls.length, 0);
|
||||
|
||||
const record = (await replayFailures(env)).find((candidate) => candidate.file_id === 'f1');
|
||||
assert.equal(record?.status, 'failed');
|
||||
assert.equal(record?.attempts, MAX_ATTEMPTS);
|
||||
assert.equal(record?.next_retry_at, null);
|
||||
assert.match(record?.error ?? '', /unknown preset/);
|
||||
});
|
||||
});
|
||||
|
||||
test('runDueRetries is skipped when disabled', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
await writeSettings({ ...envDefaults(env), enabled: false }, env);
|
||||
|
||||
const harness = makeDeps(env, {});
|
||||
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
|
||||
|
||||
assert.equal(outcome.status, 'skipped');
|
||||
assert.equal(outcome.reason, 'disabled');
|
||||
assert.equal(outcome.processed, 0);
|
||||
});
|
||||
});
|
||||
|
||||
test('runDueRetries does not run or write while run.lock is held', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const now = Date.now();
|
||||
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts: now - 1000,
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
attempt: 1,
|
||||
next_retry_at: now - 1,
|
||||
error: 'boom',
|
||||
},
|
||||
env,
|
||||
);
|
||||
|
||||
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
|
||||
assert.equal(await acquireLock(env), true);
|
||||
const before = await readFile(failuresFilePath(env), 'utf8');
|
||||
|
||||
try {
|
||||
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
|
||||
assert.equal(outcome.status, 'skipped');
|
||||
assert.equal(outcome.reason, 'already-running');
|
||||
assert.equal(outcome.processed, 0);
|
||||
assert.equal(harness.getAssetCalls.length, 0);
|
||||
assert.equal(await readFile(failuresFilePath(env), 'utf8'), before);
|
||||
assert.equal((await loadCompleted(env)).size, 0);
|
||||
} finally {
|
||||
await releaseLock(env);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
test('runDueRetries is skipped outside the maintenance window', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const now = new Date();
|
||||
const current = now.getHours() * 60 + now.getMinutes();
|
||||
const start = (current + 60) % 1440;
|
||||
const end = (start + 1) % 1440;
|
||||
const time = (minutes: number): string =>
|
||||
`${String(Math.floor(minutes / 60)).padStart(2, '0')}:${String(minutes % 60).padStart(2, '0')}`;
|
||||
|
||||
await writeSettings(
|
||||
{ ...envDefaults(env), windowStart: time(start), windowEnd: time(end) },
|
||||
env,
|
||||
);
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts: Date.now() - 1000,
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
attempt: 1,
|
||||
next_retry_at: Date.now() - 1,
|
||||
error: 'boom',
|
||||
},
|
||||
env,
|
||||
);
|
||||
|
||||
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
|
||||
const outcome = await runDueRetries(harness.deps, {});
|
||||
|
||||
assert.equal(outcome.status, 'skipped');
|
||||
assert.equal(outcome.reason, 'outside-window');
|
||||
assert.equal(harness.getAssetCalls.length, 0);
|
||||
assert.equal((await loadCompleted(env)).size, 0);
|
||||
});
|
||||
});
|
||||
|
||||
test('fail -> recover -> fail starts a new retry cycle', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const now = Date.now();
|
||||
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts: now - 10_000,
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
attempt: 1,
|
||||
next_retry_at: now - 9_000,
|
||||
error: 'first',
|
||||
},
|
||||
env,
|
||||
);
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'recovered',
|
||||
ts: now - 8_000,
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
attempt: 2,
|
||||
},
|
||||
env,
|
||||
);
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts: now - 1_000,
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
attempt: 1,
|
||||
next_retry_at: now - 1,
|
||||
error: 'again',
|
||||
},
|
||||
env,
|
||||
);
|
||||
|
||||
const due = dueRetries(await replayFailures(env), now);
|
||||
assert.equal(due.length, 1);
|
||||
assert.equal(due[0]?.attempts, 1);
|
||||
|
||||
const harness = makeDeps(env, { f1: { id: 'f1', type: 'image/jpeg' } });
|
||||
const outcome = await runDueRetries(harness.deps, { ignoreWindow: true });
|
||||
|
||||
assert.equal(outcome.processed, 1);
|
||||
assert.equal(outcome.recovered, 1);
|
||||
|
||||
const record = (await replayFailures(env)).find((candidate) => candidate.file_id === 'f1');
|
||||
assert.equal(record?.status, 'recovered');
|
||||
assert.equal(record?.attempts, 2);
|
||||
});
|
||||
});
|
||||
|
||||
test('journalFailure never writes an attempt above MAX_ATTEMPTS', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const now = Date.now();
|
||||
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts: now - 1000,
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
attempt: MAX_ATTEMPTS,
|
||||
next_retry_at: null,
|
||||
error: 'terminal',
|
||||
},
|
||||
env,
|
||||
);
|
||||
|
||||
const record = await journalFailure(env, {
|
||||
file_id: 'f1',
|
||||
preset_key: 'system-medium-cover',
|
||||
format: 'avif',
|
||||
error: 'again',
|
||||
});
|
||||
|
||||
assert.equal(record?.attempts, MAX_ATTEMPTS);
|
||||
assert.equal(record?.next_retry_at, null);
|
||||
assert.equal((await replayFailures(env))[0]?.attempts, MAX_ATTEMPTS);
|
||||
});
|
||||
});
|
||||
|
||||
test('planFileTasks skips unresolved failures and warms recovered ones', () => {
|
||||
const presets: Preset[] = [ { key: 'system-medium-cover', format: 'auto' } ];
|
||||
const key = completedKeyOf('f1', 'system-medium-cover', 'avif');
|
||||
const failed = new Map<string, FailureRecord>([
|
||||
[ key, failedRecord({ file_id: 'f1', attempts: MAX_ATTEMPTS, next_retry_at: null }) ],
|
||||
]);
|
||||
const blocked = planFileTasks('f1', presets, [ 'avif' ], new Set(), failed);
|
||||
|
||||
assert.equal(blocked.tasks.length, 0);
|
||||
assert.equal(blocked.skipped, 1);
|
||||
assert.equal(blocked.completed, 0);
|
||||
|
||||
const recovered = new Map<string, FailureRecord>([
|
||||
[ key, failedRecord({ file_id: 'f1', status: 'recovered', next_retry_at: null }) ],
|
||||
]);
|
||||
const allowed = planFileTasks('f1', presets, [ 'avif' ], new Set(), recovered);
|
||||
|
||||
assert.equal(allowed.tasks.length, 1);
|
||||
assert.equal(allowed.skipped, 0);
|
||||
});
|
||||
|
||||
test('planFileTasks keeps totals stable, resumes state.jsonl, and skips unresolved pairs', async () => {
|
||||
await withTempEnv(async (env) => {
|
||||
const presets: Preset[] = [
|
||||
{ key: 'system-medium-cover', format: 'auto' },
|
||||
{ key: 'custom-fixed', format: 'webp' },
|
||||
];
|
||||
const formats = [ 'avif', 'webp' ];
|
||||
const total = tasksPerFile(presets, formats);
|
||||
|
||||
await appendCompleted(
|
||||
{ file_id: 'f1', preset_key: 'system-medium-cover', format: 'avif', ts: 1 },
|
||||
env,
|
||||
);
|
||||
const completed = await loadCompleted(env);
|
||||
const failures = new Map<string, FailureRecord>([
|
||||
[
|
||||
completedKeyOf('f1', 'system-medium-cover', 'webp'),
|
||||
failedRecord({ file_id: 'f1', format: 'webp', attempts: 1, next_retry_at: Date.now() + 1000 }),
|
||||
],
|
||||
[
|
||||
completedKeyOf('f1', 'custom-fixed', 'webp'),
|
||||
failedRecord({
|
||||
file_id: 'f1',
|
||||
preset_key: 'custom-fixed',
|
||||
format: 'webp',
|
||||
attempts: MAX_ATTEMPTS,
|
||||
next_retry_at: null,
|
||||
}),
|
||||
],
|
||||
]);
|
||||
|
||||
const plan = planFileTasks('f1', presets, formats, completed, failures);
|
||||
|
||||
assert.equal(total, 3);
|
||||
assert.equal(plan.completed, 1);
|
||||
assert.equal(plan.skipped, 2);
|
||||
assert.equal(plan.tasks.length, 0);
|
||||
assert.equal(plan.completed + plan.skipped + plan.tasks.length, total);
|
||||
|
||||
const retryableOnly = new Map<string, FailureRecord>([
|
||||
[
|
||||
completedKeyOf('f1', 'custom-fixed', 'webp'),
|
||||
failedRecord({
|
||||
file_id: 'f1',
|
||||
preset_key: 'custom-fixed',
|
||||
format: 'webp',
|
||||
attempts: MAX_ATTEMPTS,
|
||||
next_retry_at: null,
|
||||
}),
|
||||
],
|
||||
]);
|
||||
const mixed = planFileTasks('f1', presets, formats, completed, retryableOnly);
|
||||
|
||||
assert.equal(mixed.completed, 1);
|
||||
assert.equal(mixed.skipped, 1);
|
||||
assert.equal(mixed.tasks.length, 1);
|
||||
assert.equal(mixed.completed + mixed.tasks.length + mixed.skipped, total);
|
||||
});
|
||||
});
|
||||
417
src/lib/retry.ts
Normal file
417
src/lib/retry.ts
Normal file
|
|
@ -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<string, FailureRecord>;
|
||||
logger?: any;
|
||||
}
|
||||
|
||||
export async function journalFailure(
|
||||
env: Record<string, any>,
|
||||
input: FailureInput,
|
||||
options: JournalOptions = {},
|
||||
): Promise<FailureRecord | null> {
|
||||
try {
|
||||
const records = options.records ?? (await replayFailureMap(env));
|
||||
const key = completedKeyOf(input.file_id, input.preset_key, input.format);
|
||||
const previous = records.get(key);
|
||||
const ts = input.ts ?? Date.now();
|
||||
const attempt =
|
||||
previous && previous.status === 'failed' ? Math.min(MAX_ATTEMPTS, previous.attempts + 1) : 1;
|
||||
const next = nextRetryAt(attempt, ts);
|
||||
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts,
|
||||
file_id: input.file_id,
|
||||
preset_key: input.preset_key,
|
||||
format: input.format,
|
||||
attempt,
|
||||
next_retry_at: next,
|
||||
error: input.error,
|
||||
ms: input.ms,
|
||||
},
|
||||
env,
|
||||
{ logger: options.logger },
|
||||
);
|
||||
|
||||
const record: FailureRecord = {
|
||||
file_id: input.file_id,
|
||||
preset_key: input.preset_key,
|
||||
format: input.format,
|
||||
attempts: attempt,
|
||||
status: 'failed',
|
||||
first_failed_at: previous && previous.status === 'failed' ? previous.first_failed_at : ts,
|
||||
last_failed_at: ts,
|
||||
next_retry_at: next,
|
||||
error: input.error ?? null,
|
||||
ms: input.ms ?? null,
|
||||
};
|
||||
|
||||
records.set(key, record);
|
||||
options.logger?.warn?.(
|
||||
{
|
||||
fileId: input.file_id,
|
||||
preset_key: input.preset_key,
|
||||
format: input.format,
|
||||
attempt,
|
||||
next_retry_at: next,
|
||||
error: input.error,
|
||||
},
|
||||
'image-previews: preview failed',
|
||||
);
|
||||
|
||||
return record;
|
||||
} catch (error) {
|
||||
options.logger?.error?.(error, 'image-previews: could not append failure journal');
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export async function markRecovered(
|
||||
env: Record<string, any>,
|
||||
input: FailureInput,
|
||||
options: JournalOptions = {},
|
||||
): Promise<void> {
|
||||
try {
|
||||
const records = options.records ?? (await replayFailureMap(env));
|
||||
const key = completedKeyOf(input.file_id, input.preset_key, input.format);
|
||||
const previous = records.get(key);
|
||||
|
||||
if (!previous || previous.status === 'recovered') return;
|
||||
|
||||
const ts = input.ts ?? Date.now();
|
||||
const attempt = Math.min(MAX_ATTEMPTS, previous.attempts + 1);
|
||||
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'recovered',
|
||||
ts,
|
||||
file_id: input.file_id,
|
||||
preset_key: input.preset_key,
|
||||
format: input.format,
|
||||
attempt,
|
||||
next_retry_at: null,
|
||||
ms: input.ms,
|
||||
},
|
||||
env,
|
||||
{ logger: options.logger },
|
||||
);
|
||||
|
||||
records.set(key, {
|
||||
...previous,
|
||||
attempts: attempt,
|
||||
status: 'recovered',
|
||||
next_retry_at: null,
|
||||
ms: input.ms ?? previous.ms,
|
||||
});
|
||||
|
||||
options.logger?.info?.(
|
||||
{
|
||||
fileId: input.file_id,
|
||||
preset_key: input.preset_key,
|
||||
format: input.format,
|
||||
attempt,
|
||||
ms: input.ms,
|
||||
},
|
||||
'image-previews: preview recovered',
|
||||
);
|
||||
} catch (error) {
|
||||
options.logger?.error?.(error, 'image-previews: could not append recovery journal');
|
||||
}
|
||||
}
|
||||
|
||||
export interface RetryDeps {
|
||||
services: Record<string, any>;
|
||||
database: any;
|
||||
getSchema: () => Promise<any>;
|
||||
env: Record<string, any>;
|
||||
logger: any;
|
||||
}
|
||||
|
||||
export interface RetryTarget {
|
||||
file_id: string;
|
||||
preset_key: string;
|
||||
format: string | null;
|
||||
}
|
||||
|
||||
export interface RetryOptions {
|
||||
now?: number;
|
||||
ignoreWindow?: boolean;
|
||||
limit?: number;
|
||||
targets?: RetryTarget[];
|
||||
}
|
||||
|
||||
export interface RetryOutcome {
|
||||
status: 'completed' | 'skipped' | 'failed';
|
||||
reason?: string;
|
||||
processed: number;
|
||||
recovered: number;
|
||||
failed: number;
|
||||
}
|
||||
|
||||
async function markNonRetryable(
|
||||
env: Record<string, any>,
|
||||
record: FailureRecord,
|
||||
error: string,
|
||||
records: Map<string, FailureRecord>,
|
||||
logger: any,
|
||||
): Promise<void> {
|
||||
const ts = Date.now();
|
||||
|
||||
try {
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'retry',
|
||||
ts,
|
||||
file_id: record.file_id,
|
||||
preset_key: record.preset_key,
|
||||
format: record.format,
|
||||
attempt: MAX_ATTEMPTS,
|
||||
next_retry_at: null,
|
||||
error,
|
||||
},
|
||||
env,
|
||||
{ logger },
|
||||
);
|
||||
} catch (journalError) {
|
||||
logger?.error?.(journalError, 'image-previews: could not append retry journal');
|
||||
}
|
||||
|
||||
records.set(completedKeyOf(record.file_id, record.preset_key, record.format), {
|
||||
...record,
|
||||
attempts: MAX_ATTEMPTS,
|
||||
status: 'failed',
|
||||
next_retry_at: null,
|
||||
error,
|
||||
});
|
||||
|
||||
logger?.warn?.(
|
||||
{
|
||||
fileId: record.file_id,
|
||||
preset_key: record.preset_key,
|
||||
format: record.format,
|
||||
attempt: MAX_ATTEMPTS,
|
||||
next_retry_at: null,
|
||||
error,
|
||||
},
|
||||
'image-previews: retry skipped',
|
||||
);
|
||||
}
|
||||
|
||||
export async function runDueRetries(deps: RetryDeps, options: RetryOptions = {}): Promise<RetryOutcome> {
|
||||
const outcome: RetryOutcome = { status: 'skipped', processed: 0, recovered: 0, failed: 0 };
|
||||
|
||||
try {
|
||||
const settings = await readSettings(deps.env);
|
||||
|
||||
if (!settings.enabled) return { ...outcome, reason: 'disabled' };
|
||||
if (!options.ignoreWindow && !isInWindow(settings.windowStart, settings.windowEnd)) {
|
||||
return { ...outcome, reason: 'outside-window' };
|
||||
}
|
||||
|
||||
const run = await readRun(deps.env);
|
||||
if (run?.running) return { ...outcome, reason: 'already-running' };
|
||||
if (!(await acquireLock(deps.env))) return { ...outcome, reason: 'already-running' };
|
||||
|
||||
try {
|
||||
const now = options.now ?? Date.now();
|
||||
const schema = await deps.getSchema();
|
||||
const custom = await loadCustomPresets(new deps.services['SettingsService']({ schema }));
|
||||
const presets = buildRegistry(custom, null);
|
||||
const presetByKey = new Map<string, Preset>(presets.map((preset) => [ preset.key, preset ]));
|
||||
const records = await replayFailureMap(deps.env);
|
||||
const targets = options.targets
|
||||
? new Set(options.targets.map((target) => completedKeyOf(target.file_id, target.preset_key, target.format)))
|
||||
: null;
|
||||
const limit = Math.max(1, Math.min(200, Math.trunc(options.limit ?? DEFAULT_RETRY_BATCH)));
|
||||
const candidates = dueRetries([ ...records.values() ], now)
|
||||
.filter((record) => !targets || targets.has(completedKeyOf(record.file_id, record.preset_key, record.format)))
|
||||
.sort(
|
||||
(a, b) =>
|
||||
(a.next_retry_at ?? 0) - (b.next_retry_at ?? 0) ||
|
||||
a.last_failed_at - b.last_failed_at ||
|
||||
completedKeyOf(a.file_id, a.preset_key, a.format).localeCompare(
|
||||
completedKeyOf(b.file_id, b.preset_key, b.format),
|
||||
),
|
||||
)
|
||||
.slice(0, limit);
|
||||
const grouped = new Map<string, FailureRecord[]>();
|
||||
|
||||
for (const record of candidates) {
|
||||
const group = grouped.get(record.file_id);
|
||||
if (group) group.push(record);
|
||||
else grouped.set(record.file_id, [ record ]);
|
||||
}
|
||||
|
||||
const items = new deps.services['ItemsService']('directus_files', { schema });
|
||||
const assets = new deps.services['AssetsService']({ schema });
|
||||
|
||||
for (const [ fileId, group ] of grouped) {
|
||||
let file: { id?: string; type?: string | null } | null = null;
|
||||
|
||||
try {
|
||||
file = await items.readOne(fileId, { fields: [ 'id', 'type' ] });
|
||||
} catch {
|
||||
file = null;
|
||||
}
|
||||
|
||||
if (!file || !isImageType(file.type)) {
|
||||
for (const record of group) {
|
||||
await markNonRetryable(deps.env, record, 'file not found or not an image', records, deps.logger);
|
||||
outcome.processed += 1;
|
||||
outcome.failed += 1;
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const record of group) {
|
||||
const preset = presetByKey.get(record.preset_key);
|
||||
|
||||
if (!preset) {
|
||||
await markNonRetryable(deps.env, record, `unknown preset: ${record.preset_key}`, records, deps.logger);
|
||||
outcome.processed += 1;
|
||||
outcome.failed += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
const variant = preset.format === 'auto' ? record.format : null;
|
||||
|
||||
if (preset.format === 'auto' && !variant) {
|
||||
await markNonRetryable(deps.env, record, 'missing format', records, deps.logger);
|
||||
outcome.processed += 1;
|
||||
outcome.failed += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
const result = await warmVariant(assets, fileId, preset, variant);
|
||||
const ts = Date.now();
|
||||
|
||||
recordOp({
|
||||
ts,
|
||||
fileId,
|
||||
key: result.key,
|
||||
format: result.format,
|
||||
ok: result.ok,
|
||||
ms: result.ms,
|
||||
error: result.error,
|
||||
});
|
||||
|
||||
if (result.ok) {
|
||||
await appendCompleted(
|
||||
{ file_id: fileId, preset_key: record.preset_key, format: result.format, ts },
|
||||
deps.env,
|
||||
);
|
||||
await markRecovered(
|
||||
deps.env,
|
||||
{
|
||||
file_id: fileId,
|
||||
preset_key: record.preset_key,
|
||||
format: result.format,
|
||||
ms: result.ms,
|
||||
ts,
|
||||
},
|
||||
{ records, logger: deps.logger },
|
||||
);
|
||||
outcome.recovered += 1;
|
||||
} else {
|
||||
const attempt = Math.min(MAX_ATTEMPTS, record.attempts + 1);
|
||||
const next = nextRetryAt(attempt, ts);
|
||||
|
||||
try {
|
||||
await appendFailure(
|
||||
{
|
||||
type: 'failed',
|
||||
ts,
|
||||
file_id: fileId,
|
||||
preset_key: record.preset_key,
|
||||
format: result.format,
|
||||
attempt,
|
||||
next_retry_at: next,
|
||||
error: result.error,
|
||||
ms: result.ms,
|
||||
},
|
||||
deps.env,
|
||||
{ logger: deps.logger },
|
||||
);
|
||||
} catch (journalError) {
|
||||
deps.logger?.error?.(journalError, 'image-previews: could not append retry journal');
|
||||
}
|
||||
|
||||
records.set(completedKeyOf(fileId, record.preset_key, record.format), {
|
||||
...record,
|
||||
attempts: attempt,
|
||||
status: 'failed',
|
||||
last_failed_at: ts,
|
||||
next_retry_at: next,
|
||||
error: result.error ?? null,
|
||||
ms: result.ms,
|
||||
});
|
||||
|
||||
deps.logger?.warn?.(
|
||||
{
|
||||
fileId,
|
||||
preset_key: record.preset_key,
|
||||
format: result.format,
|
||||
attempt,
|
||||
next_retry_at: next,
|
||||
error: result.error,
|
||||
},
|
||||
'image-previews: retry failed',
|
||||
);
|
||||
|
||||
outcome.failed += 1;
|
||||
}
|
||||
|
||||
outcome.processed += 1;
|
||||
}
|
||||
}
|
||||
|
||||
await compactFailures(deps.env);
|
||||
return { ...outcome, status: 'completed' };
|
||||
} finally {
|
||||
await releaseLock(deps.env);
|
||||
}
|
||||
} catch (error) {
|
||||
deps.logger?.error?.(error, 'image-previews: retry run failed');
|
||||
return { ...outcome, status: 'failed', reason: errorMessage(error) };
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<boolean> } = {},
|
||||
): Promise<WarmResultItem> {
|
||||
const transformation: Record<string, unknown> = { transformationParams: preset };
|
||||
|
||||
if (format) transformation['acceptFormat'] = format;
|
||||
|
||||
const started = Date.now();
|
||||
const outputFormat = effectiveFormat(preset, format);
|
||||
const cached = options.isCached ? await options.isCached(preset, format) : false;
|
||||
|
||||
if (cached) {
|
||||
return { key: preset.key, format: outputFormat, ok: true, ms: Date.now() - started, cached: true };
|
||||
}
|
||||
|
||||
try {
|
||||
const asset = await assets.getAsset(fileId, transformation);
|
||||
releaseStream(asset?.stream);
|
||||
return { key: preset.key, format: outputFormat, ok: true, ms: Date.now() - started, cached: false };
|
||||
} catch (error) {
|
||||
return { key: preset.key, format: outputFormat, ok: false, ms: Date.now() - started, error: errorMessage(error) };
|
||||
}
|
||||
}
|
||||
|
||||
export async function warmFile(
|
||||
assets: AssetsLike,
|
||||
fileId: string,
|
||||
presets: Preset[],
|
||||
options: WarmOptions,
|
||||
): Promise<WarmSummary> {
|
||||
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<WarmSummary> {
|
||||
const results: WarmResultItem[] = new Array(tasks.length);
|
||||
let cursor = 0;
|
||||
|
||||
|
|
|
|||
|
|
@ -195,6 +195,34 @@
|
|||
</table>
|
||||
</section>
|
||||
|
||||
<section class="ip__card">
|
||||
<header class="ip__head">
|
||||
<h2>Failed previews</h2>
|
||||
<v-button @click="loadFailures">Refresh</v-button>
|
||||
</header>
|
||||
<p class="ip__hint">
|
||||
Persisted failure journal. Retries run automatically inside the maintenance window with no
|
||||
active backfill: up to 3 attempts, then 5 and 15 minute backoff.
|
||||
</p>
|
||||
<table class="ip__table">
|
||||
<thead>
|
||||
<tr><th>file</th><th>preset</th><th>format</th><th>attempts</th><th>last error</th><th>next retry</th><th>status</th></tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
<tr v-for="failure in failures" :key="failure.file_id + failure.preset_key + (failure.format ?? '')">
|
||||
<td class="ip__mono">{{ failure.file?.filename_download ?? failure.file_id.slice(0, 8) }}</td>
|
||||
<td>{{ failure.preset_key }}</td>
|
||||
<td>{{ failure.format ?? 'original' }}</td>
|
||||
<td>{{ failure.attempts }}</td>
|
||||
<td class="ip__error">{{ failure.error ?? '' }}</td>
|
||||
<td>{{ failure.next_retry_at_human ?? '—' }}</td>
|
||||
<td>{{ failure.status }}</td>
|
||||
</tr>
|
||||
</tbody>
|
||||
</table>
|
||||
<p v-if="failures.length === 0" class="ip__hint">No failures recorded.</p>
|
||||
</section>
|
||||
|
||||
<section class="ip__card">
|
||||
<header class="ip__head"><h2>API reference</h2></header>
|
||||
<p class="ip__hint">
|
||||
|
|
@ -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<Status | null>(null);
|
||||
const config = ref<Record<string, any> | null>(null);
|
||||
const disk = ref<{ previews: Bucket; originals: Bucket; total: Bucket } | null>(null);
|
||||
const ops = ref<any[]>([]);
|
||||
const failures = ref<FailureRecord[]>([]);
|
||||
const presets = ref<PresetInfo[]>([]);
|
||||
const shadowedKeys = ref<string[]>([]);
|
||||
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);
|
||||
});
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue