Merge pull request #3553 from omnivore-app/fix/low-priority-save-page-job

fix: save url operation is delayed

packages/api/src/jobs/rss/refreshAllFeeds.ts

@@ -3,10 +3,13 @@ import { DataSource } from 'typeorm'
 import { v4 as uuid } from 'uuid'
 import { getBackendQueue, JOB_VERSION } from '../../queue-processor'
 import { validateUrl } from '../../services/create_page_save_request'
-import { RssSubscriptionGroup } from '../../utils/createTask'
+import { getJobPriority, RssSubscriptionGroup } from '../../utils/createTask'
 import { stringToHash } from '../../utils/helpers'
 import { logger } from '../../utils/logger'
 
+export const REFRESH_ALL_FEEDS_JOB_NAME = 'refresh-all-feeds'
+export const REFRESH_FEED_JOB_NAME = 'refresh-feed'
+
 export type RSSRefreshContext = {
   type: 'all' | 'user-added'
   refreshID: string
@@ -116,10 +119,10 @@ export const queueRSSRefreshAllFeedsJob = async () => {
     return false
   }
   return queue.add(
-    'refresh-all-feeds',
+    REFRESH_ALL_FEEDS_JOB_NAME,
     {},
     {
-      priority: 100,
+      priority: getJobPriority(REFRESH_ALL_FEEDS_JOB_NAME),
     }
   )
 }
