Fix wonkyness #18

Merged
myxelium merged 11 commits from fix-bug into main 2026-07-14 09:28:25 +00:00
12 changed files with 206 additions and 34 deletions
Showing only changes of commit b13f71d2d3 - Show all commits
@@ -107,7 +107,7 @@ 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 **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.
- **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. Files **`MAX_AUTO_SAVE_SIZE_BYTES` (10 MB)** assemble in memory (parallel chunk receive, immediate `file-chunk-ack`) and are persisted after completion via `shouldPersistDownloadedAttachment`. **Oversized** persistable downloads (`> 10 MB`) append directly to disk when the store supports streaming (`canStreamToDisk`) — metadata `filePath` does not force an in-memory 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** ≤ 10 MB get an immediate `objectUrl` blob; oversized 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
@@ -52,6 +52,7 @@ describe('AttachmentTransferService', () => {
getFileUrl: ReturnType<typeof vi.fn>;
resolveExistingPath: ReturnType<typeof vi.fn>;
resolveLegacyImagePath: ReturnType<typeof vi.fn>;
getFileSize: ReturnType<typeof vi.fn>;
appendBase64: ReturnType<typeof vi.fn>;
appendBytes: ReturnType<typeof vi.fn>;
createWritableFile: ReturnType<typeof vi.fn>;
@@ -94,6 +95,7 @@ describe('AttachmentTransferService', () => {
getFileUrl: vi.fn(async () => null),
resolveExistingPath: vi.fn(async () => null),
resolveLegacyImagePath: vi.fn(async () => null),
getFileSize: vi.fn(async () => null),
appendBase64: vi.fn(async () => true),
appendBytes: vi.fn(async () => true),
createWritableFile: vi.fn(async () => '/appdata/server/room/files/file-1'),
@@ -391,11 +393,43 @@ describe('AttachmentTransferService', () => {
return attachment;
}
it('streams playable media to disk when the store supports streaming', async () => {
it('assembles small images in memory even when the store supports disk streaming', async () => {
attachmentStorage.canStreamToDisk.mockReturnValue(true);
const service = createService();
const attachment = registerIncomingVideo(3);
const attachment = registerIncomingAttachment(9);
service.handleFileChunk(chunkPayload(0, 3, [
1,
2,
3
]));
service.handleFileChunk(chunkPayload(1, 3, [
4,
5,
6
]));
service.handleFileChunk(chunkPayload(2, 3, [
7,
8,
9
]));
await vi.waitFor(() => expect(attachment.available).toBe(true));
expect(attachmentStorage.createWritableFile).not.toHaveBeenCalled();
expect(attachmentStorage.appendBytes).not.toHaveBeenCalled();
expect(persistence.saveFileToDisk).toHaveBeenCalledTimes(1);
expect(attachment.objectUrl).toMatch(/^blob:/);
});
it('streams oversized playable media to disk when the store supports streaming', async () => {
attachmentStorage.canStreamToDisk.mockReturnValue(true);
const service = createService();
const attachment = registerIncomingVideo(12 * 1024 * 1024);
service.handleFileChunk(chunkPayload(0, 1, [
1,
@@ -517,11 +551,11 @@ describe('AttachmentTransferService', () => {
expect(runtimeStore.getChunkBuffer(`${MESSAGE_ID}:${FILE_ID}`)).toBeUndefined();
});
it('does not hydrate image blobs after a disk-streamed download completes', async () => {
it('does not hydrate image blobs after a disk-streamed oversized download completes', async () => {
attachmentStorage.canStreamToDisk.mockReturnValue(true);
const service = createService();
const attachment = registerIncomingAttachment(3);
const attachment = registerIncomingAttachment(12 * 1024 * 1024);
service.handleFileChunk(chunkPayload(0, 1, [
1,
@@ -537,12 +571,12 @@ describe('AttachmentTransferService', () => {
expect(persistence.ensureInlineDisplayObjectUrl).not.toHaveBeenCalled();
});
it('hydrates playable media with a native file url after disk-streamed download completes', async () => {
it('hydrates playable media with a native file url after disk-streamed oversized download completes', async () => {
attachmentStorage.canStreamToDisk.mockReturnValue(true);
attachmentStorage.getFileUrl.mockResolvedValue('file:///appdata/server/room/files/clip.mp4');
const service = createService();
const attachment = registerIncomingVideo(3);
const attachment = registerIncomingVideo(12 * 1024 * 1024);
service.handleFileChunk(chunkPayload(0, 1, [
1,
@@ -558,7 +592,7 @@ describe('AttachmentTransferService', () => {
expect(persistence.ensureInlineDisplayObjectUrl).not.toHaveBeenCalled();
});
it('falls back to inline blob hydration for playable media when no native file url exists', async () => {
it('falls back to inline blob hydration for oversized playable media when no native file url exists', async () => {
attachmentStorage.canStreamToDisk.mockReturnValue(true);
attachmentStorage.getFileUrl.mockResolvedValue(null);
persistence.ensureInlineDisplayObjectUrl.mockImplementation(async (entry) => {
@@ -568,7 +602,7 @@ describe('AttachmentTransferService', () => {
});
const service = createService();
const attachment = registerIncomingVideo(3);
const attachment = registerIncomingVideo(12 * 1024 * 1024);
service.handleFileChunk(chunkPayload(0, 1, [
1,
@@ -632,8 +666,30 @@ describe('AttachmentTransferService', () => {
expect(persistence.persistUploadCopyFromSourcePath).toHaveBeenCalled();
});
it('falls back to the in-memory upload when the resolved disk path is empty', async () => {
attachmentStorage.resolveExistingPath.mockResolvedValue('/appdata/server/room/files/photo.png');
attachmentStorage.getFileSize.mockResolvedValue(0);
const service = createService();
const attachment = registerIncomingAttachment(9);
attachment.available = true;
attachment.savedPath = '/appdata/server/room/files/photo.png';
runtimeStore.setOriginalFile(`${MESSAGE_ID}:${FILE_ID}`, new File([new Uint8Array(9)], 'photo.png', { type: 'image/png' }));
await service.handleFileRequest({
messageId: MESSAGE_ID,
fileId: FILE_ID,
fromPeerId: 'peer-2'
});
expect(transport.streamFileFromDiskToPeer).not.toHaveBeenCalled();
expect(transport.streamFileToPeer).toHaveBeenCalledTimes(1);
});
it('streams a restored oversized generic file from app data when the in-memory upload is gone', async () => {
attachmentStorage.resolveExistingPath.mockResolvedValue('/appdata/server/room/files/setup.exe');
attachmentStorage.getFileSize.mockResolvedValue(12 * 1024 * 1024);
const service = createService();
const attachment = registerIncomingGenericFile(12 * 1024 * 1024);
@@ -737,6 +793,7 @@ describe('AttachmentTransferService', () => {
it('prefers streaming from disk over an in-memory original file when both exist', async () => {
attachmentStorage.resolveExistingPath.mockResolvedValue('/appdata/server/room/files/setup.exe');
attachmentStorage.getFileSize.mockResolvedValue(12 * 1024 * 1024);
const service = createService();
const attachment = registerIncomingGenericFile(12 * 1024 * 1024);
@@ -17,6 +17,7 @@ import {
shouldPersistDownloadedAttachment,
shouldStreamAttachmentReceiveToDisk
} from '../../domain/logic/attachment.logic';
import { shouldServeAttachmentFromDiskPath } from '../../domain/logic/attachment-serve.rules';
import type { Attachment, AttachmentMeta } from '../../domain/models/attachment.model';
import {
ATTACHMENT_TRANSFER_EWMA_CURRENT_WEIGHT,
@@ -564,7 +565,7 @@ export class AttachmentTransferService {
? await this.attachmentStorage.resolveExistingPath(attachment)
: null;
if (diskPath) {
if (diskPath && shouldServeAttachmentFromDiskPath(await this.attachmentStorage.getFileSize(diskPath))) {
await this.transport.streamFileFromDiskToPeer(
fromPeerId,
messageId,
@@ -598,7 +599,7 @@ export class AttachmentTransferService {
roomName
);
if (legacyDiskPath) {
if (legacyDiskPath && shouldServeAttachmentFromDiskPath(await this.attachmentStorage.getFileSize(legacyDiskPath))) {
await this.transport.streamFileFromDiskToPeer(
fromPeerId,
messageId,
@@ -0,0 +1,15 @@
import { shouldServeAttachmentFromDiskPath } from './attachment-serve.rules';
describe('shouldServeAttachmentFromDiskPath', () => {
it('accepts paths with a positive byte length', () => {
expect(shouldServeAttachmentFromDiskPath(1)).toBe(true);
expect(shouldServeAttachmentFromDiskPath(4096)).toBe(true);
});
it('rejects empty, missing, or invalid sizes', () => {
expect(shouldServeAttachmentFromDiskPath(0)).toBe(false);
expect(shouldServeAttachmentFromDiskPath(null)).toBe(false);
expect(shouldServeAttachmentFromDiskPath(undefined)).toBe(false);
expect(shouldServeAttachmentFromDiskPath(Number.NaN)).toBe(false);
});
});
@@ -0,0 +1,4 @@
/** True when a resolved on-disk path contains bytes worth streaming to a peer. */
export function shouldServeAttachmentFromDiskPath(fileSize: number | null | undefined): boolean {
return typeof fileSize === 'number' && Number.isFinite(fileSize) && fileSize > 0;
}
@@ -60,7 +60,7 @@ describe('attachment logic', () => {
}, undefined, true)).toBe(false);
});
it('streams any persistable download to disk when the store supports streaming', () => {
it('streams only oversized persistable downloads to disk when the store supports streaming', () => {
const capabilities = {
canStreamToDisk: true,
canPersistSize: (bytes: number) => bytes <= 256 * 1024 * 1024
@@ -74,9 +74,15 @@ describe('attachment logic', () => {
expect(shouldStreamAttachmentReceiveToDisk({
size: 3,
mime: 'application/zip',
mime: 'image/png',
filePath: undefined
}, capabilities)).toBe(true);
}, capabilities)).toBe(false);
expect(shouldStreamAttachmentReceiveToDisk({
size: 10 * 1024 * 1024,
mime: 'image/jpeg',
filePath: undefined
}, capabilities)).toBe(false);
expect(shouldStreamAttachmentReceiveToDisk({
size: 200 * 1024 * 1024,
@@ -92,7 +92,10 @@ export function shouldStreamAttachmentReceiveToDisk(
return false;
}
return true;
// Small files assemble in memory (parallel chunk receive + immediate acks) and are
// persisted after completion. Disk streaming is reserved for oversized downloads
// so we never buffer an entire large file in RAM.
return attachment.size > MAX_AUTO_SAVE_SIZE_BYTES;
}
export function canReceiveAttachmentInMemory(
@@ -4,7 +4,7 @@ import {
it
} from 'vitest';
import { buildChatMessageGalleryTiles } from './chat-message-image-gallery.rules';
import { buildChatMessageGalleryTiles, resolveChatMessageGalleryTileState } from './chat-message-image-gallery.rules';
describe('buildChatMessageGalleryTiles', () => {
it('marks displayable, hydrating, downloading, and retry tiles from attachment state', () => {
@@ -51,4 +51,18 @@ describe('buildChatMessageGalleryTiles', () => {
'retry'
]);
});
it('treats zero-byte pending requests as downloading instead of retry', () => {
const attachment = {
id: 'pending',
filename: 'waiting.png',
mime: 'image/png',
isImage: true,
available: false,
size: 256
};
expect(resolveChatMessageGalleryTileState(attachment)).toBe('retry');
expect(resolveChatMessageGalleryTileState(attachment, { pendingRequest: true })).toBe('downloading');
});
});
@@ -11,17 +11,26 @@ export interface ChatMessageGalleryTile<T extends ImageAttachmentCandidate = Ima
state: ChatMessageGalleryTileState;
}
export interface ChatMessageGalleryTileOptions {
pendingRequest?: boolean;
}
export function buildChatMessageGalleryTiles<T extends ImageAttachmentCandidate>(
attachments: readonly T[]
attachments: readonly T[],
options: ChatMessageGalleryTileOptions | ((attachment: T) => ChatMessageGalleryTileOptions) = {}
): ChatMessageGalleryTile<T>[] {
return attachments.map((attachment) => ({
attachment,
state: resolveChatMessageGalleryTileState(attachment)
state: resolveChatMessageGalleryTileState(
attachment,
typeof options === 'function' ? options(attachment) : options
)
}));
}
export function resolveChatMessageGalleryTileState<T extends ImageAttachmentCandidate>(
attachment: T
attachment: T,
options: ChatMessageGalleryTileOptions = {}
): ChatMessageGalleryTileState {
if (isInlineDisplayableImage(attachment)) {
return 'displayable';
@@ -31,7 +40,7 @@ export function resolveChatMessageGalleryTileState<T extends ImageAttachmentCand
return 'hydrating';
}
if ((attachment.receivedBytes ?? 0) > 0) {
if (options.pendingRequest || (attachment.receivedBytes ?? 0) > 0) {
return 'downloading';
}
@@ -104,7 +104,26 @@ export class ChatMessagesComponent {
readonly replyTo = signal<Message | null>(null);
readonly showKlipyGifPicker = signal(false);
readonly lightboxState = signal<ChatLightboxState | null>(null);
readonly galleryAttachments = signal<Attachment[] | null>(null);
readonly galleryMessageId = signal<string | null>(null);
readonly galleryAttachmentOrder = signal<readonly string[]>([]);
readonly galleryAttachments = computed(() => {
const messageId = this.galleryMessageId();
const attachmentIds = this.galleryAttachmentOrder();
void this.attachmentsSvc.updated;
if (!messageId || attachmentIds.length === 0) {
return null;
}
const attachmentsById = new Map(
this.attachmentsSvc.getForMessage(messageId).map((attachment) => [attachment.id, attachment])
);
return attachmentIds
.map((attachmentId) => attachmentsById.get(attachmentId))
.filter((attachment): attachment is Attachment => !!attachment);
});
readonly imageContextMenu = signal<ChatMessageImageContextMenuEvent | null>(null);
constructor() {
@@ -345,13 +364,20 @@ export class ChatMessagesComponent {
return;
}
const messageId = attachments[0]?.messageId;
if (!messageId) {
return;
}
const displayableImages = attachments.filter((attachment) => attachment.available && attachment.objectUrl);
if (displayableImages.length > 0) {
this.attachmentsSvc.pinDisplayBlobs(displayableImages);
}
this.galleryAttachments.set(attachments);
this.galleryMessageId.set(messageId);
this.galleryAttachmentOrder.set(attachments.map((attachment) => attachment.id));
}
closeImageGallery(): void {
@@ -361,7 +387,8 @@ export class ChatMessagesComponent {
this.attachmentsSvc.unpinDisplayBlobs(gallery);
}
this.galleryAttachments.set(null);
this.galleryMessageId.set(null);
this.galleryAttachmentOrder.set([]);
}
openImageContextMenu(event: ChatMessageImageContextMenuEvent): void {
@@ -378,12 +405,14 @@ export class ChatMessagesComponent {
retryGalleryImage(event: ChatMessageAttachmentEvent): void {
const { messageId, attachment } = event;
const liveAttachment = this.attachmentsSvc.getForMessage(messageId).find((entry) => entry.id === attachment.id)
?? attachment;
if ((attachment.receivedBytes ?? 0) > 0 || this.attachmentsSvc.hasPendingRequest(messageId, attachment.id)) {
this.attachmentsSvc.cancelRequest(messageId, attachment);
if ((liveAttachment.receivedBytes ?? 0) > 0 || this.attachmentsSvc.hasPendingRequest(messageId, liveAttachment.id)) {
this.attachmentsSvc.cancelRequest(messageId, liveAttachment);
}
void this.attachmentsSvc.requestImageFromAnyPeer(messageId, attachment);
void this.attachmentsSvc.requestImageFromAnyPeer(messageId, liveAttachment);
}
cancelGalleryImage(event: ChatMessageAttachmentEvent): void {
@@ -18,7 +18,7 @@ import {
lucideDownload,
lucideX
} from '@ng-icons/lucide';
import { Attachment } from '../../../../../attachment';
import { Attachment, AttachmentFacade } from '../../../../../attachment';
import { canStepLightbox } from '../../../../domain/rules/chat-message-lightbox.rules';
import { buildChatMessageGalleryTiles, type ChatMessageGalleryTile } from '../../../../domain/rules/chat-message-image-gallery.rules';
import { AppI18nService, APP_TRANSLATE_IMPORTS } from '../../../../../../core/i18n';
@@ -80,7 +80,11 @@ export class ChatMessageOverlaysComponent implements OnDestroy {
return [];
}
return buildChatMessageGalleryTiles(attachments);
const messageId = attachments[0]?.messageId;
return buildChatMessageGalleryTiles(attachments, (attachment) => ({
pendingRequest: !!messageId && this.attachmentsSvc.hasPendingRequest(messageId, attachment.id)
}));
});
readonly lightboxAttachment = computed(() => {
@@ -124,6 +128,7 @@ export class ChatMessageOverlaysComponent implements OnDestroy {
});
private readonly appI18n = inject(AppI18nService);
private readonly attachmentsSvc = inject(AttachmentFacade);
private readonly LIGHTBOX_CONTROLS_IDLE_MS = 2200;
private lightboxControlsHideTimer: ReturnType<typeof setTimeout> | null = null;
@@ -119,7 +119,26 @@ export class DmChatComponent {
readonly linkMetadataByMessageId = signal<Record<string, LinkMetadata[]>>({});
readonly replyTo = signal<Message | null>(null);
readonly lightboxState = signal<ChatLightboxState | null>(null);
readonly galleryAttachments = signal<Attachment[] | null>(null);
readonly galleryMessageId = signal<string | null>(null);
readonly galleryAttachmentOrder = signal<readonly string[]>([]);
readonly galleryAttachments = computed(() => {
const messageId = this.galleryMessageId();
const attachmentIds = this.galleryAttachmentOrder();
void this.attachments.updated;
if (!messageId || attachmentIds.length === 0) {
return null;
}
const attachmentsById = new Map(
this.attachments.getForMessage(messageId).map((attachment) => [attachment.id, attachment])
);
return attachmentIds
.map((attachmentId) => attachmentsById.get(attachmentId))
.filter((attachment): attachment is Attachment => !!attachment);
});
readonly imageContextMenu = signal<ChatMessageImageContextMenuEvent | null>(null);
readonly routeConversationId = toSignal(this.route.paramMap.pipe(map((params) => params.get('conversationId'))), {
initialValue: this.route.snapshot.paramMap.get('conversationId')
@@ -495,11 +514,19 @@ export class DmChatComponent {
return;
}
this.galleryAttachments.set(attachments);
const messageId = attachments[0]?.messageId;
if (!messageId) {
return;
}
this.galleryMessageId.set(messageId);
this.galleryAttachmentOrder.set(attachments.map((attachment) => attachment.id));
}
closeImageGallery(): void {
this.galleryAttachments.set(null);
this.galleryMessageId.set(null);
this.galleryAttachmentOrder.set([]);
}
openImageContextMenu(event: ChatMessageImageContextMenuEvent): void {
@@ -516,12 +543,14 @@ export class DmChatComponent {
retryGalleryImage(event: ChatMessageAttachmentEvent): void {
const { messageId, attachment } = event;
const liveAttachment = this.attachments.getForMessage(messageId).find((entry) => entry.id === attachment.id)
?? attachment;
if ((attachment.receivedBytes ?? 0) > 0 || this.attachments.hasPendingRequest(messageId, attachment.id)) {
this.attachments.cancelRequest(messageId, attachment);
if ((liveAttachment.receivedBytes ?? 0) > 0 || this.attachments.hasPendingRequest(messageId, liveAttachment.id)) {
this.attachments.cancelRequest(messageId, liveAttachment);
}
void this.attachments.requestImageFromAnyPeer(messageId, attachment);
void this.attachments.requestImageFromAnyPeer(messageId, liveAttachment);
}
cancelGalleryImage(event: ChatMessageAttachmentEvent): void {