fix(engine): serialize blob GC unlinks with publish and persist resources after the blob row

Author: Zerlight

packages/host/engine/src/__tests__/attachment-gc.test.ts

@@ -2,7 +2,7 @@ import { createHash } from 'node:crypto';
 import { mkdtemp, rm, stat } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
-import type { AttachmentId, BlobId, UploadLease } from '@linkcode/schema';
+import type { BlobId, UploadLease } from '@linkcode/schema';
 import {
   AttachmentIdSchema,
   ConversationOperationSchema,
@@ -183,6 +183,27 @@ describe('AttachmentGc', () => {
     await expect(stat(join(f.blobs.pathOf(kept), '..', '..', '..', 'tmp'))).rejects.toMatchObject({
       code: 'ENOENT',
     });
-    expect(f.store.getAttachment('att-kept' as AttachmentId)).resolves.toBeDefined();
+    await expect(
+      f.store.getAttachment(AttachmentIdSchema.parse('att-kept')),
+    ).resolves.toBeDefined();
+  });
+
+  it('does not unlink a blob that is re-committed after the reaper transaction', async () => {
+    const f = await fixture();
+    const blobId = await f.commit('att-old', 'shared bytes');
+    f.clock.now += ATTACHMENT_GC_GRACE_MS + 1;
+
+    const originalSweep = f.store.sweep.bind(f.store);
+    f.store.sweep = async (window) => {
+      const doomed = await originalSweep(window);
+      await f.commit('att-new', 'shared bytes');
+      return doomed;
+    };
+
+    expect(await f.gc.sweep()).toEqual({ removedBlobs: [] });
+    expect(await f.blobs.stat(blobId)).toBeDefined();
+    expect(await f.store.getAttachment(AttachmentIdSchema.parse('att-new'))).toMatchObject({
+      blobId,
+    });
   });
 });

packages/host/engine/src/__tests__/blob-store.test.ts

@@ -110,10 +110,13 @@ describe('mime sniff', () => {
     expect(sniffImageMimeType(new Uint8Array(0))).toBeUndefined();
   });
 
