feat: rescan attachments stuck in "Scanning" on server start

CHANGELOG.md

@@ -4,6 +4,26 @@ All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning](http://semver.org/).
 The format is based on [Keep a Changelog](http://keepachangelog.com/).
 
+## UNRELEASED
+
+### Added
+
+- Attachments stuck in `Scanning` status after a server crash or restart can now be automatically recovered on startup. Enable via the new `rescanOnStart` flag (defaults to `false`):
+
+  ```json
+  {
+    "cds": {
+      "requires": {
+        "attachments": {
+          "rescanOnStart": true
+        }
+      }
+    }
+  }
+  ```
+
+  When enabled, the plugin queries for all attachment rows in `Scanning` status at startup and re-emits `ScanAttachmentsFile` for each, allowing the normal scanner flow to move them to a terminal status (`Clean`/`Infected`/`Failed`). The sweep runs detached from the serving path (via `cds.spawn`) and is throttled by the existing `maxConcurrentScans` semaphore.
+
 ## Version 4.0.0 - 2026-08-03
 
 **BREAKING CHANGE: The attachments plugin comes now without hyperscaler dependencies, please make sure to install them accordingly!**

README.md

@@ -321,6 +321,27 @@ According to the recommendation of the [Malware Scanning Service](http://help.sa
 
 By default, `scanExpiryMs` is set to `259200000` milliseconds (3 days). Downloading an attachment is not permitted unless its status is `Clean`.
 
+#### Recovering attachments stuck in "Scanning" on startup
+
+If the server crashes or restarts while a scan is in progress, an attachment row can be left permanently in `Scanning` status. To automatically recover such rows on startup, enable the `rescanOnStart` flag:
+
+```json
+{
+  "cds": {
+    "requires": {
+      "attachments": {
+        "rescanOnStart": true
+      }
+    }
+  }
+}
+```
+
+When enabled, the plugin queries for all attachment rows in `Scanning` status at startup and re-emits `ScanAttachmentsFile` for each. The sweep runs in the background (via `cds.spawn`) so it never delays server startup, and is throttled by the existing [`maxConcurrentScans`](#scan-concurrency-limiting) semaphore. The flag defaults to `false` and has no effect when `scan` is disabled.
+
+> [!Note]
+> In multitenancy deployments, the startup sweep only covers the default tenant. Stuck rows in other tenants are still recovered via the on-download rescan path (`scanExpiryMs`).
+
 ### Audit logging
 
 The attachment service emits the following three events:

lib/plugin.js

@@ -242,6 +242,118 @@ cds.once("served", () => {
   })
 })
 
+// On server start, re-emit scan requests for attachment rows that are stuck in "Scanning".
+// This handles the case where the process crashed or restarted while a scan was in progress,
+// leaving the row in "Scanning" permanently. Work is detached via cds.spawn so it never
+// blocks startup or holds a DB connection on the serving path (same rationale as the
+// cds.spawn wrapping in rescan() — see generic-handlers.js for context).
+cds.on("served", () => {
+  if (!cds.env.requires.attachments) return
+  if (!(cds.env.requires.attachments?.scan ?? true)) return
+  if (!cds.env.requires.attachments?.rescanOnStart) return
+  if (!cds.services.db) return
+
+  // TODO(mt): per-tenant reconciliation for multitenancy deployments; for now only the
+  // default tenant is reconciled. The on-download rescan path (enforceScanPolicy) acts
+  // as the safety net for individual tenants in a shared-DB setup.
+  cds.spawn(async () => {
+    await rescanStuckAttachments()
+  })
+})
+
+/**
+ * Queries the database for attachment rows stuck in "Scanning" status across all attachment
+ * entities (both standalone/composition and inline), and re-emits ScanAttachmentsFile for
+ * each so the normal scanner flow re-runs and moves them to a terminal status.
+ * Exported for unit-testability.
+ */
+async function rescanStuckAttachments() {
+  if (!(cds.env.requires.attachments?.scan ?? true)) return
+  if (!cds.env.requires.attachments?.rescanOnStart) return
+
+  const { db } = cds.services
+  if (!db) return
+
+  const malwareScanner = await cds.connect.to("malwareScanner")
+
+  // Track processed table/prefix combinations to avoid re-querying the same physical rows
+  // when multiple service projections map to the same base entity.
+  const seen = new Set()
+
+  for (const def of Object.values(cds.model.definitions)) {
+    if (def.kind !== "entity") continue
+
+    if (def._attachments?.isAttachmentsEntity) {
+      // Standalone / composition attachment entity: status is the "status" column.
+      // Keys include both "up__ID" and "ID" for composition entities, just "ID" for standalone.
+      const tableKey = _physicalName(def)
+      if (seen.has(tableKey)) continue
+      seen.add(tableKey)
+      await _reemitStuck(malwareScanner, def)
+    } else if (def._attachments?.hasInlineAttachments) {
+      // Inline attachment entity: each prefix has its own "<prefix>_status" column.
+      for (const prefix of def._attachments.inlineAttachmentPrefixes) {
+        const tableKey = `${_physicalName(def)}/${prefix}`
+        if (seen.has(tableKey)) continue
+        seen.add(tableKey)
+        await _reemitStuck(malwareScanner, def, prefix)
+      }
+    }
+  }
+}
+
+/**
+ * Selects rows of the given entity that are stuck in "Scanning" and re-emits
+ * ScanAttachmentsFile for each.
+ * @param {object} malwareScanner - Connected malwareScanner service
+ * @param {import('@sap/cds').entity} entity - The CDS entity definition to query
+ * @param {string} [prefix] - Inline attachment field prefix; undefined for composition-based entities
+ */
+async function _reemitStuck(malwareScanner, entity, prefix) {
+  const statusField = prefix ? `${prefix}_status` : "status"
+  const urlField = prefix ? `${prefix}_url` : undefined
+
+  const keyNames = Object.keys(entity.keys).filter(
+    (k) => k !== "IsActiveEntity",
+  )
+  const cols = urlField ? [...keyNames, urlField] : keyNames
+
+  const rows = await SELECT.from(entity)
+    .columns(cols)
+    .where({ [statusField]: "Scanning" })
+  if (!rows.length) return
+
+  LOG.info(
+    `Re-emitting scan request for ${rows.length} stuck attachment(s) in ${entity.name}${prefix ? `/${prefix}` : ""}`,
+  )
+
+  for (const row of rows) {
+    const keys = Object.fromEntries(keyNames.map((k) => [k, row[k]]))
+    const payload = { target: entity.name, keys }
+    if (prefix) {
+      payload.prefix = prefix
+      payload.url = row[urlField]
+    }
+    await malwareScanner.emit("ScanAttachmentsFile", payload)
+  }
+}
+
+/**
+ * Returns the name of the underlying base entity (the physical DB table) for a given
+ * entity definition, following projection/query chains until a non-query entity is found.
+ * Used to deduplicate the startup scan sweep across service projections of the same table.
+ * @param {import('@sap/cds').entity} def
+ * @returns {string}
+ */
+function _physicalName(def) {
+  const src = (def.query?.SELECT || def.projection)?.from?.ref?.[0]
+  if (src && cds.model.definitions[src])
+    return _physicalName(cds.model.definitions[src])
+  return def.name
+}
+
+module.exports = { rescanStuckAttachments }
+
 /**
  * Unfold the model to add necessary facets for attachments
  * @param {*} csn - CSN model

tests/unit/startupRescan.test.js

@@ -0,0 +1,170 @@
+require("../../lib/csn-runtime-extension")
+const cds = require("@sap/cds")
+const path = require("path")
+const app = path.resolve(__dirname, "../incidents-app")
+cds.test(app)
+
+const { rescanStuckAttachments } = require("../../lib/plugin")
+
+let msEmit
+
+beforeEach(() => {
+  jest.clearAllMocks()
+  cds.env.requires.attachments = { scan: true, rescanOnStart: true }
+  msEmit = jest.fn().mockResolvedValue(undefined)
+  cds.connect.to = jest.fn().mockResolvedValue({ emit: msEmit })
+})
+
+// ---------------------------------------------------------------------------
+// Guard conditions
+// ---------------------------------------------------------------------------
+
+describe("guard conditions", () => {
+  it("does not connect to malwareScanner when scan is disabled", async () => {
+    cds.env.requires.attachments.scan = false
+    await rescanStuckAttachments()
+    expect(cds.connect.to).not.toHaveBeenCalled()
+  })
+
+  it("does not connect when rescanOnStart is false (default)", async () => {
+    cds.env.requires.attachments.rescanOnStart = false
+    await rescanStuckAttachments()
+    expect(cds.connect.to).not.toHaveBeenCalled()
+  })
+
+  it("does not connect when rescanOnStart is absent (defaults to false)", async () => {
+    delete cds.env.requires.attachments.rescanOnStart
+    await rescanStuckAttachments()
+    expect(cds.connect.to).not.toHaveBeenCalled()
+  })
+
+  it("does not connect when attachments config is absent", async () => {
+    delete cds.env.requires.attachments
+    await rescanStuckAttachments()
+    expect(cds.connect.to).not.toHaveBeenCalled()
+    // Restore for subsequent tests
+    cds.env.requires.attachments = { scan: true, rescanOnStart: true }
+  })
+})
+
+// ---------------------------------------------------------------------------
+// Composition-based attachment entity (ProcessorService.Incidents.attachments)
+//   keys: up__ID (parent key) + ID (attachment key)
+//   status column: "status"
+// ---------------------------------------------------------------------------
+
+describe("composition attachment entity", () => {
+  const attachmentsEntity = "sap.capire.incidents.Incidents.attachments"
+  const upID = cds.utils.uuid()
+  const attachmentID = cds.utils.uuid()
+
+  beforeEach(async () => {
+    // Seed a row stuck in "Scanning" directly in the DB
+    await INSERT.into(cds.model.definitions[attachmentsEntity]).entries({
+      up__ID: upID,
+      ID: attachmentID,
+      filename: "stuck.pdf",
+      mimeType: "application/pdf",
+      status: "Scanning",
+    })
+  })
+
+  afterEach(async () => {
+    await DELETE.from(cds.model.definitions[attachmentsEntity]).where({
+      ID: attachmentID,
+    })
+  })
+
+  it("re-emits ScanAttachmentsFile for a row stuck in Scanning", async () => {
+    await rescanStuckAttachments()
+
+    expect(msEmit).toHaveBeenCalledWith("ScanAttachmentsFile", {
+      target: attachmentsEntity,
+      keys: { up__ID: upID, ID: attachmentID },
+    })
+  })
+
+  it("does not re-emit for a row with status Clean", async () => {
+    await UPDATE(cds.model.definitions[attachmentsEntity])
+      .where({ ID: attachmentID })
+      .set({ status: "Clean" })
+
+    await rescanStuckAttachments()
+
+    const callsForRow = (msEmit.mock.calls || []).filter(
+      ([, payload]) => payload?.keys?.ID === attachmentID,
+    )
+    expect(callsForRow).toHaveLength(0)
+  })
+
+  it("does not re-emit for a row with status Unscanned", async () => {
+    await UPDATE(cds.model.definitions[attachmentsEntity])
+      .where({ ID: attachmentID })
+      .set({ status: "Unscanned" })
+
+    await rescanStuckAttachments()
+
+    const callsForRow = (msEmit.mock.calls || []).filter(
+      ([, payload]) => payload?.keys?.ID === attachmentID,
+    )
+    expect(callsForRow).toHaveLength(0)
+  })
+})
+
+// ---------------------------------------------------------------------------
+// Inline attachment entity (ProcessorService.SingleAttachment, prefix "myAttachment")
+//   key: ID
+//   status column: "myAttachment_status", url column: "myAttachment_url"
+// ---------------------------------------------------------------------------
+
+describe("inline attachment entity", () => {
+  // Use the base entity for DB operations; the sweep may emit via any projection of it.
+  const inlineEntityBase = "sap.capire.incidents.SingleAttachment"
+  const rowID = cds.utils.uuid()
+  const objectUrl = "https://objectstore.example.com/myfile.pdf"
+
+  beforeEach(async () => {
+    await INSERT.into(cds.model.definitions[inlineEntityBase]).entries({
+      ID: rowID,
+      name: "Test inline",
+      myAttachment_status: "Scanning",
+      myAttachment_url: objectUrl,
+      myAttachment_mimeType: "application/pdf",
+      myAttachment_filename: "stuck-inline.pdf",
+    })
+  })
+
+  afterEach(async () => {
+    await DELETE.from(cds.model.definitions[inlineEntityBase]).where({
+      ID: rowID,
+    })
+  })
+
+  it("re-emits ScanAttachmentsFile with prefix and url for a stuck inline row", async () => {
+    await rescanStuckAttachments()
+
+    const call = (msEmit.mock.calls || []).find(
+      ([, payload]) => payload?.keys?.ID === rowID,
+    )
+    expect(call).toBeDefined()
+    expect(call[0]).toBe("ScanAttachmentsFile")
+    // target may be any projection of the base entity — just verify the payload shape
+    expect(call[1].target).toMatch(/SingleAttachment$/)
+    expect(call[1].keys).toEqual({ ID: rowID })
+    expect(call[1].prefix).toBe("myAttachment")
+    expect(call[1].url).toBe(objectUrl)
+  })
+
+  it("does not re-emit for a Clean inline row", async () => {
+    await UPDATE(cds.model.definitions[inlineEntityBase])
+      .where({ ID: rowID })
+      .set({ myAttachment_status: "Clean" })
+
+    await rescanStuckAttachments()
+
+    const callsForRow = (msEmit.mock.calls || []).filter(
+      ([, payload]) => payload?.keys?.ID === rowID,
+    )
+    expect(callsForRow).toHaveLength(0)
+  })
+})