@@ -129,15 +132,15 @@ type QueuePriority = 'low' | 'high'
 export const queueRSSRefreshFeedJob = async (
   jobid: string,
   payload: any,
-  options = { priority: 'high' as QueuePriority }
+  options = { priority: 'low' as QueuePriority }
 ): Promise<Job | undefined> => {
   const queue = await getBackendQueue()
   if (!queue) {
     return undefined
   }
-  return queue.add('refresh-feed', payload, {
+  return queue.add(REFRESH_FEED_JOB_NAME, payload, {
     jobId: `${jobid}_${JOB_VERSION}`,
-    priority: options.priority == 'low' ? 10 : 50,
+    priority: getJobPriority(`${REFRESH_FEED_JOB_NAME}_${options.priority}`),
     removeOnComplete: true,
     removeOnFail: true,
   })

packages/api/src/queue-processor.ts

@@ -11,11 +11,10 @@ import {
   Worker,
 } from 'bullmq'
 import express, { Express } from 'express'
-import { SnakeNamingStrategy } from 'typeorm-naming-strategies'
 import { appDataSource } from './data_source'
 import { env } from './env'
 import { bulkAction, BULK_ACTION_JOB_NAME } from './jobs/bulk_action'
-import { CALL_WEBHOOK_JOB_NAME, callWebhook } from './jobs/call_webhook'
+import { callWebhook, CALL_WEBHOOK_JOB_NAME } from './jobs/call_webhook'
 import { findThumbnail, THUMBNAIL_JOB } from './jobs/find_thumbnail'
 import { refreshAllFeeds } from './jobs/rss/refreshAllFeeds'
 import { refreshFeed } from './jobs/rss/refreshFeed'
@@ -34,7 +33,8 @@ import {
 import { updatePDFContentJob } from './jobs/update_pdf_content'
 import { redisDataSource } from './redis_data_source'
 import { CACHED_READING_POSITION_PREFIX } from './services/cached_reading_position'
-import { CustomTypeOrmLogger, logger } from './utils/logger'
+import { getJobPriority } from './utils/createTask'
+import { logger } from './utils/logger'
 
 export const QUEUE_NAME = 'omnivore-backend-queue'
 export const JOB_VERSION = 'v001'
@@ -121,7 +121,7 @@ const setupCronJobs = async () => {
     SYNC_READ_POSITIONS_JOB_NAME,
     {},
     {
-      priority: 1,
+      priority: getJobPriority(SYNC_READ_POSITIONS_JOB_NAME),
       repeat: {
         every: 60_000,
       },

packages/api/src/utils/createTask.ts

@@ -17,7 +17,12 @@ import {
 import { BulkActionData, BULK_ACTION_JOB_NAME } from '../jobs/bulk_action'
 import { CallWebhookJobData, CALL_WEBHOOK_JOB_NAME } from '../jobs/call_webhook'
 import { THUMBNAIL_JOB } from '../jobs/find_thumbnail'
-import { queueRSSRefreshFeedJob } from '../jobs/rss/refreshAllFeeds'
+import {
+  queueRSSRefreshFeedJob,
+  REFRESH_ALL_FEEDS_JOB_NAME,
+  REFRESH_FEED_JOB_NAME,
+} from '../jobs/rss/refreshAllFeeds'
+import { SYNC_READ_POSITIONS_JOB_NAME } from '../jobs/sync_read_positions'
 import { TriggerRuleJobData, TRIGGER_RULE_JOB_NAME } from '../jobs/trigger_rule'
 import {
   UpdateHighlightData,
@@ -37,6 +42,38 @@ import View = google.cloud.tasks.v2.Task.View
 // Instantiates a client.
 const client = new CloudTasksClient()
 
+/**
+ * we want to prioritized jobs by the expected time to complete
+ * lower number means higher priority
+ * priority 1: jobs that are expected to run immediately
+ * priority 5: jobs that are expected to run in less than 10 seconds
+ * priority 10: jobs that are expected to run in less than 1 minute
+ * priority 50: jobs that are expected to run in less than 30 minutes
+ * priority 100: jobs that are expected to run in less than 1 hour
+ **/
+export const getJobPriority = (jobName: string): number => {
+  switch (jobName) {
+    case UPDATE_LABELS_JOB:
+    case UPDATE_HIGHLIGHT_JOB:
+    case SYNC_READ_POSITIONS_JOB_NAME:
+      return 1
+    case TRIGGER_RULE_JOB_NAME:
+    case CALL_WEBHOOK_JOB_NAME:
+      return 5
+    case BULK_ACTION_JOB_NAME:
+    case `${REFRESH_FEED_JOB_NAME}_high`:
+      return 10
+    case `${REFRESH_FEED_JOB_NAME}_low`:
+      return 50
+    case REFRESH_ALL_FEEDS_JOB_NAME:
+    case THUMBNAIL_JOB:
+      return 100
+    default:
+      logger.error(`unknown job name: ${jobName}`)
+      return 1
+  }
+}
+
 const logError = (error: any): void => {
   if (axios.isAxiosError(error)) {
     logger.error(error.response)
@@ -283,6 +320,7 @@ export const enqueueParseRequest = async ({
     publishedAt,
     folder,
     rssFeedUrl,
+    priority,
   }
 
   // If there is no Google Cloud Project Id exposed, it means that we are in local environment
@@ -600,7 +638,7 @@ export const enqueueThumbnailJob = async (
     libraryItemId,
   }
   return queue.add(THUMBNAIL_JOB, payload, {
-    priority: 100,
+    priority: getJobPriority(THUMBNAIL_JOB),
     attempts: 1,
     removeOnComplete: true,
   })
@@ -664,7 +702,7 @@ export const enqueueTriggerRuleJob = async (data: TriggerRuleJobData) => {
   }
 
   return queue.add(TRIGGER_RULE_JOB_NAME, data, {
-    priority: 5,
+    priority: getJobPriority(TRIGGER_RULE_JOB_NAME),
     attempts: 1,
   })
 }
@@ -676,7 +714,7 @@ export const enqueueWebhookJob = async (data: CallWebhookJobData) => {
   }
 
   return queue.add(CALL_WEBHOOK_JOB_NAME, data, {
-    priority: 5,
+    priority: getJobPriority(CALL_WEBHOOK_JOB_NAME),
     attempts: 1,
   })
 }
@@ -693,7 +731,7 @@ export const bulkEnqueueUpdateLabels = async (data: UpdateLabelsData[]) => {
     opts: {
       jobId: `${UPDATE_LABELS_JOB}_${d.libraryItemId}_${JOB_VERSION}`,
       attempts: 6,
-      priority: 1,
+      priority: getJobPriority(UPDATE_LABELS_JOB),
       removeOnComplete: true,
       removeOnFail: true,
     },
@@ -717,7 +755,7 @@ export const enqueueUpdateHighlight = async (data: UpdateHighlightData) => {
     return queue.add(UPDATE_HIGHLIGHT_JOB, data, {
       jobId: `${UPDATE_HIGHLIGHT_JOB}_${data.libraryItemId}_${JOB_VERSION}`,
       attempts: 6,
-      priority: 1,
+      priority: getJobPriority(UPDATE_HIGHLIGHT_JOB),
       removeOnComplete: true,
       removeOnFail: true,
     })
@@ -737,7 +775,7 @@ export const enqueueBulkAction = async (data: BulkActionData) => {
   try {
     return queue.add(BULK_ACTION_JOB_NAME, data, {
       attempts: 1,
-      priority: 10,
+      priority: getJobPriority(BULK_ACTION_JOB_NAME),
       jobId, // deduplication
       removeOnComplete: true,
       removeOnFail: true,

packages/content-fetch/src/job.ts

@@ -9,6 +9,7 @@ interface savePageJob {
   data: unknown
   isRss: boolean
   isImport: boolean
+  priority: 'low' | 'high'
 }
 
 const queue = new Queue(QUEUE_NAME, {
@@ -29,7 +30,7 @@ const getPriority = (job: savePageJob): number => {
     return 100
   }
 
-  return 1
+  return job.priority === 'low' ? 10 : 1
 }
 
 const getAttempts = (job: savePageJob): number => {

packages/content-fetch/src/request_handler.ts

@@ -23,6 +23,7 @@ interface RequestBody {
   publishedAt?: string
   folder?: string
   users?: User[]
+  priority: 'high' | 'low'
 }
 
 interface LogRecord {
@@ -88,6 +89,7 @@ export const contentFetchRequestHandler: RequestHandler = async (req, res) => {
   const rssFeedUrl = body.rssFeedUrl
   const savedAt = body.savedAt
   const publishedAt = body.publishedAt
+  const priority = body.priority
 
   const logRecord: LogRecord = {
     url,
@@ -130,6 +132,7 @@ export const contentFetchRequestHandler: RequestHandler = async (req, res) => {
       },
       isRss: !!rssFeedUrl,
       isImport: !!taskId,
+      priority,
     }))
 
     const cacheResult = await cacheFetchResult(fetchResult)