Enhance memory management and error handling in attachment processing

lib/aws-s3.js

@@ -217,7 +217,7 @@ module.exports = class AWSAttachmentsService extends require("./basic") {
         )
       }
 
-      const { content = _content, ...metadata } = data
+      let { content = _content, ...metadata } = data
       const Key = metadata.url
 
       if (!Key) {
@@ -254,17 +254,24 @@ module.exports = class AWSAttachmentsService extends require("./basic") {
         params: input,
       })
 
-      const stored = super.put(attachments, metadata, null, isDraftEnabled)
-      await Promise.all([stored, multipartUpload.done()])
+      try {
+        const stored = super.put(attachments, metadata, null, isDraftEnabled)
+        await Promise.all([stored, multipartUpload.done()])
 
-      const duration = Date.now() - startTime
-      logConfig.debug('File upload to S3 completed successfully', {
-        filename: metadata.filename,
-        fileId: metadata.ID,
-        bucket: this.bucket,
-        key: Key,
-        duration
-      })
+        const duration = Date.now() - startTime
+        logConfig.debug('File upload to S3 completed successfully', {
+          filename: metadata.filename,
+          fileId: metadata.ID,
+          bucket: this.bucket,
+          key: Key,
+          duration
+        })
+      } finally {
+        // clear content reference after upload to minimize memory usage
+        content = null
+        data.content = null
+        input.Body = null
+      }
 
       // Initiate malware scan if configured
       if (this.kind === 's3') {
@@ -454,22 +461,29 @@ module.exports = class AWSAttachmentsService extends require("./basic") {
           client: this.client,
           params: input,
         })
-        await Promise.all([multipartUpload.done()])
 
-        const keys = { ID: targetID }
+        try {
+          await Promise.all([multipartUpload.done()])
 
-        const scanRequestJob = cds.spawn(async () => {
-          await scanRequest(req.target, keys, req)
-        })
+          const keys = { ID: targetID }
 
-        scanRequestJob.on('error', async (err) => {
-          logConfig.withSuggestion('error',
-            'Failed to initiate malware scan for attachment', err,
-            'Check malware scanner configuration and connectivity',
-            { keys, errorMessage: err.message })
-        })
+          const scanRequestJob = cds.spawn(async () => {
+            await scanRequest(req.target, keys, req)
+          })
 
-        logConfig.debug(`[S3 Upload] Uploaded file using updateContentHandler for ${req.target.name}`)
+          scanRequestJob.on('error', async (err) => {
+            logConfig.withSuggestion('error',
+              'Failed to initiate malware scan for attachment', err,
+              'Check malware scanner configuration and connectivity',
+              { keys, errorMessage: err.message })
+          })
+
+          logConfig.debug(`[S3 Upload] Uploaded file using updateContentHandler for ${req.target.name}`)
+        } finally {
+          // clear content reference after upload to minimize memory usage
+          req.data.content = null
+          input.Body = null
+        }
       }
     } else if (req?.data?.note) {
       const key = { ID: targetID }

lib/basic.js

@@ -51,6 +51,13 @@ module.exports = class AttachmentsService extends cds.Service {
           'Check database connectivity and attachment entity configuration',
           { attachmentEntity: attachments.name, recordCount: data.length, errorMessage: error.message })
         throw error
+      } finally {
+        // Cleanup: clear content reference after upload to minimize memory usage
+        data.forEach((d) => {
+          if (d.content) {
+            d.content = null
+          }
+        })
       }
     }
 
