diff --git a/packages/common/nbstore/src/sync/blob/index.ts b/packages/common/nbstore/src/sync/blob/index.ts index 82b93f5799..78736d3d46 100644 --- a/packages/common/nbstore/src/sync/blob/index.ts +++ b/packages/common/nbstore/src/sync/blob/index.ts @@ -19,6 +19,8 @@ export interface BlobSyncState { } export interface BlobSyncBlobState { + needUpload: boolean; + needDownload: boolean; uploading: boolean; downloading: boolean; errorMessage?: string | null; @@ -105,6 +107,8 @@ export class BlobSyncImpl implements BlobSync { downloading: peers.some(p => p.downloading), errorMessage: peers.find(p => p.errorMessage)?.errorMessage, overSize: peers.some(p => p.overSize), + needUpload: peers.some(p => p.needUpload), + needDownload: peers.some(p => p.needDownload), }) satisfies BlobSyncBlobState ), share({ diff --git a/packages/common/nbstore/src/sync/blob/peer.ts b/packages/common/nbstore/src/sync/blob/peer.ts index cd5cc91108..984b86218a 100644 --- a/packages/common/nbstore/src/sync/blob/peer.ts +++ b/packages/common/nbstore/src/sync/blob/peer.ts @@ -14,6 +14,8 @@ export interface BlobSyncPeerState { } export interface BlobSyncPeerBlobState { + needUpload: boolean; + needDownload: boolean; uploading: boolean; downloading: boolean; overSize: boolean; @@ -62,6 +64,7 @@ export class BlobSyncPeer { }; const promise = new Promise((resolve, reject) => { + this.status.markBlobToDownload(blobId); // mark the blob as downloading this.status.blobDownloading(blobId); @@ -80,6 +83,8 @@ export class BlobSyncPeer { new Date() ); await this.local.set(data, signal); + + this.status.blobDownloadSuccess(blobId); resolve(true); } else { // if the blob is not found, maybe the uploader have't uploaded the blob yet, we will retry several times @@ -161,6 +166,7 @@ export class BlobSyncPeer { } const promise = (async () => { + this.status.markBlobToUpload(blob.key); // mark the blob as uploading this.status.blobUploading(blob.key); await this.blobSync.setBlobUploadedAt(this.peerId, blob.key, null); @@ -173,6 +179,8 @@ export class BlobSyncPeer { new Date() ); + this.status.blobUploadSuccess(blob.key); + // free the remote storage over capacity flag this.status.remoteOverCapacityFree(); return true; @@ -249,9 +257,9 @@ export class BlobSyncPeer { if (uploadedAt === null) { needUpload.push(blob.key); } else { - // if the blob has uploaded, we clear its error flag here. + // if the blob has uploaded, we clear its states flags here. // this ensures that the sync status seen by the user is clean. - this.status.blobErrorFree(blob.key); + this.status.blobStateClear(blob.key); } } @@ -261,6 +269,7 @@ export class BlobSyncPeer { // mark all blobs as will upload for (const blobKey of needUpload) { + this.status.markBlobToUpload(blobKey); this.status.blobWillUpload(blobKey); } @@ -339,6 +348,7 @@ export class BlobSyncPeer { // mark all blobs as will download for (const blobKey of needDownload) { + this.status.markBlobToDownload(blobKey); this.status.blobWillDownload(blobKey); } @@ -370,6 +380,11 @@ export class BlobSyncPeer { class BlobSyncPeerStatus { overCapacity = false; + + // blobs that need to be uploaded or downloaded + toUpload = new Set(); + toDownload = new Set(); + willUpload = new Set(); uploading = new Set(); downloading = new Set(); @@ -403,6 +418,8 @@ class BlobSyncPeerStatus { return new Observable(subscribe => { const next = () => { subscribe.next({ + needUpload: this.toUpload.has(blobId), + needDownload: this.toDownload.has(blobId), uploading: this.willUpload.has(blobId) || this.uploading.has(blobId), downloading: this.willDownload.has(blobId) || this.downloading.has(blobId), @@ -430,6 +447,20 @@ class BlobSyncPeerStatus { private readonly statusUpdatedSubject$ = new Subject(); + markBlobToUpload(blobId: string) { + if (!this.toUpload.has(blobId)) { + this.toUpload.add(blobId); + this.statusUpdatedSubject$.next(blobId); + } + } + + markBlobToDownload(blobId: string) { + if (!this.toDownload.has(blobId)) { + this.toDownload.add(blobId); + this.statusUpdatedSubject$.next(blobId); + } + } + blobUploading(blobId: string) { if (!this.uploading.has(blobId)) { this.uploading.add(blobId); @@ -437,6 +468,15 @@ class BlobSyncPeerStatus { } } + blobUploadSuccess(blobId: string) { + this.blobUploadFinish(blobId); + this.blobErrorFree(blobId); + const deleted = this.toUpload.delete(blobId); + if (deleted) { + this.statusUpdatedSubject$.next(blobId); + } + } + blobUploadFinish(blobId: string) { let deleted = false; deleted = this.uploading.delete(blobId) || deleted; @@ -467,6 +507,15 @@ class BlobSyncPeerStatus { } } + blobDownloadSuccess(blobId: string) { + this.blobDownloadFinish(blobId); + this.blobErrorFree(blobId); + const deleted = this.toDownload.delete(blobId); + if (deleted) { + this.statusUpdatedSubject$.next(blobId); + } + } + blobDownloadFinish(blobId: string) { let deleted = false; deleted = this.willDownload.delete(blobId) || deleted; @@ -474,7 +523,6 @@ class BlobSyncPeerStatus { if (deleted) { this.statusUpdatedSubject$.next(blobId); } - this.blobErrorFree(blobId); } blobWillDownload(blobId: string) { @@ -529,4 +577,19 @@ class BlobSyncPeerStatus { this.statusUpdatedSubject$.next(blobId); } } + + blobStateClear(blobId: string) { + let deleted = false; + deleted = this.toUpload.delete(blobId) || deleted; + deleted = this.toDownload.delete(blobId) || deleted; + deleted = this.willUpload.delete(blobId) || deleted; + deleted = this.willDownload.delete(blobId) || deleted; + deleted = this.uploading.delete(blobId) || deleted; + deleted = this.downloading.delete(blobId) || deleted; + deleted = this.error.delete(blobId) || deleted; + deleted = this.overSize.delete(blobId) || deleted; + if (deleted) { + this.statusUpdatedSubject$.next(blobId); + } + } }