From 683e74b479afae2b918bb954800e3690401b4d1c Mon Sep 17 00:00:00 2001 From: Konrad Kollnig <5175206+kasnder@users.noreply.github.com> Date: Mon, 13 Jul 2026 15:32:03 +0200 Subject: [PATCH 1/4] Add migrations for state columns, jsonb/timestamptz, case-insensitive appids 007: add status/processing_started/failure_reason/failure_retryable columns to apps, backfilled from the analysis JSON state (NULL->queued, 'Processing in progress'->processing, success=false->failed, else analysed). Malformed processing timestamps fall back to NOW() instead of aborting the migration. Partial indexes support the queue scan. 008: modernise types (json->jsonb, timestamp->timestamptz assuming UTC). The pg driver treats these equivalently, so the currently deployed code keeps working during the deploy window. 009: enforce case-insensitive bundle IDs by merging rows that differ only by case (dedupe colliding app_analyses history before repointing to the keeper) and adding a unique index on lower(appid). Co-Authored-By: Claude Fable 5 --- migrations/007_apps_state_columns.sql | 69 ++++++++++++++++++++++ migrations/008_jsonb_timestamptz.sql | 14 +++++ migrations/009_case_insensitive_appids.sql | 55 +++++++++++++++++ 3 files changed, 138 insertions(+) create mode 100644 migrations/007_apps_state_columns.sql create mode 100644 migrations/008_jsonb_timestamptz.sql create mode 100644 migrations/009_case_insensitive_appids.sql diff --git a/migrations/007_apps_state_columns.sql b/migrations/007_apps_state_columns.sql new file mode 100644 index 0000000..ab1d495 --- /dev/null +++ b/migrations/007_apps_state_columns.sql @@ -0,0 +1,69 @@ +-- Move queue/analysis state out of the analysis JSON payload into real columns. +-- The analysis column keeps the raw analyser payload untouched (so the website's +-- failure display and the Raspberry Pi HTTP contract stay byte-identical); +-- scheduling state now lives in status/processing_started/failure_* columns. +-- +-- Additive and backward-compatible: the currently deployed code ignores these +-- columns entirely, and the defaults keep new rows valid until the new code boots. + +ALTER TABLE apps + ADD COLUMN status text NOT NULL DEFAULT 'queued' + CHECK (status IN ('queued', 'processing', 'analysed', 'failed')), + ADD COLUMN processing_started timestamptz, + ADD COLUMN failure_reason text, + ADD COLUMN failure_retryable boolean; + +-- Backfill from the JSON state previously embedded in analysis. +-- Order matters: the in-flight processing indicator also carries +-- success=false, so it must be classified before the failed branch. +-- +-- Timestamps are validated with a regex before casting so a malformed +-- analysis->>'timestamp' cannot abort the whole migration; unparseable +-- values fall back to NOW() so the processing lock still expires normally. +UPDATE apps +SET + status = CASE + WHEN analysis IS NULL THEN 'queued' + WHEN analysis->>'logs' = 'Processing in progress' THEN 'processing' + WHEN analysis->>'success' = 'false' THEN 'failed' + ELSE 'analysed' + END, + processing_started = CASE + WHEN analysis->>'logs' = 'Processing in progress' THEN + CASE + WHEN analysis->>'timestamp' ~ '^\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}' + THEN (analysis->>'timestamp')::timestamptz + ELSE NOW() + END + END, + failure_reason = CASE + WHEN COALESCE(analysis->>'logs', '') <> 'Processing in progress' + AND analysis->>'success' = 'false' + THEN analysis->>'reason' + END, + failure_retryable = CASE + WHEN COALESCE(analysis->>'logs', '') <> 'Processing in progress' + AND analysis->>'success' = 'false' + THEN COALESCE(analysis->>'retryable', 'true') <> 'false' + END; + +-- Indexes supporting the queue scan (models/Apps.js nextApp / countQueue). +-- A partial index per hot status keeps the candidate lookup cheap. +CREATE INDEX IF NOT EXISTS apps_status_idx + ON apps (status); + +CREATE INDEX IF NOT EXISTS apps_queued_added_idx + ON apps (added ASC) + WHERE status = 'queued'; + +CREATE INDEX IF NOT EXISTS apps_processing_started_idx + ON apps (processing_started) + WHERE status = 'processing'; + +CREATE INDEX IF NOT EXISTS apps_failed_retryable_idx + ON apps (status) + WHERE status = 'failed' AND failure_retryable; + +CREATE INDEX IF NOT EXISTS apps_analysed_version_idx + ON apps (analysisversion, analysed) + WHERE status = 'analysed'; diff --git a/migrations/008_jsonb_timestamptz.sql b/migrations/008_jsonb_timestamptz.sql new file mode 100644 index 0000000..d4cb373 --- /dev/null +++ b/migrations/008_jsonb_timestamptz.sql @@ -0,0 +1,14 @@ +-- Modernise column types: json -> jsonb and timestamp -> timestamptz. +-- Historical timestamps were written in UTC, so reinterpret them as UTC. +-- schema_migrations is deliberately left untouched. + +ALTER TABLE apps + ALTER COLUMN details TYPE jsonb USING details::jsonb, + ALTER COLUMN analysis TYPE jsonb USING analysis::jsonb, + ALTER COLUMN added TYPE timestamptz USING (added AT TIME ZONE 'UTC'), + ALTER COLUMN analysed TYPE timestamptz USING (analysed AT TIME ZONE 'UTC'); + +ALTER TABLE app_analyses + ALTER COLUMN analysis TYPE jsonb USING analysis::jsonb, + ALTER COLUMN analysed TYPE timestamptz USING (analysed AT TIME ZONE 'UTC'), + ALTER COLUMN app_store_updated TYPE timestamptz USING (app_store_updated AT TIME ZONE 'UTC'); diff --git a/migrations/009_case_insensitive_appids.sql b/migrations/009_case_insensitive_appids.sql new file mode 100644 index 0000000..d61dfee --- /dev/null +++ b/migrations/009_case_insensitive_appids.sql @@ -0,0 +1,55 @@ +-- The App Store treats bundle IDs case-insensitively; enforce that in the +-- database. Merge rows whose appid differs only by case, then add a unique +-- index on lower(appid). + +-- Rank rows within each case-insensitive duplicate group. The keeper (rn = 1) +-- prefers status = 'analysed', then the most recent analysis, then the +-- earliest-added row; appid is a final tie-breaker for determinism. +CREATE TEMP TABLE apps_case_dupes ON COMMIT DROP AS +SELECT + appid, + lower(appid) AS lower_appid, + row_number() OVER ( + PARTITION BY lower(appid) + ORDER BY + (status = 'analysed') DESC, + analysed DESC NULLS LAST, + added ASC, + appid ASC + ) AS rn +FROM apps +WHERE lower(appid) IN ( + SELECT lower(appid) + FROM apps + GROUP BY lower(appid) + HAVING count(*) > 1 +); + +-- Before repointing history rows to the keeper, drop rows that would collide +-- on the (appid, analysed) unique index once every appid in the group becomes +-- the keeper's: keep the lowest id per (group, analysed) pair. +DELETE FROM app_analyses a +USING apps_case_dupes ma, app_analyses b, apps_case_dupes mb +WHERE ma.appid = a.appid + AND mb.appid = b.appid + AND ma.lower_appid = mb.lower_appid + AND a.analysed = b.analysed + AND a.id > b.id; + +-- Repoint the losers' remaining history rows to the keeper. +UPDATE app_analyses a +SET appid = keeper.appid +FROM apps_case_dupes loser +JOIN apps_case_dupes keeper + ON keeper.lower_appid = loser.lower_appid AND keeper.rn = 1 +WHERE a.appid = loser.appid + AND loser.rn > 1; + +-- Drop the loser apps rows (their history has been moved to the keeper). +DELETE FROM apps a +USING apps_case_dupes loser +WHERE loser.rn > 1 + AND a.appid = loser.appid; + +CREATE UNIQUE INDEX IF NOT EXISTS apps_appid_lower_unique + ON apps (lower(appid)); From c2bca2160d552d0111430e77f6dd9ae6cd608203 Mon Sep 17 00:00:00 2001 From: Konrad Kollnig <5175206+kasnder@users.noreply.github.com> Date: Mon, 13 Jul 2026 15:32:13 +0200 Subject: [PATCH 2/4] Drive scheduling from status columns in models/Apps.js Replace the analysis-JSON state machine with the real status columns: - findApp: case-insensitive lookup (lower(appid) = lower($1)). - countQueue / countAnalysed / getAllApps / getSiteDataSignature / lastAnalysed: filter on status instead of analysis JSON. - nextApp: pick candidates by status (queued, expired processing lock, stale analysed, or retryable failed) and take a non-destructive lock (status='processing', processing_started=NOW()) without overwriting the analysis payload. History snapshots now happen only in updateAnalysis. - updateAnalysis: still writes the raw payload to apps.analysis (website failure display + Pi HTTP responses stay byte-identical) and additionally sets status/failure_reason/failure_retryable via the new pure deriveAnalysisState helper, which is unit-tested. The Raspberry Pi analyser HTTP contract (/queue, /uploadAnalysis, /reportAnalysisFailure, /ping) is unchanged. Co-Authored-By: Claude Fable 5 --- models/Apps.js | 132 ++++++++++++++++++++--------------------- test/appsState.test.js | 38 ++++++++++++ 2 files changed, 103 insertions(+), 67 deletions(-) create mode 100644 test/appsState.test.js diff --git a/models/Apps.js b/models/Apps.js index e155ea2..a07d357 100644 --- a/models/Apps.js +++ b/models/Apps.js @@ -14,8 +14,27 @@ pool.on('error', (err) => { console.error('Unexpected PostgreSQL pool error:', err.message); }); +// Derive the scheduling-state columns from an analyser payload. +// A payload with success === false is a failure; everything else is a +// successful analysis. Kept as a pure, exported helper so the mapping can be +// unit-tested without a database. +function deriveAnalysisState(analysis) { + if (analysis && analysis.success === false) { + return { + status: 'failed', + failureReason: (analysis.reason || analysis.logs) || null, + failureRetryable: analysis.retryable !== false + }; + } + return { + status: 'analysed', + failureReason: null, + failureRetryable: null + }; +} + const lastAnalysed = async () => { - const result = await pool.query('SELECT * FROM apps WHERE analysis IS NOT NULL ORDER BY analysed DESC LIMIT 5'); + const result = await pool.query("SELECT * FROM apps WHERE status = 'analysed' ORDER BY analysed DESC LIMIT 5"); return result.rows; } @@ -26,7 +45,7 @@ const healthCheck = async () => { const findApp = async (appId) => { if (!isValidAppId(appId)) return null; - const result = await pool.query('SELECT * FROM apps WHERE appid = $1', [appId]); + const result = await pool.query('SELECT * FROM apps WHERE lower(appid) = lower($1)', [appId]); if (result.rows.length == 0) return null; @@ -38,7 +57,7 @@ const countQueue = async (added) => { const result = await pool.query(` SELECT COUNT(*) FROM apps - WHERE analysis IS NULL + WHERE status = 'queued' AND added < $1 AND appid ~ $2 AND appid !~ '[.]$' @@ -49,7 +68,7 @@ const countQueue = async (added) => { const result = await pool.query(` SELECT COUNT(*) FROM apps - WHERE analysis IS NULL + WHERE status = 'queued' AND appid ~ $1 AND appid !~ '[.]$' AND length(appid) <= $2 @@ -76,71 +95,40 @@ const currentAnalysisVersion = parseInt(process.env.CURRENT_ANALYSIS_VERSION || const staleAnalysisDays = parseInt(process.env.STALE_ANALYSIS_DAYS || '180', 10); const processingTimeoutMinutes = parseInt(process.env.PROCESSING_TIMEOUT_MINUTES || '120', 10); -async function snapshotCurrentAnalysis(client, appId) { - await client.query(` - INSERT INTO app_analyses ( - appid, - analysis, - analysisversion, - analysed, - app_version, - app_store_updated, - analysis_source, - success - ) - SELECT - appid, - analysis, - analysisversion, - COALESCE(analysed, NOW()), - details->>'version', - NULLIF(details->>'updated', '')::timestamp, - COALESCE(analysis->>'analysis_source', 'legacy'), - COALESCE((analysis->>'success')::boolean, true) - FROM apps - WHERE appid = $1 - AND analysis IS NOT NULL - AND NOT EXISTS ( - SELECT 1 - FROM app_analyses existing - WHERE existing.appid = apps.appid - AND existing.analysed = COALESCE(apps.analysed, NOW()) - ) - ON CONFLICT (appid, analysed) DO NOTHING - `, [appId]); -} - const nextApp = async () => { - const processingIndicator = { - success: false, - logs: 'Processing in progress', - timestamp: new Date().toISOString() // ISO 8601 format - }; - const client = await pool.connect(); try { await client.query('BEGIN'); + // Candidate selection keys off the real status columns instead of the + // analysis JSON. The lock is non-destructive: we flip status to + // 'processing' and stamp processing_started, but leave the analysis + // payload alone so the last good result stays visible on the website + // while a refetch is in flight. History snapshots happen in + // updateAnalysis when the new result lands. const candidate = await client.query(` - SELECT appid, analysis + SELECT appid FROM apps WHERE appid ~ $4 AND appid !~ '[.]$' AND length(appid) <= $5 - AND COALESCE(analysis->>'retryable', 'true') <> 'false' AND ( - analysis IS NULL + status = 'queued' OR ( - analysis->>'logs' = 'Processing in progress' - AND (analysis->>'timestamp')::timestamptz < NOW() - ($3::int * INTERVAL '1 minute') + status = 'processing' + AND processing_started < NOW() - ($3::int * INTERVAL '1 minute') ) OR ( - coalesce(analysis->>'logs', '') <> 'Processing in progress' + status = 'analysed' AND ( analysisversion IS DISTINCT FROM $1 OR analysed < NOW() - ($2::int * INTERVAL '1 day') ) ) + OR ( + status = 'failed' + AND failure_retryable + ) ) ORDER BY ${popularityExpression} DESC, added ASC LIMIT 1 @@ -158,17 +146,13 @@ const nextApp = async () => { return null; } - const app = candidate.rows[0]; - if (app.analysis && app.analysis.logs !== 'Processing in progress') { - await snapshotCurrentAnalysis(client, app.appid); - } - const result = await client.query(` UPDATE apps - SET analysis = $1 - WHERE appid = $2 + SET status = 'processing', + processing_started = NOW() + WHERE appid = $1 RETURNING appid - `, [JSON.stringify(processingIndicator), app.appid]); + `, [candidate.rows[0].appid]); await client.query('COMMIT'); @@ -185,13 +169,28 @@ const nextApp = async () => { const updateAnalysis = async (appId, analysis, analysisVersion) => { if (!isValidAppId(appId)) throw new TypeError('Invalid App Store bundle ID'); + const { status, failureReason, failureRetryable } = deriveAnalysisState(analysis); + const client = await pool.connect(); try { await client.query('BEGIN'); + // Keep writing the raw payload to apps.analysis (the website failure + // display and existing HTTP responses depend on it) AND set the new + // scheduling columns. processing_started is cleared now that the lock + // is resolved. const result = await client.query( - 'UPDATE apps SET analysis = $1, analysisVersion = $2, analysed = NOW() WHERE appid = $3 RETURNING appid, details, analysed', - [analysis, analysisVersion, appId] + `UPDATE apps + SET analysis = $1, + analysisVersion = $2, + analysed = NOW(), + status = $4, + failure_reason = $5, + failure_retryable = $6, + processing_started = NULL + WHERE appid = $3 + RETURNING appid, details, analysed`, + [analysis, analysisVersion, appId, status, failureReason, failureRetryable] ); if (result.rowCount > 0) { @@ -241,7 +240,7 @@ const updateAnalysis = async (appId, analysis, analysisVersion) => { } const getAllApps = async () => { - const result = await pool.query('SELECT * FROM apps WHERE analysis IS NOT NULL'); + const result = await pool.query("SELECT * FROM apps WHERE status = 'analysed'"); return result.rows; } @@ -249,13 +248,11 @@ const getSiteDataSignature = async () => { const result = await pool.query(` SELECT COUNT(*) FILTER ( - WHERE analysis IS NOT NULL - AND COALESCE(analysis->>'success', 'true') != 'false' + WHERE status = 'analysed' AND analysis->'trackers' IS NOT NULL ) AS app_count, MAX(analysed) FILTER ( - WHERE analysis IS NOT NULL - AND COALESCE(analysis->>'success', 'true') != 'false' + WHERE status = 'analysed' AND analysis->'trackers' IS NOT NULL ) AS latest_analysis FROM apps @@ -269,7 +266,7 @@ const getSiteDataSignature = async () => { } const countAnalysed = async () => { - const result = await pool.query("SELECT COUNT(*) FROM apps WHERE analysis IS NOT NULL AND COALESCE(analysis->>'success', 'true') != 'false'"); + const result = await pool.query("SELECT COUNT(*) FROM apps WHERE status = 'analysed'"); return parseInt(result.rows[0].count, 10); } @@ -283,5 +280,6 @@ module.exports = { updateAnalysis, getAllApps, getSiteDataSignature, - healthCheck + healthCheck, + deriveAnalysisState } diff --git a/test/appsState.test.js b/test/appsState.test.js new file mode 100644 index 0000000..cb72c40 --- /dev/null +++ b/test/appsState.test.js @@ -0,0 +1,38 @@ +const assert = require('node:assert/strict'); +const test = require('node:test'); +const { deriveAnalysisState } = require('../models/Apps'); + +test('successful analysis maps to analysed with no failure fields', () => { + assert.deepEqual( + deriveAnalysisState({ success: true, trackers: {} }), + { status: 'analysed', failureReason: null, failureRetryable: null } + ); +}); + +test('missing success flag is treated as a successful analysis', () => { + assert.deepEqual( + deriveAnalysisState({ trackers: {} }), + { status: 'analysed', failureReason: null, failureRetryable: null } + ); +}); + +test('failure keeps reason and defaults retryable to true', () => { + assert.deepEqual( + deriveAnalysisState({ success: false, reason: 'analysis_failed', logs: 'boom' }), + { status: 'failed', failureReason: 'analysis_failed', failureRetryable: true } + ); +}); + +test('failure with retryable false is non-retryable', () => { + assert.deepEqual( + deriveAnalysisState({ success: false, reason: 'paid_app', retryable: false }), + { status: 'failed', failureReason: 'paid_app', failureRetryable: false } + ); +}); + +test('failure falls back to logs when reason is absent', () => { + assert.deepEqual( + deriveAnalysisState({ success: false, logs: 'raw log text' }), + { status: 'failed', failureReason: 'raw log text', failureRetryable: true } + ); +}); From 0d6bc7b47cd81dae8d9e4e7dec889d9bff14a29c Mon Sep 17 00:00:00 2001 From: Konrad Kollnig <5175206+kasnder@users.noreply.github.com> Date: Mon, 13 Jul 2026 15:32:20 +0200 Subject: [PATCH 3/4] Port operational scripts onto status columns queue-refetch, priority-report and queue-status now classify queue/lock/ failure state via the status/processing_started/failure_retryable columns instead of analysis-JSON predicates. queue-refetch also resets status back to 'queued' (and clears the failure/lock columns) when it queues a refetch. Co-Authored-By: Claude Fable 5 --- scripts/priority-report.js | 58 +++++++++++------------ scripts/queue-refetch.js | 17 +++++-- scripts/queue-status.js | 95 +++++++++++++++++--------------------- 3 files changed, 81 insertions(+), 89 deletions(-) diff --git a/scripts/priority-report.js b/scripts/priority-report.js index d416d9d..c602498 100644 --- a/scripts/priority-report.js +++ b/scripts/priority-report.js @@ -21,21 +21,22 @@ const reviewsExpr = ` `; const queueCandidateWhere = ` - coalesce(analysis->>'retryable', 'true') <> 'false' + status = 'queued' + OR ( + status = 'processing' + AND processing_started < NOW() - ($3::int * INTERVAL '1 minute') + ) + OR ( + status = 'analysed' AND ( - analysis IS NULL - OR ( - analysis->>'logs' = 'Processing in progress' - AND (analysis->>'timestamp')::timestamptz < NOW() - ($3::int * INTERVAL '1 minute') - ) - OR ( - coalesce(analysis->>'logs', '') <> 'Processing in progress' - AND ( - analysisversion IS DISTINCT FROM $1 - OR analysed < NOW() - ($2::int * INTERVAL '1 day') - ) - ) + analysisversion IS DISTINCT FROM $1 + OR analysed < NOW() - ($2::int * INTERVAL '1 day') ) + ) + OR ( + status = 'failed' + AND failure_retryable + ) `; function printSection(title, rows, dateField) { @@ -72,31 +73,24 @@ async function main() { const counts = await client.query(` SELECT count(*)::int AS total, - count(*) FILTER (WHERE analysis IS NULL)::int AS queued, + count(*) FILTER (WHERE status = 'queued')::int AS queued, count(*) FILTER ( - WHERE analysis IS NOT NULL - AND coalesce(analysis->>'logs', '') <> 'Processing in progress' + WHERE status = 'analysed' AND ( analysisversion IS DISTINCT FROM $1 OR analysed < NOW() - ($2::int * INTERVAL '1 day') ) )::int AS stale, count(*) FILTER ( - WHERE analysis->>'logs' = 'Processing in progress' - AND (analysis->>'timestamp')::timestamptz >= NOW() - ($3::int * INTERVAL '1 minute') + WHERE status = 'processing' + AND processing_started >= NOW() - ($3::int * INTERVAL '1 minute') )::int AS processing, count(*) FILTER ( - WHERE analysis->>'logs' = 'Processing in progress' - AND (analysis->>'timestamp')::timestamptz < NOW() - ($3::int * INTERVAL '1 minute') + WHERE status = 'processing' + AND processing_started < NOW() - ($3::int * INTERVAL '1 minute') )::int AS expired_processing, - count(*) FILTER ( - WHERE analysis->>'success' = 'false' - AND coalesce(analysis->>'logs', '') <> 'Processing in progress' - )::int AS failed, - count(*) FILTER ( - WHERE analysis IS NOT NULL - AND coalesce(analysis->>'success', 'true') <> 'false' - )::int AS successful, + count(*) FILTER (WHERE status = 'failed')::int AS failed, + count(*) FILTER (WHERE status = 'analysed')::int AS successful, max(analysed) AS newest_analysed FROM apps `, [currentAnalysisVersion, staleAnalysisDays, processingTimeoutMinutes]); @@ -109,8 +103,9 @@ async function main() { ${reviewsExpr} AS reviews, round((details->>'score')::numeric, 2) AS score, CASE - WHEN analysis IS NULL THEN 'never analysed' - WHEN analysis->>'logs' = 'Processing in progress' THEN 'expired processing marker' + WHEN status = 'queued' THEN 'never analysed' + WHEN status = 'processing' THEN 'expired processing lock' + WHEN status = 'failed' THEN 'retryable failure' WHEN analysisversion IS DISTINCT FROM $1 THEN 'old analysis version' ELSE 'older than stale cutoff' END AS reason, @@ -130,8 +125,7 @@ async function main() { round((details->>'score')::numeric, 2) AS score, analysed FROM apps - WHERE analysis->>'success' = 'false' - AND coalesce(analysis->>'logs', '') <> 'Processing in progress' + WHERE status = 'failed' ORDER BY ${reviewsExpr} DESC, analysed ASC LIMIT $1 `, [limit]); diff --git a/scripts/queue-refetch.js b/scripts/queue-refetch.js index 7c59aa5..949faec 100644 --- a/scripts/queue-refetch.js +++ b/scripts/queue-refetch.js @@ -101,10 +101,9 @@ async function main() { } } else { const where = failed - ? `analysis->>'success' = 'false' - AND coalesce(analysis->>'logs', '') <> 'Processing in progress' - ${includeNonRetryable ? '' : "AND coalesce(analysis->>'retryable', 'true') <> 'false'"}` - : "analysis IS NOT NULL AND coalesce(analysis->>'success', 'true') <> 'false'"; + ? `status = 'failed' + ${includeNonRetryable ? '' : 'AND failure_retryable'}` + : "status = 'analysed'"; const result = await client.query(` SELECT appid, details->>'title' AS title, ${reviewsExpr} AS reviews, analysed @@ -129,7 +128,15 @@ async function main() { for (const row of rows) { await snapshotCurrentAnalysis(client, row.appid); await client.query( - 'UPDATE apps SET analysis = NULL, analysisversion = NULL, analysed = NULL WHERE appid = $1', + `UPDATE apps + SET analysis = NULL, + analysisversion = NULL, + analysed = NULL, + status = 'queued', + processing_started = NULL, + failure_reason = NULL, + failure_retryable = NULL + WHERE appid = $1`, [row.appid] ); } diff --git a/scripts/queue-status.js b/scripts/queue-status.js index 5a4c329..1e11c96 100644 --- a/scripts/queue-status.js +++ b/scripts/queue-status.js @@ -58,57 +58,46 @@ async function main() { const summaryResult = await client.query(` SELECT count(*)::int AS total, - count(*) FILTER (WHERE analysis IS NULL)::int AS never_or_reset, + count(*) FILTER (WHERE status = 'queued')::int AS never_or_reset, count(*) FILTER ( - WHERE analysis->>'logs' = 'Processing in progress' - AND (analysis->>'timestamp')::timestamptz >= NOW() - ($3::int * INTERVAL '1 minute') + WHERE status = 'processing' + AND processing_started >= NOW() - ($3::int * INTERVAL '1 minute') )::int AS active_processing, count(*) FILTER ( - WHERE analysis->>'logs' = 'Processing in progress' - AND (analysis->>'timestamp')::timestamptz < NOW() - ($3::int * INTERVAL '1 minute') + WHERE status = 'processing' + AND processing_started < NOW() - ($3::int * INTERVAL '1 minute') )::int AS expired_processing, + count(*) FILTER (WHERE status = 'failed')::int AS failed, + count(*) FILTER (WHERE status = 'analysed')::int AS analysed_success, count(*) FILTER ( - WHERE analysis IS NOT NULL - AND coalesce(analysis->>'logs', '') <> 'Processing in progress' - AND coalesce(analysis->>'success', 'true') = 'false' - )::int AS failed, - count(*) FILTER ( - WHERE analysis IS NOT NULL - AND coalesce(analysis->>'logs', '') <> 'Processing in progress' - AND coalesce(analysis->>'success', 'true') <> 'false' - )::int AS analysed_success, - count(*) FILTER ( - WHERE analysis IS NOT NULL - AND coalesce(analysis->>'logs', '') <> 'Processing in progress' - AND coalesce(analysis->>'success', 'true') <> 'false' + WHERE status = 'analysed' AND analysisversion = $1 AND analysed >= NOW() - ($2::int * INTERVAL '1 day') )::int AS fresh_success, count(*) FILTER ( - WHERE analysis IS NOT NULL - AND coalesce(analysis->>'logs', '') <> 'Processing in progress' - AND coalesce(analysis->>'success', 'true') <> 'false' + WHERE status = 'analysed' AND ( analysisversion IS DISTINCT FROM $1 OR analysed < NOW() - ($2::int * INTERVAL '1 day') ) )::int AS stale_success, count(*) FILTER ( - WHERE coalesce(analysis->>'retryable', 'true') <> 'false' - AND ( - analysis IS NULL - OR ( - analysis->>'logs' = 'Processing in progress' - AND (analysis->>'timestamp')::timestamptz < NOW() - ($3::int * INTERVAL '1 minute') - ) - OR ( - coalesce(analysis->>'logs', '') <> 'Processing in progress' - AND ( - analysisversion IS DISTINCT FROM $1 - OR analysed < NOW() - ($2::int * INTERVAL '1 day') - ) + WHERE status = 'queued' + OR ( + status = 'processing' + AND processing_started < NOW() - ($3::int * INTERVAL '1 minute') + ) + OR ( + status = 'analysed' + AND ( + analysisversion IS DISTINCT FROM $1 + OR analysed < NOW() - ($2::int * INTERVAL '1 day') ) ) + OR ( + status = 'failed' + AND failure_retryable + ) )::int AS queue_backlog, count(*) FILTER (WHERE analysed >= NOW() - INTERVAL '1 hour')::int AS analysed_1h, count(*) FILTER (WHERE analysed >= NOW() - INTERVAL '24 hours')::int AS analysed_24h, @@ -122,10 +111,10 @@ async function main() { SELECT appid, details->>'title' AS title, - (analysis->>'timestamp')::timestamptz AS started_at, - EXTRACT(EPOCH FROM (NOW() - (analysis->>'timestamp')::timestamptz))::int AS age_seconds + processing_started AS started_at, + EXTRACT(EPOCH FROM (NOW() - processing_started))::int AS age_seconds FROM apps - WHERE analysis->>'logs' = 'Processing in progress' + WHERE status = 'processing' ORDER BY started_at ASC LIMIT 10 `); @@ -136,27 +125,29 @@ async function main() { details->>'title' AS title, ${reviewsExpr} AS reviews, CASE - WHEN analysis IS NULL THEN 'never analysed/reset' - WHEN analysis->>'logs' = 'Processing in progress' THEN 'expired processing' + WHEN status = 'queued' THEN 'never analysed/reset' + WHEN status = 'processing' THEN 'expired processing' + WHEN status = 'failed' THEN 'retryable failure' WHEN analysisversion IS DISTINCT FROM $1 THEN 'old analysis version' ELSE 'stale by age' END AS reason FROM apps - WHERE coalesce(analysis->>'retryable', 'true') <> 'false' - AND ( - analysis IS NULL - OR ( - analysis->>'logs' = 'Processing in progress' - AND (analysis->>'timestamp')::timestamptz < NOW() - ($3::int * INTERVAL '1 minute') - ) - OR ( - coalesce(analysis->>'logs', '') <> 'Processing in progress' - AND ( - analysisversion IS DISTINCT FROM $1 - OR analysed < NOW() - ($2::int * INTERVAL '1 day') - ) + WHERE status = 'queued' + OR ( + status = 'processing' + AND processing_started < NOW() - ($3::int * INTERVAL '1 minute') + ) + OR ( + status = 'analysed' + AND ( + analysisversion IS DISTINCT FROM $1 + OR analysed < NOW() - ($2::int * INTERVAL '1 day') ) ) + OR ( + status = 'failed' + AND failure_retryable + ) ORDER BY ${reviewsExpr} DESC, added ASC LIMIT 10 `, [currentAnalysisVersion, staleAnalysisDays, processingTimeoutMinutes]); From df880c2a95e88eaff26a6c0152583ea069175d92 Mon Sep 17 00:00:00 2001 From: Konrad Kollnig <5175206+kasnder@users.noreply.github.com> Date: Mon, 13 Jul 2026 15:37:00 +0200 Subject: [PATCH 4/4] Gate failure retries behind version/staleness and harden addApp conflict handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The status-column port made status='failed' AND failure_retryable rows immediately eligible in nextApp, but the old JSON predicate only retried failures once their analysisversion was outdated or the analysis was stale — without that gate a popular, persistently failing app would be handed to the analyser in a tight loop and starve the queue. Restore the gate in nextApp and the two reporting scripts, and shape the failed partial index to match. addApp now uses a bare ON CONFLICT so inserts racing across case-variants of the same bundle ID (unique lower(appid) from 009) are ignored instead of erroring. Co-Authored-By: Claude Fable 5 --- migrations/007_apps_state_columns.sql | 2 +- models/Apps.js | 8 +++++++- scripts/priority-report.js | 4 ++++ scripts/queue-status.js | 8 ++++++++ 4 files changed, 20 insertions(+), 2 deletions(-) diff --git a/migrations/007_apps_state_columns.sql b/migrations/007_apps_state_columns.sql index ab1d495..de29b40 100644 --- a/migrations/007_apps_state_columns.sql +++ b/migrations/007_apps_state_columns.sql @@ -61,7 +61,7 @@ CREATE INDEX IF NOT EXISTS apps_processing_started_idx WHERE status = 'processing'; CREATE INDEX IF NOT EXISTS apps_failed_retryable_idx - ON apps (status) + ON apps (analysisversion, analysed) WHERE status = 'failed' AND failure_retryable; CREATE INDEX IF NOT EXISTS apps_analysed_version_idx diff --git a/models/Apps.js b/models/Apps.js index a07d357..f5f4fff 100644 --- a/models/Apps.js +++ b/models/Apps.js @@ -81,7 +81,9 @@ const addApp = async (appId, details) => { if (!isValidAppId(appId)) throw new TypeError('Invalid App Store bundle ID'); if (!details || details.appId !== appId) throw new TypeError('App Store bundle ID mismatch'); - const result = await pool.query('INSERT INTO apps (appid, details) VALUES ($1, $2) ON CONFLICT (appid) DO NOTHING', [appId, details]); + // Bare ON CONFLICT covers both the appid primary key and the + // case-insensitive lower(appid) unique index from migration 009. + const result = await pool.query('INSERT INTO apps (appid, details) VALUES ($1, $2) ON CONFLICT DO NOTHING', [appId, details]); return result; } @@ -128,6 +130,10 @@ const nextApp = async () => { OR ( status = 'failed' AND failure_retryable + AND ( + analysisversion IS DISTINCT FROM $1 + OR analysed < NOW() - ($2::int * INTERVAL '1 day') + ) ) ) ORDER BY ${popularityExpression} DESC, added ASC diff --git a/scripts/priority-report.js b/scripts/priority-report.js index c602498..b0257d0 100644 --- a/scripts/priority-report.js +++ b/scripts/priority-report.js @@ -36,6 +36,10 @@ const queueCandidateWhere = ` OR ( status = 'failed' AND failure_retryable + AND ( + analysisversion IS DISTINCT FROM $1 + OR analysed < NOW() - ($2::int * INTERVAL '1 day') + ) ) `; diff --git a/scripts/queue-status.js b/scripts/queue-status.js index 1e11c96..64f26ae 100644 --- a/scripts/queue-status.js +++ b/scripts/queue-status.js @@ -97,6 +97,10 @@ async function main() { OR ( status = 'failed' AND failure_retryable + AND ( + analysisversion IS DISTINCT FROM $1 + OR analysed < NOW() - ($2::int * INTERVAL '1 day') + ) ) )::int AS queue_backlog, count(*) FILTER (WHERE analysed >= NOW() - INTERVAL '1 hour')::int AS analysed_1h, @@ -147,6 +151,10 @@ async function main() { OR ( status = 'failed' AND failure_retryable + AND ( + analysisversion IS DISTINCT FROM $1 + OR analysed < NOW() - ($2::int * INTERVAL '1 day') + ) ) ORDER BY ${reviewsExpr} DESC, added ASC LIMIT 10