diff --git a/packages/common/nbstore/src/sync/blob/index.ts b/packages/common/nbstore/src/sync/blob/index.ts index ead12a5a74..1464eaa693 100644 --- a/packages/common/nbstore/src/sync/blob/index.ts +++ b/packages/common/nbstore/src/sync/blob/index.ts @@ -102,16 +102,42 @@ export class BlobSyncImpl implements BlobSync { readonly blobSync: BlobSyncStorage ) {} - downloadBlob(blobId: string) { + downloadBlob(blobId: string): Promise { const signal = this.abortController.signal; - return Promise.race( - this.peers.map(p => p.downloadBlob(blobId, signal)) - ).catch(err => { - if (err === MANUALLY_STOP) { + + return new Promise((resolve, reject) => { + let completed = 0; + const totalPeers = this.peers.length; + + if (totalPeers === 0) { + resolve(); return; } - // should never reach here, `downloadBlob()` should never throw - console.error(err); + + // download from all peers concurrently + // resolve if any peer has success + this.peers.forEach(peer => { + peer + .downloadBlob(blobId, signal) + .then(result => { + if (result === true) { + // resolve if the peer has success + resolve(); + } + }) + .catch(err => { + // should never throw + // unless the signal is aborted + reject(err); + }) + .finally(() => { + completed++; + if (completed === totalPeers) { + // resolve if all peers finish + resolve(); + } + }); + }); }); } diff --git a/packages/common/nbstore/src/sync/blob/peer.ts b/packages/common/nbstore/src/sync/blob/peer.ts index cc49fc6be9..ad6f62dd30 100644 --- a/packages/common/nbstore/src/sync/blob/peer.ts +++ b/packages/common/nbstore/src/sync/blob/peer.ts @@ -38,9 +38,15 @@ export class BlobSyncPeer { readonly blobSync: BlobSyncStorage ) {} - private readonly downloadingPromise = new Map>(); + private readonly downloadingPromise = new Map>(); - downloadBlob(blobId: string, signal?: AbortSignal): Promise { + /** + * Downloads a blob from the peer with retry logic + * @returns true if the blob is downloaded successfully, false if the blob is not found or encounters an error + * + * @throws This method will never throw (errors are saved to the sync status) unless the signal is aborted + */ + downloadBlob(blobId: string, signal?: AbortSignal): Promise { // if the blob is already downloading, return the existing promise const existing = this.downloadingPromise.get(blobId); if (existing) { @@ -53,7 +59,7 @@ export class BlobSyncPeer { count: this.remote.isReadonly ? 1 : 5, // readonly remote storage will not retry }; - const promise = new Promise((resolve, reject) => { + const promise = new Promise((resolve, reject) => { // mark the blob as downloading this.status.blobDownloading(blobId); @@ -72,6 +78,7 @@ export class BlobSyncPeer { new Date() ); await this.local.set(data, signal); + resolve(true); } else { // if the blob is not found, maybe the uploader have't uploaded the blob yet, we will retry several times attempts++; @@ -82,9 +89,11 @@ export class BlobSyncPeer { ); // eslint-disable-next-line @typescript-eslint/no-misused-promises setTimeout(attempt, waitTime); + } else { + // reach the max retry times, resolve the promise with false + resolve(false); } } - resolve(); } catch (error) { // if we encounter any error, reject without retry reject(error); @@ -102,6 +111,7 @@ export class BlobSyncPeer { blobId, error instanceof Error ? error.message : String(error) ); + return false; }) .finally(() => { this.status.blobDownloadFinish(blobId);