myxelium/Toju
1
0
Fork 0
Code Issues 16 Pull Requests Actions Packages Projects 2 Releases 39 Wiki Activity

fix: Bug - Attachments gets syncronized corrupt

Browse Source
This commit is contained in:
myxelium
2026-06-11 00:37:06 +02:00
parent d174536272
commit 5bf4f698df
7 changed files with 605 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
toju-app/src/app/domains/attachment/application/services/attachment-transfer.service.ts
+163 -98
View File
@@ -55,6 +55,20 @@ interface ValidFileChunkPayload {
total: number;
}
function isValidFileChunkPayload(payload: FileChunkPayload): payload is FileChunkPayload & ValidFileChunkPayload {
const { messageId, fileId, index, total, data } = payload;
return !!messageId && !!fileId &&
typeof data === 'string' &&
typeof index === 'number' &&
typeof total === 'number' &&
Number.isInteger(index) &&
Number.isInteger(total) &&
total > 0 &&
index >= 0 &&
index < total;
}
@Injectable({ providedIn: 'root' })
export class AttachmentTransferService {
private readonly ngrxStore = inject(Store);
@@ -67,6 +81,7 @@ export class AttachmentTransferService {
private readonly diskReceiveAssemblies = new Map<string, DiskReceiveAssembly>();
private readonly diskReceiveChains = new Map<string, Promise<void>>();
private readonly activeOutboundTransfers = new Set<string>();
getAttachmentMetasForMessages(messageIds: string[]): Record<string, AttachmentMeta[]> {
const result: Record<string, AttachmentMeta[]> = {};
@@ -135,12 +150,19 @@ export class AttachmentTransferService {
}
async requestFromAnyPeer(messageId: string, attachment: Attachment): Promise<void> {
const requestKey = this.buildRequestKey(messageId, attachment.id);
const clearedRequestError = this.clearAttachmentRequestError(attachment);
// Mark the request pending synchronously so concurrent triggers (file-announce,
// message sync, peer connect) cannot double-request the same file - a duplicate
// request makes the sender stream the file twice and corrupts byte accounting.
this.runtimeStore.setPendingRequestPeers(requestKey, new Set<string>());
if (!attachment.available) {
const restoredLocally = await this.persistence.tryRestoreAttachmentFromLocal(attachment);
if (restoredLocally) {
this.runtimeStore.deletePendingRequest(requestKey);
this.runtimeStore.touch();
return;
}
@@ -153,6 +175,7 @@ export class AttachmentTransferService {
attachment.uploaderPeerId === currentUserId;
if (connectedPeers.length === 0) {
this.runtimeStore.deletePendingRequest(requestKey);
attachment.requestError = isUploader
? this.appI18n.instant(UPLOADER_LOCAL_FILE_MISSING_ERROR_KEY)
: this.appI18n.instant(NO_CONNECTED_PEERS_REQUEST_ERROR_KEY);
@@ -165,11 +188,6 @@ export class AttachmentTransferService {
if (clearedRequestError)
this.runtimeStore.touch();
this.runtimeStore.setPendingRequestPeers(
this.buildRequestKey(messageId, attachment.id),
new Set<string>()
);
this.sendFileRequestToNextPeer(messageId, attachment.id, attachment.uploaderPeerId);
}
@@ -334,28 +352,23 @@ export class AttachmentTransferService {
}
handleFileChunk(payload: FileChunkPayload): void {
const { messageId, fileId, fromPeerId, index, total, data } = payload;
if (
!messageId || !fileId ||
typeof index !== 'number' ||
typeof total !== 'number' ||
typeof data !== 'string' ||
!Number.isInteger(index) ||
!Number.isInteger(total) ||
total <= 0 ||
index < 0 ||
index >= total
) {
if (!isValidFileChunkPayload(payload)) {
return;
}
const { messageId, fileId, fromPeerId, index, total, data } = payload;
const list = this.runtimeStore.getAttachmentsForMessage(messageId);
const attachment = list.find((entry) => entry.id === fileId);
if (!attachment)
return;
if (attachment.available) {
// Transfer already completed (or restored locally) - trailing chunks from
// a redundant stream must not restart accounting or rewrite the file.
return;
}
if ((attachment.receivedBytes ?? 0) > attachment.size) {
return;
}
@@ -386,11 +399,14 @@ export class AttachmentTransferService {
const chunkBuffer = this.getOrCreateChunkBuffer(assemblyKey, total);
if (!chunkBuffer[index]) {
chunkBuffer[index] = decodedBytes.buffer as ArrayBuffer;
this.runtimeStore.setChunkCount(assemblyKey, (this.runtimeStore.getChunkCount(assemblyKey) ?? 0) + 1);
if (index >= chunkBuffer.length || chunkBuffer[index]) {
// Duplicate delivery (e.g. a redundant concurrent stream) - the chunk is
// already buffered, so it must not count toward transfer progress.
return;
}
chunkBuffer[index] = decodedBytes.buffer as ArrayBuffer;
this.runtimeStore.setChunkCount(assemblyKey, (this.runtimeStore.getChunkCount(assemblyKey) ?? 0) + 1);
this.updateTransferProgress(attachment, decodedBytes, fromPeerId);
this.runtimeStore.touch();
@@ -403,6 +419,110 @@ export class AttachmentTransferService {
if (!messageId || !fileId || !fromPeerId)
return;
const transferKey = this.buildTransferKey(messageId, fileId, fromPeerId);
if (this.activeOutboundTransfers.has(transferKey)) {
// A stream for this exact request is already in flight - sending the file
// twice in parallel duplicates chunks and corrupts the receiver's assembly.
return;
}
// A fresh request supersedes any earlier cancellation from this peer.
this.runtimeStore.deleteCancelledTransfer(transferKey);
this.activeOutboundTransfers.add(transferKey);
try {
await this.streamRequestedFile(messageId, fileId, fromPeerId);
} finally {
this.activeOutboundTransfers.delete(transferKey);
}
}
cancelRequest(messageId: string, attachment: Attachment): void {
const targetPeerId = attachment.uploaderPeerId;
if (!targetPeerId)
return;
try {
const assemblyKey = `${messageId}:${attachment.id}`;
this.runtimeStore.deleteChunkBuffer(assemblyKey);
this.runtimeStore.deleteChunkCount(assemblyKey);
void this.deleteDiskReceiveAssembly(assemblyKey);
attachment.receivedBytes = 0;
attachment.speedBps = 0;
attachment.startedAtMs = undefined;
attachment.lastUpdateMs = undefined;
if (attachment.objectUrl) {
try {
URL.revokeObjectURL(attachment.objectUrl);
} catch { /* ignore */ }
attachment.objectUrl = undefined;
}
attachment.available = false;
this.runtimeStore.touch();
const fileCancelEvent: FileCancelEvent = {
type: 'file-cancel',
messageId,
fileId: attachment.id
};
this.webrtc.sendToPeer(targetPeerId, fileCancelEvent);
} catch { /* best-effort */ }
}
handleFileCancel(payload: FileCancelPayload): void {
const { messageId, fileId, fromPeerId } = payload;
if (!messageId || !fileId || !fromPeerId)
return;
this.runtimeStore.addCancelledTransfer(
this.buildTransferKey(messageId, fileId, fromPeerId)
);
}
async fulfillRequestWithFile(
messageId: string,
fileId: string,
targetPeerId: string,
file: File
): Promise<void> {
this.runtimeStore.setOriginalFile(`${messageId}:${fileId}`, file);
const transferKey = this.buildTransferKey(messageId, fileId, targetPeerId);
if (this.activeOutboundTransfers.has(transferKey)) {
return;
}
this.runtimeStore.deleteCancelledTransfer(transferKey);
this.activeOutboundTransfers.add(transferKey);
try {
await this.transport.streamFileToPeer(
targetPeerId,
messageId,
fileId,
file,
() => this.isTransferCancelled(targetPeerId, messageId, fileId)
);
} finally {
this.activeOutboundTransfers.delete(transferKey);
}
}
private async streamRequestedFile(
messageId: string,
fileId: string,
fromPeerId: string
): Promise<void> {
const exactKey = `${messageId}:${fileId}`;
const originalFile = this.runtimeStore.getOriginalFile(exactKey)
?? this.runtimeStore.findOriginalFileByFileId(fileId);
@@ -484,72 +604,6 @@ export class AttachmentTransferService {
this.webrtc.sendToPeer(fromPeerId, fileNotFoundEvent);
}
cancelRequest(messageId: string, attachment: Attachment): void {
const targetPeerId = attachment.uploaderPeerId;
if (!targetPeerId)
return;
try {
const assemblyKey = `${messageId}:${attachment.id}`;
this.runtimeStore.deleteChunkBuffer(assemblyKey);
this.runtimeStore.deleteChunkCount(assemblyKey);
void this.deleteDiskReceiveAssembly(assemblyKey);
attachment.receivedBytes = 0;
attachment.speedBps = 0;
attachment.startedAtMs = undefined;
attachment.lastUpdateMs = undefined;
if (attachment.objectUrl) {
try {
URL.revokeObjectURL(attachment.objectUrl);
} catch { /* ignore */ }
attachment.objectUrl = undefined;
}
attachment.available = false;
this.runtimeStore.touch();
const fileCancelEvent: FileCancelEvent = {
type: 'file-cancel',
messageId,
fileId: attachment.id
};
this.webrtc.sendToPeer(targetPeerId, fileCancelEvent);
} catch { /* best-effort */ }
}
handleFileCancel(payload: FileCancelPayload): void {
const { messageId, fileId, fromPeerId } = payload;
if (!messageId || !fileId || !fromPeerId)
return;
this.runtimeStore.addCancelledTransfer(
this.buildTransferKey(messageId, fileId, fromPeerId)
);
}
async fulfillRequestWithFile(
messageId: string,
fileId: string,
targetPeerId: string,
file: File
): Promise<void> {
this.runtimeStore.setOriginalFile(`${messageId}:${fileId}`, file);
await this.transport.streamFileToPeer(
targetPeerId,
messageId,
fileId,
file,
() => this.isTransferCancelled(targetPeerId, messageId, fileId)
);
}
private resolveCurrentUserId(): Promise<string | null> {
return new Promise<string | null>((resolve) => {
this.ngrxStore
@@ -617,14 +671,16 @@ export class AttachmentTransferService {
return true;
}
private getOrCreateChunkBuffer(assemblyKey: string, total: number): ArrayBuffer[] {
private getOrCreateChunkBuffer(assemblyKey: string, total: number): (ArrayBuffer | undefined)[] {
const existingChunkBuffer = this.runtimeStore.getChunkBuffer(assemblyKey);
if (existingChunkBuffer) {
return existingChunkBuffer;
}
const createdChunkBuffer = new Array(total);
// Dense initialization - sparse arrays from `new Array(total)` skip holes in
// `every`/`some`, which lets incomplete transfers pass completion checks.
const createdChunkBuffer = Array.from<ArrayBuffer | undefined>({ length: total });
this.runtimeStore.setChunkBuffer(assemblyKey, createdChunkBuffer);
this.runtimeStore.setChunkCount(assemblyKey, 0);
@@ -671,18 +727,21 @@ export class AttachmentTransferService {
const receivedChunkCount = this.runtimeStore.getChunkCount(assemblyKey) ?? 0;
const completeBuffer = this.runtimeStore.getChunkBuffer(assemblyKey);
if (
!completeBuffer
|| (receivedChunkCount !== total && (attachment.receivedBytes ?? 0) < attachment.size)
|| !completeBuffer.every((part) => part instanceof ArrayBuffer)
) {
if (!completeBuffer || receivedChunkCount < total) {
return;
}
const blob = new Blob(completeBuffer, { type: attachment.mime });
const bufferedChunks = completeBuffer.filter(
(part): part is ArrayBuffer => part instanceof ArrayBuffer
);
this.runtimeStore.deleteChunkBuffer(assemblyKey);
this.runtimeStore.deleteChunkCount(assemblyKey);
// Every chunk index must be present - byte counters are never a substitute
// for chunk completeness, otherwise partial files finalize as corrupt blobs.
if (bufferedChunks.length !== total || completeBuffer.length !== total) {
return;
}
const blob = new Blob(bufferedChunks, { type: attachment.mime });
if (shouldPersistDownloadedAttachment(attachment)) {
await this.persistence.saveFileToDisk(attachment, blob);
@@ -691,6 +750,12 @@ export class AttachmentTransferService {
attachment.objectUrl = URL.createObjectURL(blob);
attachment.available = true;
// Release assembly state only after the attachment is marked available so a
// trailing duplicate chunk cannot restart accounting in a fresh buffer.
this.runtimeStore.deleteChunkBuffer(assemblyKey);
this.runtimeStore.deleteChunkCount(assemblyKey);
this.runtimeStore.touch();
void this.persistence.persistAttachmentMeta(attachment);
}
@@ -758,7 +823,7 @@ export class AttachmentTransferService {
this.updateTransferProgress(attachment, decodedBytes, payload.fromPeerId);
this.runtimeStore.touch();
if (assembly.receivedCount < assembly.total && (attachment.receivedBytes ?? 0) < attachment.size) {
if (assembly.receivedCount < assembly.total) {
return;
}