-  it('holds image declarations to their bytes and trusts the rest', () => {
+  it('holds sniffable image declarations to their bytes and trusts the rest', () => {
     expect(declaredMimeTypeMatches('image/png', png)).toBe(true);
     expect(declaredMimeTypeMatches('image/jpeg', png)).toBe(false);
-    expect(declaredMimeTypeMatches('image/svg+xml', Buffer.from('<svg/>'))).toBe(false);
+    expect(declaredMimeTypeMatches('image/jpeg', Buffer.from('<svg/>'))).toBe(false);
+    expect(declaredMimeTypeMatches('image/svg+xml', Buffer.from('<svg/>'))).toBe(true);
+    expect(declaredMimeTypeMatches('image/svg+xml', png)).toBe(false);
+    expect(declaredMimeTypeMatches('image/heic', Buffer.from('ftypheic'))).toBe(true);
     expect(declaredMimeTypeMatches('application/pdf', Buffer.from('%PDF-1.7'))).toBe(true);
     expect(declaredMimeTypeMatches('text/plain', png)).toBe(true);
   });

packages/host/engine/src/__tests__/engine-resources.test.ts

@@ -140,6 +140,36 @@ describe('engine session resources', () => {
     await h.inject({ kind: 'resource.list', clientReqId: 'list-after-mismatch', sessionId });
     expect(listedResources(h.sent, 'list-after-mismatch')).toEqual([]);
 
+    await h.inject({
+      kind: 'resource.source.upload',
+      clientReqId: 'upload-svg',
+      sessionId,
+      name: 'icon.svg',
+      mimeType: 'image/svg+xml',
+      data: Buffer.from('<svg xmlns="http://www.w3.org/2000/svg"/>').toString('base64'),
+    });
+    await vi.waitFor(() => {
+      expect(h.sent).toContainEqual(
+        expect.objectContaining({ kind: 'resource.uploaded', replyTo: 'upload-svg' }),
+      );
+    });
+    await h.inject({ kind: 'resource.list', clientReqId: 'list-svg', sessionId });
+    expect(listedResources(h.sent, 'list-svg')).toEqual([
+      expect.objectContaining({
+        name: 'icon.svg',
+        status: 'ready',
+        mimeType: 'image/svg+xml',
+      }),
+    ]);
+    await h.inject({
+      kind: 'resource.remove',
+      clientReqId: 'remove-svg',
+      resourceId: listedResources(h.sent, 'list-svg')[0].resourceId,
+    });
+    await vi.waitFor(() => {
+      expect(h.sent).toContainEqual({ kind: 'request.succeeded', replyTo: 'remove-svg' });
+    });
+
     await h.inject({
       kind: 'resource.source.upload',
       clientReqId: 'upload-for-delete',

packages/host/engine/src/attachment/blob-store.ts

@@ -149,7 +149,16 @@ export class FsBlobStore implements BlobStore {
       await rename(stagingPath, dest);
     } catch (error) {
       if ((await this.stat(blobId)) === undefined) throw error;
+      // Dest existing is not proof of identical bytes: Windows rename does not replace a
+      // read-only dest, so a truncated or bitrot file would otherwise count as success.
+      const existing = await sha256OfFile(dest);
+      const expected = blobId.slice(BLOB_ID_PREFIX.length);
       await rm(stagingPath, { force: true });
+      if (existing !== expected) {
+        throw new BlobIntegrityError('Existing blob bytes do not match the declared SHA-256', {
+          cause: error,
+        });
+      }
     }
   }
 }

packages/host/engine/src/attachment/gc.ts

@@ -2,6 +2,7 @@ import type { BlobId } from '@linkcode/schema';
 import { Effect, Schedule } from 'effect';
 import type { AttachmentStore } from './attachment-store';
 import type { BlobStore } from './blob-store';
+import { AttachmentIoMutex } from './io-mutex';
 
 /** A draft's upload outlives any composer session, not a forgotten one. */
 export const UPLOAD_LEASE_TTL_MS = 24 * 60 * 60 * 1000;
@@ -19,41 +20,53 @@ export class AttachmentGc {
     private readonly store: AttachmentStore,
     private readonly blobs: BlobStore,
     private readonly clock: () => number = Date.now,
+    private readonly io: AttachmentIoMutex = new AttachmentIoMutex(),
   ) {}
 
   /** A failed unlink leaves an orphan file the next boot sweep removes. */
   async sweep(): Promise<AttachmentGcReport> {
+    return this.io.run(() => this.sweepBody());
+  }
+
+  private async sweepBody(): Promise<AttachmentGcReport> {
     const now = this.clock();
     const removedBlobs = await this.store.sweep({
       now,
       graceBefore: now - ATTACHMENT_GC_GRACE_MS,
     });
+    const unlinked: BlobId[] = [];
     for (let i = 0, len = removedBlobs.length; i < len; i++) {
+      const blobId = removedBlobs[i];
       // eslint-disable-next-line no-await-in-loop -- sequential unlinks of a short list
-      await this.blobs.delete(removedBlobs[i]);
+      if (await this.store.getBlob(blobId)) continue;
+      // eslint-disable-next-line no-await-in-loop -- same
+      await this.blobs.delete(blobId);
+      unlinked.push(blobId);
     }
-    return { removedBlobs };
+    return { removedBlobs: unlinked };
   }
 
   /**
    * Run before requests are accepted: no upload survives a restart, so staging goes wholesale;
    * then a sweep; then bytes with no blob row (a crash between publish and the row insert).
-   * Safe only while no commit can be publishing — that is what the boot ordering guarantees.
+   * File unlinks share the publish mutex: they are only safe while no commit can be publishing.
    */
   async bootSweep(): Promise<AttachmentGcReport> {
-    await this.blobs.purgeStaging();
-    const report = await this.sweep();
-    const onDisk = await this.blobs.list();
-    const orphans: BlobId[] = [];
-    for (let i = 0, len = onDisk.length; i < len; i++) {
-      const blobId = onDisk[i];
-      // eslint-disable-next-line no-await-in-loop -- one row lookup per file on disk
-      if (await this.store.getBlob(blobId)) continue;
-      // eslint-disable-next-line no-await-in-loop -- same
-      await this.blobs.delete(blobId);
-      orphans.push(blobId);
-    }
-    return { removedBlobs: [...report.removedBlobs, ...orphans] };
+    return this.io.run(async () => {
+      await this.blobs.purgeStaging();
+      const report = await this.sweepBody();
+      const onDisk = await this.blobs.list();
+      const orphans: BlobId[] = [];
+      for (let i = 0, len = onDisk.length; i < len; i++) {
+        const blobId = onDisk[i];
+        // eslint-disable-next-line no-await-in-loop -- one row lookup per file on disk
+        if (await this.store.getBlob(blobId)) continue;
+        // eslint-disable-next-line no-await-in-loop -- same
+        await this.blobs.delete(blobId);
+        orphans.push(blobId);
+      }
+      return { removedBlobs: [...report.removedBlobs, ...orphans] };
+    });
   }
 
   /** One sweep per interval until interrupted; a failed sweep is logged and the cadence goes on. */

packages/host/engine/src/attachment/io-mutex.ts

@@ -0,0 +1,18 @@
+import { noop } from 'foxts/noop';
+
+/**
+ * Serializes blob publish + row insert with GC unlinks. A doomed id can grow a new row between
+ * the reaper transaction and the unlink; those two must not overlap.
+ */
+export class AttachmentIoMutex {
+  private tail: Promise<void> = Promise.resolve();
+
+  run<T>(work: () => Promise<T>): Promise<T> {
+    const previous = this.tail;
+    let release: () => void = noop;
+    this.tail = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    return previous.catch(noop).then(work).finally(release);
+  }
+}

packages/host/engine/src/attachment/mime-sniff.ts

@@ -23,9 +23,19 @@ export function sniffImageMimeType(head: Uint8Array): SupportedAttachmentImageMi
   return undefined;
 }
 
-/** A declared `image/*` type must match its bytes — model APIs refuse the mismatch later and
- * less legibly. Other declarations have no reliable sniff and are trusted. */
+const SNIFFABLE_IMAGE_TYPES = new Set<string>([
+  'image/jpeg',
+  'image/png',
+  'image/gif',
+  'image/webp',
+]);
+
+/** A declared sniffable `image/*` type must match its bytes — model APIs refuse the mismatch
+ * later and less legibly. Other declarations (svg, heic, pdf, …) have no reliable sniff here
+ * and are trusted. */
 export function declaredMimeTypeMatches(declared: string, head: Uint8Array): boolean {
   if (!declared.startsWith('image/')) return true;
-  return sniffImageMimeType(head) === declared;
+  const sniffed = sniffImageMimeType(head);
+  if (sniffed !== undefined) return sniffed === declared;
+  return !SNIFFABLE_IMAGE_TYPES.has(declared);
 }

packages/host/engine/src/engine.ts

@@ -19,6 +19,7 @@ import { ManagedAssetService } from './asset/service';
 import { InMemoryAttachmentStore } from './attachment/attachment-store';
 import { FsBlobStore } from './attachment/blob-store';
 import { AttachmentGc } from './attachment/gc';
+import { AttachmentIoMutex } from './attachment/io-mutex';
 import {
   InMemoryLoopStore,
   InMemoryScheduleStore,
@@ -128,7 +129,8 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* (
         ? resourceStore.referencedAttachmentIds()
         : []),
     ]);
-  const attachmentGc = new AttachmentGc(attachmentStore, blobStore);
+  const attachmentIo = new AttachmentIoMutex();
+  const attachmentGc = new AttachmentGc(attachmentStore, blobStore, Date.now, attachmentIo);
   const resources = new ResourceService(
     transport,
     resourceStore,
@@ -137,6 +139,7 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* (
     fileHost,
     blobStore,
     attachmentStore,
+    attachmentIo,
   );
   const plugins = new PluginService(deps.pluginFactory ?? createPluginProviderAdapter);
   const translator = deps.translator;

packages/host/engine/src/resource/service.ts

@@ -16,6 +16,7 @@ import { Effect } from 'effect';
 import { noop } from 'foxts/noop';
 import type { AttachmentStore } from '../attachment/attachment-store';
 import type { BlobStore } from '../attachment/blob-store';
+import { AttachmentIoMutex } from '../attachment/io-mutex';
 import { declaredMimeTypeMatches } from '../attachment/mime-sniff';
 import { OperationError, RequestError } from '../failure';
 import type { FileHostService } from '../preview/file-host-service';
@@ -67,6 +68,7 @@ export class ResourceService {
     private readonly fileHost: FileHostService,
     private readonly blobs: BlobStore,
     private readonly attachments: AttachmentStore,
+    private readonly io: AttachmentIoMutex = new AttachmentIoMutex(),
   ) {}
 
   list(sessionId: SessionId): Effect.Effect<SessionResource[], OperationError> {
@@ -79,7 +81,7 @@ export class ResourceService {
     mimeType: string | undefined,
     data: string,
   ): Effect.Effect<SessionResource, OperationError | RequestError> {
-    const { attachments, blobs, records, transport } = this;
+    const { attachments, blobs, io, records, transport } = this;
     return Effect.gen({ self: this }, function* () {
       if (!records.has(sessionId)) {
         return yield* new RequestError({ code: 'not_found', message: 'Session not found' });
@@ -102,59 +104,69 @@ export class ResourceService {
       const sha256 = createHash('sha256').update(bytes).digest('hex');
       const blobId = blobIdFromSha256(sha256);
       const now = Date.now();
-      // The harness reads the blob's own path: immutable, shared, mode 0444.
-      let resource: SessionResource = {
-        resourceId,
-        sessionId,
-        direction: 'source',
-        name,
-        kind: classify(name, mimeType),
-        status: 'processing',
-        locator: { type: 'managed-file', path: blobs.pathOf(blobId) },
-        attachmentId,
-        mimeType,
-        sizeBytes: bytes.byteLength,
-        createdAt: now,
-        updatedAt: now,
-      };
-      yield* this.run('save', () => this.store.save(resource));
-      transport.send(createWireMessage({ kind: 'resource.changed', resource }));
+      const kind = classify(name, mimeType);
+      const locator = { type: 'managed-file' as const, path: blobs.pathOf(blobId) };
       const written = yield* Effect.tryPromise({
         async try() {
-          const stage = await blobs.stage(resourceId);
-          try {
-            await stage.write(0, bytes);
-            await stage.commit({ sha256, sizeBytes: bytes.byteLength });
-          } catch (error) {
-            await stage.abort().catch(noop);
-            throw error;
-          }
-          await attachments.commitAttachment({
-            blob: { blobId, sizeBytes: bytes.byteLength, createdAt: now },
-            attachment: {
-              attachmentId,
-              kind: resource.kind,
-              name,
-              mimeType: mimeType ?? 'application/octet-stream',
-              sizeBytes: bytes.byteLength,
-              metadata: {},
-              createdAt: now,
-            },
+          await io.run(async () => {
+            const stage = await blobs.stage(resourceId);
+            try {
+              await stage.write(0, bytes);
+              await stage.commit({ sha256, sizeBytes: bytes.byteLength });
+              await attachments.commitAttachment({
+                blob: { blobId, sizeBytes: bytes.byteLength, createdAt: now },
+                attachment: {
+                  attachmentId,
+                  kind,
+                  name,
+                  mimeType: mimeType ?? 'application/octet-stream',
+                  sizeBytes: bytes.byteLength,
+                  metadata: {},
+                  createdAt: now,
+                },
+              });
+            } catch (error) {
+              await stage.abort().catch(noop);
+              await blobs.delete(blobId);
+              throw error;
+            }
           });
         },
         catch: (cause) => cause,
       }).pipe(
         Effect.as(true),
         Effect.catch(() => Effect.succeed(false)),
       );
-      resource = written
-        ? { ...resource, status: 'ready', updatedAt: Date.now() }
-        : {
-            ...resource,
-            status: 'failed',
-            error: 'Failed to persist uploaded resource',
-            updatedAt: Date.now(),
-          };
+      if (!written) {
+        return {
+          resourceId,
+          sessionId,
+          direction: 'source',
+          name,
+          kind,
+          status: 'failed',
+          locator,
+          error: 'Failed to persist uploaded resource',
+          mimeType,
+          sizeBytes: bytes.byteLength,
+          createdAt: now,
+          updatedAt: Date.now(),
+        };
+      }
+      const resource: SessionResource = {
+        resourceId,
+        sessionId,
+        direction: 'source',
+        name,
+        kind,
+        status: 'ready',
+        locator,
+        attachmentId,
+        mimeType,
+        sizeBytes: bytes.byteLength,
+        createdAt: now,
+        updatedAt: Date.now(),
+      };
       yield* this.run('save', () => this.store.save(resource));
       transport.send(createWireMessage({ kind: 'resource.changed', resource }));
       return resource;