fix: Bug - Sending files and attachment issues

Hydrate playable media after disk receive, relay file-announce to sibling devices via account_sync, bind DM attachments to pre-allocated message ids, and improve gallery retry/cancel UX with bounded parallel auto-downloads. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-14 13:05:23 +02:00
parent bb0ac930ad
commit fa45052432
30 changed files with 785 additions and 71 deletions
@@ -107,14 +107,14 @@ Concurrent triggers (file-announce, message sync, peer connect) can race to requ

 - **Requester:** `requestFromAnyPeer` marks the request pending *synchronously* before any async work, so the manager's `hasPendingRequest` gate closes the double-request race window.
 - **Sender:** `handleFileRequest` / `fulfillRequestWithFile` track active outbound streams per `(messageId, fileId, peerId)` and ignore duplicate requests while a stream is in flight. A fresh `file-request` clears any earlier `file-cancel` marker from that peer.
- **Receiver:** chunk buffers are dense (`Array.from({ length: total })`, never sparse `new Array(total)`); a chunk index that is already buffered is ignored entirely and never counts toward `receivedBytes`; a transfer finalizes only when *every* chunk index is present — byte counters are never a substitute for chunk completeness. Assembly state is released only after the attachment is marked `available`, and chunks arriving for an already-available attachment are dropped. When the active store supports streaming (`canStreamToDisk`), **all** persistable downloads append directly to disk — metadata `filePath` does not force an in-memory assembly fallback. Disk-streamed receives decode each chunk once, append bytes through Electron IPC (`append-file-bytes`), and acknowledge the sender with `file-chunk-ack` so only one chunk is in flight at a time (preventing unbounded base64 retention in the renderer). Completed media stays on `savedPath` until inline display hydration runs on demand.
+- **Receiver:** chunk buffers are dense (`Array.from({ length: total })`, never sparse `new Array(total)`); a chunk index that is already buffered is ignored entirely and never counts toward `receivedBytes`; a transfer finalizes only when *every* chunk index is present — byte counters are never a substitute for chunk completeness. Assembly state is released only after the attachment is marked `available`, and chunks arriving for an already-available attachment are dropped. When the active store supports streaming (`canStreamToDisk`), **all** persistable downloads append directly to disk — metadata `filePath` does not force an in-memory assembly fallback. Disk-streamed receives decode each chunk once, append bytes through Electron IPC (`append-file-bytes`), and acknowledge the sender with `file-chunk-ack` so only one chunk is in flight at a time (preventing unbounded base64 retention in the renderer). Completed **images** stay on `savedPath` until inline display hydration runs on demand; completed **audio/video** immediately resolve a playable URL via `attachmentStorage.getFileUrl(savedPath)` (Electron/Capacitor) or `ensureInlineDisplayObjectUrl` in the browser.
 - **Sender:** after each `file-chunk` the transport awaits the matching `file-chunk-ack` before sending the next chunk, in addition to data-channel bufferedAmount back-pressure.

 ### Failure handling

 If the sender cannot find the file, it replies with `file-not-found`. The transfer service then tries the next connected peer that has announced the same attachment. Either side can send `file-cancel` to abort a transfer in progress.

