fix(storage): abort backend multipart when reaping abandoned sys_upload_session (#2970)

.changeset/upload-session-multipart-abort-guard.md

@@ -0,0 +1,20 @@
+---
+'@objectstack/service-storage': patch
+---
+
+fix(storage): abort the backend multipart upload when reaping an abandoned sys_upload_session (#2970)
+
+The `sys_upload_session` lifecycle (added in #2984) reaps abandoned/terminal
+chunked-upload session ROWS, but not the underlying backend multipart upload —
+on S3 an initiated-but-not-completed multipart keeps its already-uploaded parts
+billable and invisible to normal listing until an explicit
+`AbortMultipartUpload`, so reaping only the row stranded them (with
+`backend_upload_id`, the sole pointer, gone).
+
+`createUploadSessionReapGuard` registers a `LifecycleReapGuard` on
+`sys_upload_session` that aborts the backend multipart before the row is
+deleted: it skips `completed` sessions (their multipart already became a real
+object — an abort would `NoSuchUpload`-error), re-seeds the S3 adapter's
+`uploadId → key` map from the row (a cold sweep lacks the live in-process map),
+and vetoes (keeps the row for retry) on abort failure so the pointer survives.
+The local adapter's parts directory is removed the same way.

packages/dogfood/test/attachments-permission-matrix.dogfood.test.ts

@@ -476,19 +476,34 @@ describe('attachments permission matrix (#2755)', () => {
       expect(await ql.findOne('sys_file', { where: { id: data.fileId }, context: SYS })).toBeNull();
     });
 
-    it('(item 4) abandoned sys_upload_session rows are reaped past their expiry window', async () => {
-      // Initiate a chunked upload (creates a sys_upload_session) but never
-      // complete it.
+    it('(item 4 + multipart-abort guard) an abandoned chunked upload is reaped AND its uploaded parts are aborted', async () => {
+      // Initiate a chunked upload (creates a sys_upload_session) and upload one
+      // chunk — but never complete it. The chunk lands as a part on disk.
       const init = await stack.api('/storage/upload/chunked', {
         method: 'POST',
         headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${memberATok}` },
         body: JSON.stringify({ filename: 'big.bin', mimeType: 'application/octet-stream', totalSize: 10_485_760 }),
       });
       expect(init.status).toBe(200);
-      const { uploadId } = ((await init.json()) as any).data;
+      const { uploadId, resumeToken } = ((await init.json()) as any).data;
       const session = await ql.findOne('sys_upload_session', { where: { id: uploadId }, context: SYS });
       expect(session?.id, 'session row created').toBeTruthy();
 
+      const chunkRes = await stack.api(`/storage/upload/chunked/${uploadId}/chunk/0`, {
+        method: 'PUT',
+        headers: {
+          Authorization: `Bearer ${memberATok}`,
+          'Content-Type': 'application/octet-stream',
+          'x-resume-token': resumeToken,
+        },
+        body: Buffer.from('a partial chunk of bytes'),
+      });
+      expect(chunkRes.status, await chunkRes.clone().text()).toBeLessThan(300);
+
+      // The local adapter stores parts under `<rootDir>/.parts/<uploadId>/`.
+      const partsDir = join(rootDir, '.parts', String(uploadId));
+      await expect(fs.access(partsDir), 'part exists before the sweep').resolves.toBeUndefined();
+
       // Backdate expires_at past the 1d TTL grace.
       await ql.update(
         'sys_upload_session',
@@ -498,7 +513,9 @@ describe('attachments permission matrix (#2755)', () => {
 
       const report = await lifecycle.sweep();
       expect(report.errors, JSON.stringify(report.errors)).toEqual([]);
+      // Row gone AND the backend multipart parts aborted (dir removed).
       expect(await ql.findOne('sys_upload_session', { where: { id: uploadId }, context: SYS })).toBeNull();
+      await expect(fs.access(partsDir), 'parts aborted by the reap guard').rejects.toThrow();
     });
   });
 });

packages/services/service-storage/src/attachment-lifecycle.test.ts

@@ -4,6 +4,7 @@ import { describe, it, expect, vi } from 'vitest';
 import {
   installAttachmentLifecycleHooks,
   createSysFileReapGuard,
+  createUploadSessionReapGuard,
   type AttachmentLifecycleEngine,
 } from './attachment-lifecycle.js';
 
@@ -247,3 +248,93 @@ describe('createSysFileReapGuard', () => {
     expect(confirmed).toEqual([]);
   });
 });
+
+describe('createUploadSessionReapGuard', () => {
+  /** Swappable-style storage fake: `getInner()` exposes an S3-like inner with
+   * `setUploadKey`; `abortChunkedUpload` is forwarded. */
+  const s3Storage = (abortImpl?: () => Promise<void>) => {
+    const inner = {
+      setUploadKey: vi.fn((_id: string, _key: string) => {}),
+    };
+    return {
+      getInner: () => inner,
+      abortChunkedUpload: vi.fn(abortImpl ?? (async () => {})),
+      _inner: inner,
+    } as any;
+  };
+
+  it('aborts the backend multipart (re-seeding the key) then reaps an abandoned session', async () => {
+    const s = s3Storage();
+    const guard = createUploadSessionReapGuard(() => s, silentLogger());
+
+    const confirmed = await guard('sys_upload_session', [
+      { id: 'u1', backend_upload_id: 'mp-1', key: 'attachments/u1.bin', status: 'in_progress' },
+    ]);
+
+    expect(s._inner.setUploadKey).toHaveBeenCalledWith('mp-1', 'attachments/u1.bin');
+    expect(s.abortChunkedUpload).toHaveBeenCalledWith('mp-1');
+    expect(confirmed).toEqual(['u1']);
+  });
+
+  it('does NOT abort a completed session (its multipart is already an object) — just reaps the row', async () => {
+    const s = s3Storage();
+    const guard = createUploadSessionReapGuard(() => s, silentLogger());
+
+    const confirmed = await guard('sys_upload_session', [
+      { id: 'u2', backend_upload_id: 'mp-2', key: 'attachments/u2.bin', status: 'completed' },
+    ]);
+
+    expect(s.abortChunkedUpload).not.toHaveBeenCalled();
+    expect(confirmed).toEqual(['u2']);
+  });
+
+  it('reaps a session with no backend_upload_id without calling abort', async () => {
+    const s = s3Storage();
+    const guard = createUploadSessionReapGuard(() => s, silentLogger());
+
+    const confirmed = await guard('sys_upload_session', [
+      { id: 'u3', key: 'attachments/u3.bin', status: 'expired' },
+    ]);
+
+    expect(s.abortChunkedUpload).not.toHaveBeenCalled();
+    expect(confirmed).toEqual(['u3']);
+  });
+
+  it('VETOES on abort failure so backend_upload_id survives for the retry', async () => {
+    const s = s3Storage(async () => {
+      throw new Error('S3 abort transient failure');
+    });
+    const logger = silentLogger();
+    const guard = createUploadSessionReapGuard(() => s, logger);
+
+    const confirmed = await guard('sys_upload_session', [
+      { id: 'u4', backend_upload_id: 'mp-4', key: 'attachments/u4.bin', status: 'failed' },
+    ]);
+
+    expect(confirmed).toEqual([]); // kept
+    expect(logger.warn).toHaveBeenCalled();
+  });
+
+  it('works with a local-style adapter (no setUploadKey / getInner) — aborts by id', async () => {
+    const local = { abortChunkedUpload: vi.fn(async () => {}) } as any;
+    const guard = createUploadSessionReapGuard(() => local, silentLogger());
+
+    const confirmed = await guard('sys_upload_session', [
+      { id: 'u5', backend_upload_id: 'local-5', key: 'attachments/u5.bin', status: 'expired' },
+    ]);
+
+    expect(local.abortChunkedUpload).toHaveBeenCalledWith('local-5');
+    expect(confirmed).toEqual(['u5']);
+  });
+
+  it('reaps the row when the adapter cannot abort at all', async () => {
+    const noAbort = {} as any;
+    const guard = createUploadSessionReapGuard(() => noAbort, silentLogger());
+
+    const confirmed = await guard('sys_upload_session', [
+      { id: 'u6', backend_upload_id: 'mp-6', key: 'k', status: 'in_progress' },
+    ]);
+
+    expect(confirmed).toEqual(['u6']);
+  });
+});

packages/services/service-storage/src/attachment-lifecycle.ts

@@ -259,3 +259,67 @@ export function createSysFileReapGuard(
     return confirmed;
   };
 }
+
+/**
+ * The `sys_upload_session` reap guard (#2970 sub-follow-up). The
+ * LifecycleService reaps abandoned/terminal chunked-upload session ROWS by the
+ * declared TTL (`expires_at`) / retention (terminal statuses); this guard
+ * aborts the underlying BACKEND multipart upload before the row is deleted, so
+ * a session's already-uploaded parts don't leak. On S3 an initiated-but-not-
+ * completed multipart keeps its parts billable and invisible to normal listing
+ * until an explicit AbortMultipartUpload — reaping only the row would strand
+ * them, with `backend_upload_id` (the sole pointer) gone.
+ *
+ *  - `completed`: the multipart was already finalized into a real object —
+ *    nothing to abort (an abort would `NoSuchUpload`-error and wedge the reap).
+ *    Confirm the row.
+ *  - no `backend_upload_id`, or an adapter without `abortChunkedUpload`:
+ *    nothing to abort. Confirm.
+ *  - otherwise (in_progress / failed / expired with a backend upload): abort
+ *    the backend multipart, re-seeding the S3 `uploadId → key` map from the row
+ *    first (a cold sweep lacks the live in-process map; a no-op for adapters
+ *    that don't track keys, e.g. local). Confirm on success; VETO on failure
+ *    so the row — the only pointer to the leaked multipart — is retried.
+ */
+export function createUploadSessionReapGuard(
+  getStorage: () => IStorageService | null | undefined,
+  logger: AttachmentLifecycleLogger,
+): (object: string, rows: Array<Record<string, unknown>>) => Promise<Array<string | number>> {
+  return async (_object, rows) => {
+    const confirmed: Array<string | number> = [];
+    const storage = getStorage();
+    for (const row of rows) {
+      const id = row?.id as string | number | undefined;
+      if (id === undefined || id === null) continue;
+
+      const backendId = typeof row.backend_upload_id === 'string' ? row.backend_upload_id : '';
+      // Nothing to abort → just reap the row: no backend multipart, an already
+      // -completed upload (its parts became an object), or an adapter that
+      // can't abort.
+      if (!backendId || row.status === 'completed' || !storage || typeof storage.abortChunkedUpload !== 'function') {
+        confirmed.push(id);
+        continue;
+      }
+
+      try {
+        // A cold sweep runs long after the live session, so the S3 adapter's
+        // in-process `uploadId → key` map (populated by `setUploadKey` during
+        // upload) is empty — re-seed it from the row so the abort can resolve
+        // the S3 key. `setUploadKey` is S3-specific and not forwarded by the
+        // swappable proxy, so reach the inner adapter; a no-op for `local`.
+        const inner: any = typeof (storage as any).getInner === 'function' ? (storage as any).getInner() : storage;
+        if (typeof row.key === 'string' && row.key && typeof inner?.setUploadKey === 'function') {
+          inner.setUploadKey(backendId, row.key);
+        }
+        await storage.abortChunkedUpload(backendId);
+        confirmed.push(id);
+      } catch (err) {
+        logger.warn(
+          `[storage] reap guard: multipart abort failed for sys_upload_session ${id} (${(err as Error)?.message ?? err}); retrying next sweep`,
+        );
+        // veto — keep the row so `backend_upload_id` survives for the retry.
+      }
+    }
+    return confirmed;
+  };
+}

packages/services/service-storage/src/index.ts

@@ -12,7 +12,7 @@ export type { FileRecord, UploadSessionRecord } from './metadata-store.js';
 export { registerStorageRoutes } from './storage-routes.js';
 export type { StorageRoutesOptions, FileReadVerdict } from './storage-routes.js';
 export { SystemFile, SystemUploadSession } from './objects/index.js';
-export { installAttachmentLifecycleHooks, createSysFileReapGuard } from './attachment-lifecycle.js';
+export { installAttachmentLifecycleHooks, createSysFileReapGuard, createUploadSessionReapGuard } from './attachment-lifecycle.js';
 export type { AttachmentLifecycleEngine, AttachmentLifecycleLogger } from './attachment-lifecycle.js';
 export { installAttachmentAccessHooks, installAttachmentReadVisibility } from './attachment-access-hooks.js';
 export type { AttachmentSharingLike } from './attachment-access-hooks.js';

packages/services/service-storage/src/storage-service-plugin.test.ts

@@ -229,9 +229,9 @@ describe('StorageServicePlugin: sys_file orphan lifecycle wiring (#2755)', () =>
       'beforeDelete',
       'beforeInsert',
     ]);
-    expect(guards).toHaveLength(1);
-    expect(guards[0].object).toBe('sys_file');
-    expect(typeof guards[0].guard).toBe('function');
+    // Two reap guards: sys_file (#2755) + sys_upload_session multipart-abort (#2970).
+    expect(guards.map((g) => g.object).sort()).toEqual(['sys_file', 'sys_upload_session']);
+    expect(guards.every((g) => typeof g.guard === 'function')).toBe(true);
     // Read-visibility middleware registered on sys_attachment (#2970 item 1).
     expect(middlewares).toEqual([{ object: 'sys_attachment' }]);
   });

packages/services/service-storage/src/storage-service-plugin.ts

@@ -16,7 +16,7 @@ import { StorageMetadataStore } from './metadata-store.js';
 import type { FileRecord } from './metadata-store.js';
 import { registerStorageRoutes } from './storage-routes.js';
 import type { FileReadVerdict } from './storage-routes.js';
-import { installAttachmentLifecycleHooks, createSysFileReapGuard } from './attachment-lifecycle.js';
+import { installAttachmentLifecycleHooks, createSysFileReapGuard, createUploadSessionReapGuard } from './attachment-lifecycle.js';
 import { installAttachmentAccessHooks, installAttachmentReadVisibility } from './attachment-access-hooks.js';
 import { SystemFile, SystemUploadSession } from './objects/index.js';
 // ADR-0052 §3 ownership: `sys_attachment` (a file↔record link) belongs with the
@@ -228,7 +228,13 @@ export class StorageServicePlugin implements Plugin {
               'sys_file',
               createSysFileReapGuard(engine as any, () => this.storage, ctx.logger),
             );
-            ctx.logger.info('StorageServicePlugin: sys_file reap guard registered with the lifecycle service');
+            // Abort the backend multipart upload before an abandoned/terminal
+            // sys_upload_session row is reaped, so its parts don't leak (#2970).
+            lifecycle.registerReapGuard(
+              'sys_upload_session',
+              createUploadSessionReapGuard(() => this.storage, ctx.logger),
+            );
+            ctx.logger.info('StorageServicePlugin: sys_file + sys_upload_session reap guards registered with the lifecycle service');
           }
         } catch {
           // lifecycle service absent (bare kernel) — the sys_file lifecycle