From 99c16be31af2d1db8914498ea3cefa2def1718e7 Mon Sep 17 00:00:00 2001 From: Rohit Amarnath <88762+ramarnat@users.noreply.github.com> Date: Wed, 28 Jan 2026 11:46:34 -0500 Subject: [PATCH] content-fetch: mark final-attempt failures as FAILED Ensure items don"t remain stuck in PROCESSING by marking the backing library_item as FAILED when content-fetch exhausts retries (or hits a blocked domain). --- packages/api/src/jobs/save_page.ts | 17 ++++ packages/content-fetch/src/request_handler.ts | 85 ++++++++++++++++++- packages/content-fetch/src/worker.ts | 29 ++++++- packages/content-fetch/start.sh | 2 +- 4 files changed, 125 insertions(+), 8 deletions(-) diff --git a/packages/api/src/jobs/save_page.ts b/packages/api/src/jobs/save_page.ts index a721f4643..382e54733 100644 --- a/packages/api/src/jobs/save_page.ts +++ b/packages/api/src/jobs/save_page.ts @@ -6,10 +6,12 @@ import { ArticleSavingRequestStatus, CreateLabelInput, } from '../generated/graphql' +import { LibraryItemState } from '../entity/library_item' import { redisDataSource } from '../redis_data_source' import { userRepository } from '../repository/user' import { saveFile } from '../services/save_file' import { savePage } from '../services/save_page' +import { updateLibraryItem } from '../services/library_item' import { uploadFile } from '../services/upload_file' import { logError, logger } from '../utils/logger' import { @@ -285,6 +287,21 @@ export const savePageJob = async (data: Data, attemptsMade: number) => { isSaved = true } catch (e) { logError(e) + try { + await updateLibraryItem( + articleSavingRequestId, + { + state: LibraryItemState.Failed, + }, + userId + ) + } catch (updateError) { + logger.error('Failed to mark library item as FAILED', { + userId, + articleSavingRequestId, + updateError, + }) + } throw e } finally { diff --git a/packages/content-fetch/src/request_handler.ts b/packages/content-fetch/src/request_handler.ts index 34c147924..ee2a03ae2 100644 --- a/packages/content-fetch/src/request_handler.ts +++ b/packages/content-fetch/src/request_handler.ts @@ -17,7 +17,8 @@ interface UserConfig { export interface JobData { url: string userId?: string - saveRequestId: string + // Back-compat: older payloads used saveRequestId + userId; newer payloads use users[]. + saveRequestId?: string state?: string labels?: string[] source?: string @@ -34,7 +35,7 @@ export interface JobData { interface LogRecord { url: string - articleSavingRequestId: string + articleSavingRequestId?: string labels: { source: string } @@ -75,9 +76,29 @@ const signToken = promisify(jwt.sign) const IMPORTER_METRICS_COLLECTOR_URL = process.env.IMPORTER_METRICS_COLLECTOR_URL const JWT_SECRET = process.env.JWT_SECRET +const REST_BACKEND_ENDPOINT = process.env.REST_BACKEND_ENDPOINT + +const API_GRAPHQL_ENDPOINT = + process.env.API_GRAPHQL_ENDPOINT || + (REST_BACKEND_ENDPOINT + ? REST_BACKEND_ENDPOINT.replace(/\/api\/?$/i, '/api/graphql') + : undefined) const MAX_IMPORT_ATTEMPTS = 1 +const UPDATE_PAGE_MUTATION = ` + mutation UpdatePage($input: UpdatePageInput!) { + updatePage(input: $input) { + ... on UpdatePageSuccess { + updatedPage { id } + } + ... on UpdatePageError { + errorCodes + } + } + } +` + const uploadToBucket = async (filePath: string, data: string) => { await storage .bucket(bucketName) @@ -219,6 +240,58 @@ const sendImportStatusUpdate = async ( } } +export const markFetchFailure = async ( + users: UserConfig[], + errorMessage: string +) => { + if (!JWT_SECRET || !API_GRAPHQL_ENDPOINT) { + console.error('JWT_SECRET or API_GRAPHQL_ENDPOINT is not set, cannot mark failure', { + hasJwtSecret: !!JWT_SECRET, + apiGraphqlEndpoint: API_GRAPHQL_ENDPOINT, + }) + return + } + + await Promise.all( + users.map(async (user) => { + try { + const auth = (await signToken( + { uid: user.id }, + JWT_SECRET + )) as string + + await axios.post( + API_GRAPHQL_ENDPOINT, + { + query: UPDATE_PAGE_MUTATION, + variables: { + input: { + pageId: user.libraryItemId, + state: 'FAILED', + // Keep existing savedAt/publishedAt/title/etc. + }, + }, + }, + { + headers: { + 'Omnivore-Authorization': auth, + 'Content-Type': 'application/json', + }, + timeout: 5000, + } + ) + } catch (e) { + console.error('Failed to mark fetch failure', { + userId: user.id, + libraryItemId: user.libraryItemId, + errorMessage, + }) + console.error(e) + } + }) + ) +} + export const processFetchContentJob = async ( redisDataSource: RedisDataSource, data: JobData, @@ -235,9 +308,9 @@ export const processFetchContentJob = async ( { id: userId, folder: data.folder, - libraryItemId: data.saveRequestId, + libraryItemId: data.saveRequestId || '', }, - ] + ].filter((u) => !!u.libraryItemId) } const articleSavingRequestId = data.saveRequestId const state = data.state @@ -278,6 +351,10 @@ export const processFetchContentJob = async ( console.log('domain is blocked', domain) logRecord.error = 'domain is blocked' + if (users.length) { + await markFetchFailure(users, logRecord.error) + } + return } diff --git a/packages/content-fetch/src/worker.ts b/packages/content-fetch/src/worker.ts index f209b279b..01081d307 100644 --- a/packages/content-fetch/src/worker.ts +++ b/packages/content-fetch/src/worker.ts @@ -1,6 +1,6 @@ import { RedisDataSource } from '@omnivore/utils' import { Job, Queue, RedisClient, Worker } from 'bullmq' -import { JobData, processFetchContentJob } from './request_handler' +import { JobData, markFetchFailure, processFetchContentJob } from './request_handler' export const QUEUE = 'omnivore-content-fetch-queue' @@ -34,8 +34,31 @@ export const createWorker = ( const worker = new Worker( queueName, async (job: Job) => { - // process the job - await processFetchContentJob(redisDataSource, job.data, job.attemptsMade) + const maxAttempts = job.opts.attempts ?? 1 + try { + // process the job + await processFetchContentJob(redisDataSource, job.data, job.attemptsMade) + } catch (e) { + const lastAttempt = job.attemptsMade + 1 >= maxAttempts + if (lastAttempt) { + const errorMessage = e instanceof Error ? e.message : 'unknown error' + let users = job.data.users || [] + if (job.data.userId) { + // Back-compat payload path + users = [ + { + id: job.data.userId, + folder: job.data.folder, + libraryItemId: job.data.saveRequestId || '', + }, + ].filter((u) => u.libraryItemId) + } + if (users.length) { + await markFetchFailure(users, errorMessage) + } + } + throw e + } }, { connection: redisDataSource.queueRedisClient, diff --git a/packages/content-fetch/start.sh b/packages/content-fetch/start.sh index 783ecfa15..2ace16534 100644 --- a/packages/content-fetch/start.sh +++ b/packages/content-fetch/start.sh @@ -1,3 +1,3 @@ #!/bin/sh cat hosts >> /etc/hosts -yarn workspace @omnivore/content-fetch start \ No newline at end of file +yarn workspace @omnivore/content-fetch start