-Peers that finish downloading a file re-announce it and register themselves as mirror hosts. New download requests prefer mirror hosts over the original uploader so the sharer's device is not the only upload source. Repeat `file-announce` events for already-known attachments update the host list but do not re-trigger auto-download.
+Peers that finish downloading a file re-announce it and register themselves as mirror hosts. New download requests prefer mirror hosts over the original uploader so the sharer's device is not the only upload source. Repeat `file-announce` events for already-known attachments update the host list but do not re-trigger auto-download. Outgoing `file-announce` broadcasts are also relayed to sibling devices through `account_sync` (see `infrastructure/realtime/account-sync/account-sync.rules.ts`) so a second client of the same user learns attachment metadata even when it cannot P2P to itself.

 ```mermaid
 sequenceDiagram
@@ -144,7 +144,7 @@ When the user navigates to a room, the manager watches the route and decides whi

 The decision lives in `shouldAutoRequestWhenWatched()` which calls `isAttachmentMedia()` and checks against `MAX_AUTO_SAVE_SIZE_BYTES`.

-Direct-message routes (`/dm/:conversationId` and `/pm/:conversationId`) are treated as watched attachment containers named `direct-message:<conversationId>`, so image/video metadata announced for the visible conversation is eligible for the same automatic request path as server-room media.
+Direct-message routes (`/dm/:conversationId` and `/pm/:conversationId`) are treated as watched attachment containers named `direct-message:<conversationId>`, so image/video metadata announced for the visible conversation is eligible for the same automatic request path as server-room media. Auto-download work fans out with bounded concurrency (`ATTACHMENT_AUTO_DOWNLOAD_CONCURRENCY`, default 3 files at a time per watched room) so multiple pending files can progress in parallel without removing the per-file chunk-ack memory safety invariant.

 Browser chat views render audio/video larger than 50 MB with the same generic file interface as other downloads, even after the bytes are available. Attachments with audio/video MIME types that Chromium reports as unsupported also use the generic file interface instead of a broken native player.

@@ -135,6 +135,12 @@ export class AttachmentFacade {
    return this.manager.cancelRequest(...args);
  }

+  hasPendingRequest(
+    ...args: Parameters<AttachmentManagerService['hasPendingRequest']>
+  ): ReturnType<AttachmentManagerService['hasPendingRequest']> {
+    return this.manager.hasPendingRequest(...args);
+  }
+
  handleFileCancel(
    ...args: Parameters<AttachmentManagerService['handleFileCancel']>
  ): ReturnType<AttachmentManagerService['handleFileCancel']> {
@@ -16,6 +16,7 @@ import {
  isDirectMessageAttachmentRoomId,
  shouldAutoRequestWhenWatched
 } from '../../domain/logic/attachment.logic';
+import { ATTACHMENT_AUTO_DOWNLOAD_CONCURRENCY, runTasksWithBoundedConcurrency } from '../../domain/logic/attachment-autodownload-concurrency.rules';
 import type { Attachment, AttachmentMeta } from '../../domain/models/attachment.model';
 import type {
  FileAnnouncePayload,
@@ -153,6 +154,10 @@ export class AttachmentManagerService {
    return this.transfer.requestImageFromAnyPeer(messageId, attachment);
  }

+  hasPendingRequest(messageId: string, attachmentId: string): boolean {
+    return this.transfer.hasPendingRequest(messageId, attachmentId);
+  }
+
  async tryRestoreAttachmentFromLocal(attachment: Attachment): Promise<boolean> {
    const restored = await this.persistence.tryRestoreAttachmentFromLocal(attachment);

@@ -316,33 +321,40 @@ export class AttachmentManagerService {

    await this.restoreLocalAttachmentsForRoom(roomId);

-    if (isDirectMessageAttachmentRoomId(roomId)) {
-      await this.requestAutoDownloadsForRuntimeRoom(roomId);
-      return;
-    }
+    let messageIds: string[];

-    if (this.database.isReady()) {
+    if (isDirectMessageAttachmentRoomId(roomId)) {
+      messageIds = await this.collectMessageIdsForAttachmentsInRoom(roomId);
+    } else if (this.database.isReady()) {
      const messages = await this.database.getMessages(roomId, 500, 0);

      for (const message of messages) {
        this.runtimeStore.rememberMessageRoom(message.id, message.roomId);
-        await this.requestAutoDownloadsForMessage(message.id);
      }

-      return;
+      messageIds = messages.map((message) => message.id);
+    } else {
+      messageIds = await this.collectMessageIdsForAttachmentsInRoom(roomId);
    }

-    await this.requestAutoDownloadsForRuntimeRoom(roomId);
+    await runTasksWithBoundedConcurrency(
+      messageIds.map((messageId) => () => this.requestAutoDownloadsForMessage(messageId)),
+      ATTACHMENT_AUTO_DOWNLOAD_CONCURRENCY
+    );
  }

-  private async requestAutoDownloadsForRuntimeRoom(roomId: string): Promise<void> {
+  private async collectMessageIdsForAttachmentsInRoom(roomId: string): Promise<string[]> {
+    const messageIds: string[] = [];
+
    for (const [messageId] of this.runtimeStore.getAttachmentEntries()) {
      const attachmentRoomId = await this.persistence.resolveMessageRoomId(messageId);

      if (attachmentRoomId === roomId) {
-        await this.requestAutoDownloadsForMessage(messageId);
+        messageIds.push(messageId);
      }
    }
+
+    return messageIds;
  }

  private async requestAutoDownloadsForMessage(messageId: string, attachmentId?: string): Promise<void> {
@@ -414,6 +414,7 @@ describe('AttachmentTransferService', () => {
      fileId: FILE_ID,
      index: 0
    });
+
    expect(persistence.saveFileToDisk).not.toHaveBeenCalled();
  });

@@ -480,6 +481,7 @@ describe('AttachmentTransferService', () => {
      fileId: FILE_ID,
      index: 0
    });
+
    expect(persistence.ensureInlineDisplayObjectUrl).not.toHaveBeenCalled();
    expect(persistence.saveFileToDisk).not.toHaveBeenCalled();
    expect(attachment.objectUrl).toBeUndefined();
@@ -510,15 +512,16 @@ describe('AttachmentTransferService', () => {
      fileId: FILE_ID,
      index: 0
    });
+
    expect(persistence.saveFileToDisk).not.toHaveBeenCalled();
    expect(runtimeStore.getChunkBuffer(`${MESSAGE_ID}:${FILE_ID}`)).toBeUndefined();
  });

-  it('does not hydrate media blobs after a disk-streamed download completes', async () => {
+  it('does not hydrate image blobs after a disk-streamed download completes', async () => {
    attachmentStorage.canStreamToDisk.mockReturnValue(true);

    const service = createService();
-    const attachment = registerIncomingVideo(3);
+    const attachment = registerIncomingAttachment(3);

    service.handleFileChunk(chunkPayload(0, 1, [
      1,
@@ -530,9 +533,55 @@ describe('AttachmentTransferService', () => {

    expect(attachment.savedPath).toBeTruthy();
    expect(attachment.objectUrl).toBeUndefined();
+    expect(attachmentStorage.getFileUrl).not.toHaveBeenCalled();
    expect(persistence.ensureInlineDisplayObjectUrl).not.toHaveBeenCalled();
  });

+  it('hydrates playable media with a native file url after disk-streamed download completes', async () => {
+    attachmentStorage.canStreamToDisk.mockReturnValue(true);
+    attachmentStorage.getFileUrl.mockResolvedValue('file:///appdata/server/room/files/clip.mp4');
+
+    const service = createService();
+    const attachment = registerIncomingVideo(3);
+
+    service.handleFileChunk(chunkPayload(0, 1, [
+      1,
+      2,
+      3
+    ]));
+
+    await vi.waitFor(() => expect(attachment.objectUrl).toBe('file:///appdata/server/room/files/clip.mp4'));
+
+    expect(attachment.available).toBe(true);
+    expect(attachment.savedPath).toBeTruthy();
+    expect(attachmentStorage.getFileUrl).toHaveBeenCalledWith(attachment.savedPath);
+    expect(persistence.ensureInlineDisplayObjectUrl).not.toHaveBeenCalled();
+  });
+
+  it('falls back to inline blob hydration for playable media when no native file url exists', async () => {
+    attachmentStorage.canStreamToDisk.mockReturnValue(true);
+    attachmentStorage.getFileUrl.mockResolvedValue(null);
+    persistence.ensureInlineDisplayObjectUrl.mockImplementation(async (entry) => {
+      entry.objectUrl = 'blob:http://localhost/clip';
+      entry.available = true;
+      return true;
+    });
+
+    const service = createService();
+    const attachment = registerIncomingVideo(3);
+
+    service.handleFileChunk(chunkPayload(0, 1, [
+      1,
+      2,
+      3
+    ]));
+
+    await vi.waitFor(() => expect(persistence.ensureInlineDisplayObjectUrl).toHaveBeenCalled());
+
+    expect(attachment.objectUrl).toBe('blob:http://localhost/clip');
+    expect(attachment.available).toBe(true);
+  });
+
  it('rejects oversized browser downloads before requesting peers', async () => {
    attachmentStorage.canStreamToDisk.mockReturnValue(false);
    attachmentStorage.canPersistSize.mockImplementation((bytes: number) => bytes <= 50 * 1024 * 1024);
@@ -727,4 +776,37 @@ describe('AttachmentTransferService', () => {
    expect(attachment.available).toBe(true);
    expect(attachment.savedPath).toBe('/appdata/server/room/files/setup.exe');
  });
+
+  it('sends file-cancel to pending request peers instead of only the uploader', async () => {
+    const mirrorPeer = 'mirror-peer';
+
+    webrtc.getConnectedPeers.mockReturnValue([mirrorPeer]);
+    webrtc.sendToPeer.mockClear();
+
+    const service = createService();
+    const attachment = registerIncomingAttachment(3_000);
+
+    attachment.uploaderPeerId = 'uploader-peer';
+    runtimeStore.addAnnouncedHost(`${MESSAGE_ID}:${FILE_ID}`, mirrorPeer);
+    runtimeStore.addAnnouncedHost(`${MESSAGE_ID}:${FILE_ID}`, attachment.uploaderPeerId);
+
+    await service.requestFromAnyPeer(MESSAGE_ID, attachment);
+
+    expect(webrtc.sendToPeer).toHaveBeenCalledWith(mirrorPeer, expect.objectContaining({
+      type: 'file-request'
+    }));
+
+    attachment.receivedBytes = 512;
+    service.cancelRequest(MESSAGE_ID, attachment);
+
+    expect(webrtc.sendToPeer).toHaveBeenCalledWith(mirrorPeer, {
+      type: 'file-cancel',
+      messageId: MESSAGE_ID,
+      fileId: FILE_ID
+    });
+
+    expect(attachment.receivedBytes).toBe(0);
+    expect(attachment.available).toBe(false);
+    expect(service.hasPendingRequest(MESSAGE_ID, FILE_ID)).toBe(false);
+  });
 });
@@ -229,6 +229,10 @@ export class AttachmentTransferService {
  }

  requestImageFromAnyPeer(messageId: string, attachment: Attachment): Promise<void> {
+    if ((attachment.receivedBytes ?? 0) > 0 || this.hasPendingRequest(messageId, attachment.id)) {
+      this.cancelRequest(messageId, attachment);
+    }
+
    return this.requestFromAnyPeer(messageId, attachment);
  }

@@ -456,22 +460,22 @@ export class AttachmentTransferService {
  }

  cancelRequest(messageId: string, attachment: Attachment): void {
-    const targetPeerId = attachment.uploaderPeerId;
-
-    if (!targetPeerId)
-      return;
-
    try {
+      const requestKey = this.buildRequestKey(messageId, attachment.id);
      const assemblyKey = `${messageId}:${attachment.id}`;
+      const pendingPeers = this.runtimeStore.getPendingRequestPeers(requestKey);

      this.runtimeStore.deleteChunkBuffer(assemblyKey);
      this.runtimeStore.deleteChunkCount(assemblyKey);
+      this.runtimeStore.deletePendingRequest(requestKey);
      void this.deleteDiskReceiveAssembly(assemblyKey);
+      this.chunkAcks.cancelPendingForFile(messageId, attachment.id);

      attachment.receivedBytes = 0;
      attachment.speedBps = 0;
      attachment.startedAtMs = undefined;
      attachment.lastUpdateMs = undefined;
+      attachment.requestError = undefined;

      if (attachment.objectUrl) {
        try {
@@ -489,8 +493,21 @@ export class AttachmentTransferService {
        messageId,
        fileId: attachment.id
      };
+      const peersToNotify = new Set<string>();

-      this.webrtc.sendToPeer(targetPeerId, fileCancelEvent);
+      if (pendingPeers) {
+        for (const peerId of pendingPeers) {
+          peersToNotify.add(peerId);
+        }
+      }
+
+      if (attachment.uploaderPeerId) {
+        peersToNotify.add(attachment.uploaderPeerId);
+      }
+
+      for (const peerId of peersToNotify) {
+        this.webrtc.sendToPeer(peerId, fileCancelEvent);
+      }
    } catch { /* best-effort */ }
  }

@@ -997,6 +1014,23 @@ export class AttachmentTransferService {
    this.runtimeStore.touch();
    void this.persistence.persistAttachmentMeta(attachment);
    void this.announceLocalHost(attachment);
+    void this.hydratePlayableMediaAfterDiskReceive(attachment);
+  }
+
+  private async hydratePlayableMediaAfterDiskReceive(attachment: Attachment): Promise<void> {
+    if (!this.isPlayableMedia(attachment) || !attachment.savedPath) {
+      return;
+    }
+
+    const nativeUrl = await this.attachmentStorage.getFileUrl(attachment.savedPath);
+
+    if (nativeUrl) {
+      attachment.objectUrl = nativeUrl;
+      this.runtimeStore.touch();
+      return;
+    }
+
+    await this.persistence.ensureInlineDisplayObjectUrl(attachment);
  }

  private async getOrCreateDiskReceiveAssembly(
@@ -0,0 +1,29 @@
+import { runTasksWithBoundedConcurrency } from './attachment-autodownload-concurrency.rules';
+
+describe('attachment-autodownload-concurrency.rules', () => {
+  it('runs tasks with bounded concurrency', async () => {
+    let active = 0;
+    let maxActive = 0;
+
+    const tasks = Array.from({ length: 6 }, (_, index) => async () => {
+      active += 1;
+      maxActive = Math.max(maxActive, active);
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      active -= 1;
+
+      return index;
+    });
+    const results = await runTasksWithBoundedConcurrency(tasks, 2);
+
+    expect(results).toEqual([
+      0,
+      1,
+      2,
+      3,
+      4,
+      5
+    ]);
+
+    expect(maxActive).toBeLessThanOrEqual(2);
+  });
+});
@@ -0,0 +1,29 @@
+/** Default parallel attachment auto-download limit per watched room. */
+export const ATTACHMENT_AUTO_DOWNLOAD_CONCURRENCY = 3;
+
+export async function runTasksWithBoundedConcurrency<T>(
+  tasks: readonly (() => Promise<T>)[],
+  concurrency: number
+): Promise<T[]> {
+  if (tasks.length === 0) {
+    return [];
+  }
+
+  const limit = Math.max(1, Math.min(concurrency, tasks.length));
+  const results: T[] = new Array(tasks.length);
+
+  let nextIndex = 0;
+
+  async function runWorker(): Promise<void> {
+    while (nextIndex < tasks.length) {
+      const currentIndex = nextIndex;
+
+      nextIndex += 1;
+      results[currentIndex] = await tasks[currentIndex]();
+    }
+  }
+
+  await Promise.all(Array.from({ length: limit }, () => runWorker()));
+
+  return results;
+}
@@ -4,10 +4,7 @@ import {
  it
 } from 'vitest';

-import {
-  base64DecodedByteLength,
-  decodeBase64ToUint8Array
-} from './attachment-blob.rules';
+import { base64DecodedByteLength, decodeBase64ToUint8Array } from './attachment-blob.rules';

 describe('attachment blob rules', () => {
  it('decodes base64 payloads into byte arrays', () => {
@@ -4,10 +4,7 @@ import {
  it
 } from 'vitest';

-import {
-  canDownloadAttachment,
-  resolveAttachmentDiskPath
-} from './attachment-download.rules';
+import { canDownloadAttachment, resolveAttachmentDiskPath } from './attachment-download.rules';

 describe('attachment-download.rules', () => {
  it('allows download when a completed disk-only attachment has no object URL', () => {
@@ -1,6 +1,8 @@
 import {
  getWatchedAttachmentRoomIdFromUrl,
+  isAttachmentPendingMediaHydration,
  isDirectMessageAttachmentRoomId,
+  isPlayableAttachmentMedia,
  shouldCopyUploaderMediaToAppData,
  shouldCopyLargeUploaderFileToAppData,
  shouldStreamAttachmentReceiveToDisk,
@@ -83,6 +85,24 @@ describe('attachment logic', () => {
    }, capabilities)).toBe(true);
  });

+  it('identifies playable media pending hydration from disk paths', () => {
+    expect(isPlayableAttachmentMedia({ mime: 'video/mp4' })).toBe(true);
+    expect(isPlayableAttachmentMedia({ mime: 'image/png' })).toBe(false);
+
+    expect(isAttachmentPendingMediaHydration({
+      mime: 'audio/mpeg',
+      available: true,
+      savedPath: '/data/song.mp3'
+    })).toBe(true);
+
+    expect(isAttachmentPendingMediaHydration({
+      mime: 'audio/mpeg',
+      available: true,
+      objectUrl: 'file:///data/song.mp3',
+      savedPath: '/data/song.mp3'
+    })).toBe(false);
+  });
+
  it('receives browser-sized files in memory when disk streaming is unavailable', () => {
    const browserCapabilities = {
      canStreamToDisk: false,
@@ -11,6 +11,27 @@ export function isAttachmentMedia(attachment: Pick<Attachment, 'mime'>): boolean
    attachment.mime.startsWith('audio/');
 }

+export function isPlayableAttachmentMedia(attachment: Pick<Attachment, 'mime'>): boolean {
+  return attachment.mime.startsWith('video/') || attachment.mime.startsWith('audio/');
+}
+
+export function isAttachmentPendingMediaHydration(
+  attachment: Pick<
+    Attachment,
+    'available' | 'filePath' | 'mime' | 'objectUrl' | 'receivedBytes' | 'savedPath'
+  >
+): boolean {
+  if (!isPlayableAttachmentMedia(attachment) || attachment.objectUrl) {
+    return false;
+  }
+
+  if ((attachment.receivedBytes ?? 0) > 0 && attachment.available !== true) {
+    return false;
+  }
+
+  return !!(attachment.savedPath?.trim() || attachment.filePath?.trim());
+}
+
 export function shouldAutoRequestWhenWatched(attachment: Attachment): boolean {
  return attachment.isImage ||
    (isAttachmentMedia(attachment) && attachment.size <= MAX_AUTO_SAVE_SIZE_BYTES);