@@ -139,14 +146,24 @@ module.exports = class AttachmentsService extends cds.Service {
    * Handles non-draft attachment updates by uploading content to the database
    * @param {Express.Request} req - The request object
    * @param {cds.Entity} attachment - Attachments entity definition
-   * @returns 
+   * @returns
    */
   async nonDraftHandler(req, attachment) {
     if (req?.content?.url?.endsWith("/content")) {
       const attachmentID = req.content.url.match(attachmentIDRegex)[1]
       const data = { ID: attachmentID, content: req.content }
       const isDraftEnabled = false
-      return this.put(attachment, [data], null, isDraftEnabled)
+      try {
+        return await this.put(attachment, [data], null, isDraftEnabled)
+      } finally {
+        // Cleanup: clear content reference after upload to minimize memory usage
+        if (data.content) {
+          data.content = null
+        }
+        if (req.content) {
+          req.content = null
+        }
+      }
     }
   }
 

lib/malwareScanner.js

@@ -80,14 +80,10 @@ async function scanRequest(Attachments, key, req) {
   logConfig.debug('Fetching file content for scanning', { fileId: key.ID })
   const contentStream = await AttachmentsSrv.get(currEntity, key)
 
-  let fileContent
-  try {
-    logConfig.verbose('Converting file stream to string for scanning', { fileId: key.ID })
-    fileContent = await streamToString(contentStream)
-  } catch (err) {
+  if (!contentStream) {
     logConfig.withSuggestion('error',
-      'Cannot read file content for malware scanning', err,
-      'Check file integrity and storage accessibility',
+      'Cannot fetch file content for malware scanning', null,
+      'Check file exists and storage accessibility',
       { fileId: key.ID })
     await updateStatus(AttachmentsSrv, key, "Failed", currEntity, draftEntity, activeEntity)
     return
@@ -97,19 +93,20 @@ async function scanRequest(Attachments, key, req) {
   const scanStartTime = Date.now()
 
   try {
-    logConfig.debug('Sending file to Malware Scanning Service', {
+    logConfig.debug('Streaming file to Malware Scanning Service', {
       fileId: key.ID,
-      scannerUri: credentials.uri,
-      contentLength: fileContent.length
+      scannerUri: credentials.uri
     })
 
+    // Stream the file directly to the scanner without loading into memory
     response = await fetch(`https://${credentials.uri}/scan`, {
       method: "POST",
       headers: {
         Authorization:
           "Basic " + Buffer.from(`${credentials.username}:${credentials.password}`, "binary").toString("base64"),
       },
-      body: fileContent,
+      body: contentStream,
+      duplex: 'half' // Required for streaming request bodies
     })
 
   } catch (error) {
@@ -119,6 +116,11 @@ async function scanRequest(Attachments, key, req) {
       'Check malware scanner service binding and network connectivity',
       { fileId: key.ID, scanDuration, scannerUri: credentials?.uri })
     await updateStatus(AttachmentsSrv, key, "Failed", currEntity, draftEntity, activeEntity)
+
+    // Cleanup: destroy stream on error
+    if (contentStream && typeof contentStream.destroy === 'function') {
+      contentStream.destroy()
+    }
     return
   }
 
@@ -151,6 +153,11 @@ async function scanRequest(Attachments, key, req) {
       'Check malware scanner service health and response format',
       { fileId: key.ID, scanDuration, httpStatus: response?.status })
     await updateStatus(AttachmentsSrv, key, "Failed", currEntity, draftEntity, activeEntity)
+  } finally {
+    // Cleanup: destroy stream after scan completes
+    if (contentStream && typeof contentStream.destroy === 'function') {
+      contentStream.destroy()
+    }
   }
 }
 
@@ -206,15 +213,6 @@ function getCredentials() {
   }
 }
 
-function streamToString(stream) {
-  const chunks = []
-  return new Promise((resolve, reject) => {
-    stream.on('data', (chunk) => chunks.push(Buffer.from(chunk)))
-    stream.on('error', (err) => reject(err))
-    stream.on('end', () => resolve(Buffer.concat(chunks).toString('utf8')))
-  })
-}
-
 module.exports = {
   scanRequest
 }

lib/plugin.js

@@ -118,13 +118,14 @@ cds.once("served", async function registerPluginHandlers() {
    * Reads the attachment content if requested
    * @param {[cds.Entity]} param0
    * @param {import('@sap/cds').Request} req - The request object
-   * @returns 
+   * @returns
    */
   async function readAttachment([attachment], req) {
     if (req._.readAfterWrite || !req?.req?.url?.endsWith("/content") || !attachment || attachment?.content) return
     let keys = { ID: req.req.url.match(attachmentIDRegex)[1] }
     let { target } = req
-    attachment.content = await AttachmentsSrv.get(target, keys) //Dependency -> sending req object for usage in SDM plugin
+    const content = await AttachmentsSrv.get(target, keys)
+    attachment.content = content
   }
 })