Merge pull request #3859 from omnivore-app/fix/pdf

fix memory leak from axios error logging

packages/api/src/jobs/save_page.ts

@@ -11,14 +11,14 @@ import { userRepository } from '../repository/user'
 import { saveFile } from '../services/save_file'
 import { savePage } from '../services/save_page'
 import { uploadFile } from '../services/upload_file'
-import { logger } from '../utils/logger'
+import { logError, logger } from '../utils/logger'
 
 const signToken = promisify(jwt.sign)
 
 const IMPORTER_METRICS_COLLECTOR_URL = env.queue.importerMetricsUrl
 const JWT_SECRET = env.server.jwtSecret
 
-const MAX_ATTEMPTS = 2
+const MAX_IMPORT_ATTEMPTS = 1
 const REQUEST_TIMEOUT = 30000 // 30 seconds
 
 interface Data {
@@ -52,29 +52,32 @@ const uploadToSignedUrl = async (
   contentType: string,
   contentObjUrl: string
 ) => {
-  logger.info('uploading to signed url', {
-    uploadSignedUrl,
-    contentType,
+  const maxContentLength = 10 * 1024 * 1024 // 10MB
+
+  logger.info('downloading content', {
     contentObjUrl,
   })
 
-  try {
-    const stream = await axios.get(contentObjUrl, {
-      responseType: 'stream',
-      timeout: REQUEST_TIMEOUT,
-    })
-    return await axios.put(uploadSignedUrl, stream.data, {
-      headers: {
-        'Content-Type': contentType,
-      },
-      maxBodyLength: 1000000000,
-      maxContentLength: 100000000,
-      timeout: REQUEST_TIMEOUT,
-    })
-  } catch (error) {
-    logger.error('error uploading to signed url', error)
-    return null
-  }
+  // download the content as stream and max 10MB
+  const response = await axios.get(contentObjUrl, {
+    responseType: 'stream',
+    maxContentLength,
+    timeout: REQUEST_TIMEOUT,
+  })
+
+  logger.info('uploading to signed url', {
+    uploadSignedUrl,
+    contentType,
+  })
+
+  // upload the stream to the signed url
+  await axios.put(uploadSignedUrl, response.data, {
+    headers: {
+      'Content-Type': contentType,
+    },
+    maxBodyLength: maxContentLength,
+    timeout: REQUEST_TIMEOUT,
+  })
 }
 
 const uploadPdf = async (
@@ -95,14 +98,13 @@ const uploadPdf = async (
     throw new Error('error while getting upload id and signed url')
   }
 
-  const uploaded = await uploadToSignedUrl(
-    result.uploadSignedUrl,
-    'application/pdf',
-    url
-  )
-  if (!uploaded) {
-    throw new Error('error while uploading pdf')
-  }
+  await uploadToSignedUrl(result.uploadSignedUrl, 'application/pdf', url)
+
+  logger.info('pdf uploaded successfully', {
+    url,
+    uploadFileId: result.id,
+    itemId: result.createdPageId,
+  })
 
   return {
     uploadFileId: result.id,
@@ -134,7 +136,7 @@ const sendImportStatusUpdate = async (
       }
     )
   } catch (e) {
-    logger.error('error while sending import status update', e)
+    logError(e)
   }
 }
 
@@ -268,20 +270,14 @@ export const savePageJob = async (data: Data, attemptsMade: number) => {
     isImported = true
     isSaved = true
   } catch (e) {
-    if (e instanceof Error) {
-      logger.error(`error while saving page: ${e.message}`)
-    } else {
-      logger.error('error while saving page: unknown error')
-    }
+    logError(e)
 
     throw e
   } finally {
-    const lastAttempt = attemptsMade === MAX_ATTEMPTS - 1
-    if (lastAttempt) {
-      logger.info(`last attempt reached ${data.url}`)
-    }
+    const lastAttempt = attemptsMade + 1 === MAX_IMPORT_ATTEMPTS
 
     if (taskId && (isSaved || lastAttempt)) {
+      logger.info('sending import status update')
       // send import status to update the metrics for importer
       await sendImportStatusUpdate(userId, taskId, isImported)
     }

packages/api/src/utils/createTask.ts

@@ -60,7 +60,7 @@ import { signFeatureToken } from '../services/features'
 import { OmnivoreAuthorizationHeader } from './auth'
 import { CreateTaskError } from './errors'
 import { stringToHash } from './helpers'
-import { logger } from './logger'
+import { logError, logger } from './logger'
 import View = google.cloud.tasks.v2.Task.View
 
 // Instantiates a client.
@@ -106,14 +106,6 @@ export const getJobPriority = (jobName: string): number => {
   }
 }
 
-const logError = (error: any): void => {
-  if (axios.isAxiosError(error)) {
-    logger.error(error.response)
-  } else {
-    logger.error(error)
-  }
-}
-
 const createHttpTaskWithToken = async ({
   project = process.env.GOOGLE_CLOUD_PROJECT,
   queue = env.queue.name,

packages/api/src/utils/logger.ts

@@ -1,6 +1,7 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 /* eslint-disable @typescript-eslint/restrict-template-expressions */
 import { LoggingWinston } from '@google-cloud/logging-winston'
+import axios from 'axios'
 import jsonStringify from 'fast-safe-stringify'
 import { cloneDeep, isArray, isObject, isString, truncate } from 'lodash'
 import { DateTime } from 'luxon'
@@ -168,6 +169,19 @@ export interface LogRecord {
   [key: string]: any
 }
 
+export const logError = (error: any): void => {
+  if (axios.isAxiosError(error)) {
+    logger.error(error.message, {
+      response: error.response?.data,
+      stack: error.stack,
+    })
+  } else if (error instanceof Error) {
+    logger.error(error.message, { stack: error.stack })
+  } else {
+    logger.error(error)
+  }
+}
+
 export const logger = buildLogger('app')
 
 export default {}

packages/content-fetch/src/job.ts

@@ -59,7 +59,7 @@ const getAttempts = (job: SavePageJob): number => {
 
 const getOpts = (job: SavePageJob): BulkJobOptions => {
   return {
-    jobId: `save-page_${job.userId}_${job.data.finalUrl}`, // make sure we don't have duplicate jobs
+    jobId: `${JOB_NAME}_${job.userId}_${job.data.finalUrl}`, // make sure we don't have duplicate jobs
     removeOnComplete: true,
     removeOnFail: true,
     attempts: getAttempts(job),