Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

perf(core): only full sync before exporting #10408

Merged
merged 1 commit into from
Feb 25, 2025
Merged

perf(core): only full sync before exporting #10408

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions packages/common/nbstore/src/frontend/blob.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,12 @@ export class BlobFrontend {
return this.sync.uploadBlob(blob);
}

fullSync() {
return this.sync.fullSync();
fullDownload() {
return this.sync.fullDownload();
}

fullUpload() {
return this.sync.fullUpload();
}

addPriority(_id: string, _priority: number) {
Expand Down
178 changes: 106 additions & 72 deletions packages/common/nbstore/src/sync/blob/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@ import type { PeerStorageOptions } from '../types';

export interface BlobSyncState {
isStorageOverCapacity: boolean;
total: number;
synced: number;
}

export interface BlobSync {
Expand All @@ -18,14 +20,17 @@ export interface BlobSync {
signal?: AbortSignal
): Promise<BlobRecord | null>;
uploadBlob(blob: BlobRecord, signal?: AbortSignal): Promise<void>;
fullSync(signal?: AbortSignal): Promise<void>;
fullDownload(signal?: AbortSignal): Promise<void>;
fullUpload(signal?: AbortSignal): Promise<void>;
setMaxBlobSize(size: number): void;
onReachedMaxBlobSize(cb: (byteSize: number) => void): () => void;
}

export class BlobSyncImpl implements BlobSync {
readonly state$ = new BehaviorSubject<BlobSyncState>({
isStorageOverCapacity: false,
total: Object.values(this.storages.remotes).length ? 1 : 0,
synced: 0,
});
private abort: AbortController | null = null;
private maxBlobSize: number = 1024 * 1024 * 100; // 100MB
Expand All @@ -34,19 +39,24 @@ export class BlobSyncImpl implements BlobSync {
constructor(readonly storages: PeerStorageOptions<BlobStorage>) {}

async downloadBlob(blobId: string, signal?: AbortSignal) {
const localBlob = await this.storages.local.get(blobId, signal);
if (localBlob) {
return localBlob;
}
try {
const localBlob = await this.storages.local.get(blobId, signal);
if (localBlob) {
return localBlob;
}

for (const storage of Object.values(this.storages.remotes)) {
const data = await storage.get(blobId, signal);
if (data) {
await this.storages.local.set(data, signal);
return data;
for (const storage of Object.values(this.storages.remotes)) {
const data = await storage.get(blobId, signal);
if (data) {
await this.storages.local.set(data, signal);
return data;
}
}
return null;
} catch (e) {
console.error('error when download blob', e);
return null;
}
return null;
}

async uploadBlob(blob: BlobRecord, signal?: AbortSignal) {
Expand All @@ -62,79 +72,107 @@ export class BlobSyncImpl implements BlobSync {
return await remote.set(blob, signal);
} catch (err) {
if (err instanceof OverCapacityError) {
this.state$.next({ isStorageOverCapacity: true });
this.state$.next({
isStorageOverCapacity: true,
total: this.state$.value.total,
synced: this.state$.value.synced,
});
}
throw err;
}
})
);
}

async fullSync(signal?: AbortSignal) {
async fullDownload(signal?: AbortSignal) {
throwIfAborted(signal);

await this.storages.local.connection.waitForConnected(signal);
const localList = (await this.storages.local.list(signal)).map(b => b.key);
this.state$.next({
...this.state$.value,
synced: localList.length,
});

for (const [remotePeer, remote] of Object.entries(this.storages.remotes)) {
let localList: string[] = [];
let remoteList: string[] = [];
await Promise.allSettled(
Object.entries(this.storages.remotes).map(
async ([remotePeer, remote]) => {
await remote.connection.waitForConnected(signal);

await remote.connection.waitForConnected(signal);
const remoteList = (await remote.list(signal)).map(b => b.key);

try {
localList = (await this.storages.local.list(signal)).map(b => b.key);
throwIfAborted(signal);
remoteList = (await remote.list(signal)).map(b => b.key);
throwIfAborted(signal);
} catch (err) {
if (err === MANUALLY_STOP) {
throw err;
}
console.error(`error when sync`, err);
continue;
}
this.state$.next({
...this.state$.value,
total: Math.max(this.state$.value.total, remoteList.length),
});

const needUpload = difference(localList, remoteList);
for (const key of needUpload) {
try {
const data = await this.storages.local.get(key, signal);
throwIfAborted(signal);
if (data) {
await remote.set(data, signal);
throwIfAborted(signal);
}
} catch (err) {
if (err === MANUALLY_STOP) {
throw err;

const needDownload = difference(remoteList, localList);
for (const key of needDownload) {
try {
const data = await remote.get(key, signal);
throwIfAborted(signal);
if (data) {
await this.storages.local.set(data, signal);
this.state$.next({
...this.state$.value,
synced: this.state$.value.synced + 1,
});
throwIfAborted(signal);
}
} catch (err) {
if (err === MANUALLY_STOP) {
throw err;
}
console.error(
`error when sync ${key} from [${remotePeer}] to [local]`,
err
);
}
}
console.error(
`error when sync ${key} from [local] to [${remotePeer}]`,
err
);
}
}
)
);
}

const needDownload = difference(remoteList, localList);
async fullUpload(signal?: AbortSignal) {
throwIfAborted(signal);

await this.storages.local.connection.waitForConnected(signal);
const localList = (await this.storages.local.list(signal)).map(b => b.key);

await Promise.allSettled(
Object.entries(this.storages.remotes).map(
async ([remotePeer, remote]) => {
await remote.connection.waitForConnected(signal);

const remoteList = (await remote.list(signal)).map(b => b.key);

for (const key of needDownload) {
try {
const data = await remote.get(key, signal);
throwIfAborted(signal);
if (data) {
await this.storages.local.set(data, signal);
throwIfAborted(signal);
}
} catch (err) {
if (err === MANUALLY_STOP) {
throw err;

const needUpload = difference(localList, remoteList);
for (const key of needUpload) {
try {
const data = await this.storages.local.get(key, signal);
throwIfAborted(signal);
if (data) {
await remote.set(data, signal);
throwIfAborted(signal);
}
} catch (err) {
if (err === MANUALLY_STOP) {
throw err;
}
console.error(
`error when sync ${key} from [local] to [${remotePeer}]`,
err
);
}
}
console.error(
`error when sync ${key} from [${remotePeer}] to [local]`,
err
);
}
}
}
)
);
}

start() {
Expand All @@ -144,16 +182,12 @@ export class BlobSyncImpl implements BlobSync {

const abort = new AbortController();
this.abort = abort;

// TODO(@eyhn): fix this, large blob may cause iOS to crash?
if (!BUILD_CONFIG.isIOS) {
this.fullSync(abort.signal).catch(error => {
if (error === MANUALLY_STOP) {
return;
}
console.error('sync blob error', error);
});
}
this.fullUpload(abort.signal).catch(error => {
if (error === MANUALLY_STOP) {
return;
}
console.error('sync blob error', error);
});
}

stop() {
Expand Down
31 changes: 14 additions & 17 deletions packages/common/nbstore/src/worker/client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -257,26 +257,23 @@ class WorkerBlobSync implements BlobSync {
uploadBlob(blob: BlobRecord, _signal?: AbortSignal): Promise<void> {
return this.client.call('blobSync.uploadBlob', blob);
}
fullSync(signal?: AbortSignal): Promise<void> {
return new Promise((resolve, reject) => {
const abortListener = () => {
reject(signal?.reason);
subscription.unsubscribe();
};
fullDownload(signal?: AbortSignal): Promise<void> {
const download = this.client.call('blobSync.fullDownload');

signal?.addEventListener('abort', abortListener);
signal?.addEventListener('abort', () => {
download.cancel();
});

const subscription = this.client.ob$('blobSync.fullSync').subscribe({
next() {
signal?.removeEventListener('abort', abortListener);
resolve();
},
error(err) {
signal?.removeEventListener('abort', abortListener);
reject(err);
},
});
return download;
}
fullUpload(signal?: AbortSignal): Promise<void> {
const upload = this.client.call('blobSync.fullUpload');

signal?.addEventListener('abort', () => {
upload.cancel();
});

return upload;
}
}

Expand Down
18 changes: 4 additions & 14 deletions packages/common/nbstore/src/worker/consumer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,20 +234,10 @@ class StoreConsumer {
'docSync.resetSync': () => this.docSync.resetSync(),
'blobSync.downloadBlob': key => this.blobSync.downloadBlob(key),
'blobSync.uploadBlob': blob => this.blobSync.uploadBlob(blob),
'blobSync.fullSync': () =>
new Observable(subscriber => {
const abortController = new AbortController();
this.blobSync
.fullSync(abortController.signal)
.then(() => {
subscriber.next(true);
subscriber.complete();
})
.catch(error => {
subscriber.error(error);
});
return () => abortController.abort(MANUALLY_STOP);
}),
'blobSync.fullDownload': (_, { signal }) =>
this.blobSync.fullDownload(signal),
'blobSync.fullUpload': (_, { signal }) =>
this.blobSync.fullUpload(signal),
'blobSync.state': () => this.blobSync.state$,
'blobSync.setMaxBlobSize': size => this.blobSync.setMaxBlobSize(size),
'blobSync.onReachedMaxBlobSize': () =>
Expand Down
3 changes: 2 additions & 1 deletion packages/common/nbstore/src/worker/ops.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,8 @@ interface GroupedWorkerOps {
blobSync: {
downloadBlob: [string, BlobRecord | null];
uploadBlob: [BlobRecord, void];
fullSync: [void, boolean];
fullDownload: [void, void];
fullUpload: [void, void];
setMaxBlobSize: [number, void];
onReachedMaxBlobSize: [void, number];
state: [void, BlobSyncState];
Expand Down
Loading
Loading