luulam/AFFiNE
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix(core): merge updates before push to storage (#4986)

Browse Source
This commit is contained in:
EYHN
2023-11-20 23:26:19 +08:00
committed by GitHub
parent 9370110cdc
commit 90c130cf15
3 changed files with 31 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
packages/frontend/workspace/src/providers/storage/indexeddb/index.ts
View File

@@ -13,6 +13,12 @@ export const dbVersion = 1;
export const DEFAULT_DB_NAME = 'affine-local'; export const DEFAULT_DB_NAME = 'affine-local';
export function mergeUpdates(updates: Uint8Array[]) { export function mergeUpdates(updates: Uint8Array[]) {
if (updates.length === 0) {
return new Uint8Array();
}
if (updates.length === 1) {
return updates[0];
}
const doc = new Doc(); const doc = new Doc();
doc.transact(() => { doc.transact(() => {
updates.forEach(update => { updates.forEach(update => {

31
packages/frontend/workspace/src/providers/sync/peer.ts
View File

@@ -4,7 +4,7 @@ import { isEqual } from '@blocksuite/global/utils';
import type { Doc } from 'yjs'; import type { Doc } from 'yjs';
import { applyUpdate, encodeStateAsUpdate, encodeStateVector } from 'yjs'; import { applyUpdate, encodeStateAsUpdate, encodeStateVector } from 'yjs';
import type { Storage } from '../storage'; import { mergeUpdates, type Storage } from '../storage';
import { AsyncQueue } from '../utils/async-queue'; import { AsyncQueue } from '../utils/async-queue';
import { throwIfAborted } from '../utils/throw-if-aborted'; import { throwIfAborted } from '../utils/throw-if-aborted';
import { MANUALLY_STOP } from './engine'; import { MANUALLY_STOP } from './engine';
@@ -154,7 +154,7 @@ export class SyncPeer {
connectedDocs: Map<string, Doc>; connectedDocs: Map<string, Doc>;
pushUpdatesQueue: AsyncQueue<{ pushUpdatesQueue: AsyncQueue<{
docId: string; docId: string;
data: Uint8Array; data: Uint8Array[];
}>; }>;
pushingUpdate: boolean; pushingUpdate: boolean;
pullUpdatesQueue: AsyncQueue<{ pullUpdatesQueue: AsyncQueue<{
@@ -262,14 +262,16 @@ export class SyncPeer {
this.state.pushingUpdate = true; this.state.pushingUpdate = true;
this.reportSyncStatus(); this.reportSyncStatus();
const merged = mergeUpdates(data);
// don't push empty data or Uint8Array([0, 0]) // don't push empty data or Uint8Array([0, 0])
if ( if (
!( !(
data.byteLength === 0 || merged.byteLength === 0 ||
(data.byteLength === 2 && data[0] === 0 && data[1] === 0) (merged.byteLength === 2 && merged[0] === 0 && merged[1] === 0)
) )
) { ) {
await this.storage.push(docId, data); await this.storage.push(docId, merged);
} }
this.state.pushingUpdate = false; this.state.pushingUpdate = false;
@@ -298,7 +300,7 @@ export class SyncPeer {
// diff root doc and in-storage, save updates to pendingUpdates // diff root doc and in-storage, save updates to pendingUpdates
this.state.pushUpdatesQueue.push({ this.state.pushUpdatesQueue.push({
docId: doc.guid, docId: doc.guid,
data: encodeStateAsUpdate(doc, inStorageState), data: [encodeStateAsUpdate(doc, inStorageState)],
}); });
this.state.connectedDocs.set(doc.guid, doc); this.state.connectedDocs.set(doc.guid, doc);
@@ -324,10 +326,19 @@ export class SyncPeer {
if (origin === this.name) { if (origin === this.name) {
return; return;
} }
this.state.pushUpdatesQueue.push({
docId: doc.guid, const exist = this.state.pushUpdatesQueue.find(
data: update, ({ docId }) => docId === doc.guid
}); );
if (exist) {
exist.data.push(update);
} else {
this.state.pushUpdatesQueue.push({
docId: doc.guid,
data: [update],
});
}
this.reportSyncStatus(); this.reportSyncStatus();
}; };

4
packages/frontend/workspace/src/providers/utils/async-queue.ts
View File

@@ -56,6 +56,10 @@ export class AsyncQueue<T> {
} }
} }
find(predicate: (update: T) => boolean) {
return this._queue.find(predicate);
}
clear() { clear() {
this._queue = []; this._queue = [];
} }