From 64f385817dea8becb8c5d79782d8ee38057cfe7d Mon Sep 17 00:00:00 2001 From: DarkSky <25152247+darkskygit@users.noreply.github.com> Date: Fri, 27 Mar 2026 19:50:53 +0800 Subject: [PATCH] feat(server): add perf metrics for apply update (#14736) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #### PR Dependency Tree * **PR #14736** 👈 This tree was auto-generated by [Charcoal](https://github.com/danerwilliams/charcoal) ## Summary by CodeRabbit * **Chores** * Enhanced Kubernetes deployment health check configurations with explicit timeout, period, failure threshold, and success threshold settings for improved reliability. * Improved document synchronization infrastructure with enhanced codec comparison and merge update capabilities for better data consistency handling. --- .../charts/graphql/templates/deployment.yaml | 8 + .../helm/affine/charts/graphql/values.yaml | 7 + .../__snapshots__/codec-compare.spec.ts.md | 57 +++ .../__snapshots__/codec-compare.spec.ts.snap | Bin 0 -> 840 bytes .../core/doc/__tests__/codec-compare.spec.ts | 358 +++++++++++++++++ .../server/src/core/doc/codec-compare.ts | 363 ++++++++++++++++++ .../server/src/core/doc/merge-updates.ts | 233 +++++++++++ .../backend/server/src/core/doc/options.ts | 79 +--- .../server/src/core/doc/storage/doc.ts | 18 +- .../backend/server/src/core/sync/gateway.ts | 8 +- 10 files changed, 1061 insertions(+), 70 deletions(-) create mode 100644 packages/backend/server/src/core/doc/__tests__/__snapshots__/codec-compare.spec.ts.md create mode 100644 packages/backend/server/src/core/doc/__tests__/__snapshots__/codec-compare.spec.ts.snap create mode 100644 packages/backend/server/src/core/doc/__tests__/codec-compare.spec.ts create mode 100644 packages/backend/server/src/core/doc/codec-compare.ts create mode 100644 packages/backend/server/src/core/doc/merge-updates.ts diff --git a/.github/helm/affine/charts/graphql/templates/deployment.yaml b/.github/helm/affine/charts/graphql/templates/deployment.yaml index 598cca53dd..d8ae967958 100644 --- a/.github/helm/affine/charts/graphql/templates/deployment.yaml +++ b/.github/helm/affine/charts/graphql/templates/deployment.yaml @@ -95,11 +95,19 @@ spec: path: /info port: http initialDelaySeconds: {{ .Values.probe.initialDelaySeconds }} + timeoutSeconds: {{ default .Values.probe.timeoutSeconds .Values.probe.liveness.timeoutSeconds }} + periodSeconds: {{ default .Values.probe.periodSeconds .Values.probe.liveness.periodSeconds }} + failureThreshold: {{ default .Values.probe.failureThreshold .Values.probe.liveness.failureThreshold }} + successThreshold: {{ default .Values.probe.successThreshold .Values.probe.liveness.successThreshold }} readinessProbe: httpGet: path: /info port: http initialDelaySeconds: {{ .Values.probe.initialDelaySeconds }} + timeoutSeconds: {{ default .Values.probe.timeoutSeconds .Values.probe.readiness.timeoutSeconds }} + periodSeconds: {{ default .Values.probe.periodSeconds .Values.probe.readiness.periodSeconds }} + failureThreshold: {{ default .Values.probe.failureThreshold .Values.probe.readiness.failureThreshold }} + successThreshold: {{ default .Values.probe.successThreshold .Values.probe.readiness.successThreshold }} resources: {{- toYaml .Values.resources | nindent 12 }} {{- with .Values.nodeSelector }} diff --git a/.github/helm/affine/charts/graphql/values.yaml b/.github/helm/affine/charts/graphql/values.yaml index 873ec68840..084489ea69 100644 --- a/.github/helm/affine/charts/graphql/values.yaml +++ b/.github/helm/affine/charts/graphql/values.yaml @@ -36,6 +36,13 @@ resources: probe: initialDelaySeconds: 20 + timeoutSeconds: 5 + periodSeconds: 10 + failureThreshold: 6 + successThreshold: 1 + liveness: + failureThreshold: 12 + readiness: {} nodeSelector: {} tolerations: [] diff --git a/packages/backend/server/src/core/doc/__tests__/__snapshots__/codec-compare.spec.ts.md b/packages/backend/server/src/core/doc/__tests__/__snapshots__/codec-compare.spec.ts.md new file mode 100644 index 0000000000..37058ab274 --- /dev/null +++ b/packages/backend/server/src/core/doc/__tests__/__snapshots__/codec-compare.spec.ts.md @@ -0,0 +1,57 @@ +# Snapshot report for `src/core/doc/__tests__/codec-compare.spec.ts` + +The actual snapshot is saved in `codec-compare.spec.ts.snap`. + +Generated by [AVA](https://avajs.dev). + +## compareCodecResult ignores map ordering differences + +> ignores map ordering differences + + { + matches: true, + } + +## compareCodecResult reports nested map text and xml changes + +> reports nested map text and xml changes + + { + matches: false, + treeDiff: [ + '$.blocks.meta.checked: yjs=false yocto=true', + '$.content.attrs.kind: yjs="intro" yocto="summary"', + '$.meta.info.flags.priority: yjs=3 yocto=7', + '$.meta.info.tags[1]: yjs="beta" yocto="gamma"', + '$.titles.main: yjs="Complex document" yocto="Complex document updated"', + ], + } + +## compareCodecResult reports weird mixed arrays nulls binary and nested text changes + +> reports weird mixed arrays nulls binary and nested text changes + + { + matches: false, + treeDiff: [ + '$.attachments[0].notes: yjs="first file" yocto="first file updated"', + '$.attachments[2]: yjs=true yocto=false', + '$.attachments[4] only exists in yocto: "tail"', + '$.meta.nested[0].labels[1]: yjs="y" yocto="changed"', + '$.rich.attrs.mood: yjs="calm" yocto="loud"', + '$.rich.children[0].children[0]: yjs="hello" yocto="hello there"', + ], + } + +## compareCodecResult reports root text and reordered array content changes + +> reports root text and reordered array content changes + + { + matches: false, + treeDiff: [ + '$.logline: yjs="affine" yocto="affine cloud"', + '$.queue: yjs="abc" yocto="acb"', + '$.settings.enabled: yjs=true yocto=false', + ], + } diff --git a/packages/backend/server/src/core/doc/__tests__/__snapshots__/codec-compare.spec.ts.snap b/packages/backend/server/src/core/doc/__tests__/__snapshots__/codec-compare.spec.ts.snap new file mode 100644 index 0000000000000000000000000000000000000000..77986e3e75d050b475c0b27793cecca52b9a8568 GIT binary patch literal 840 zcmV-O1GoG^RzV9cxUK#vxhS9?QYGzUISw*^! zsMuKc-XDty00000000BUltF77M-<23NS19%S}7#wFo-2>+-9ilgaj88ic2V=Q1GFo z7#~KnZ?zN8&ZskQiIiKQU!#Yn?VfSl>2Vjfp!%_G))4@34*+hk&FgIQ+tD*r#8`VA3ujr#q;d@Pz0*e{M`)&3eoWKI!6l7&oH$Q`TMUG0U4n2*T-uaR3&T||?f8;@zGr_g zHPqb$fQJC^-PKKX);2W&_7P(f0BGKClSlkYXIW*m`5OTI0RV3RV21%91NfCi)83cf z%Ngr5r+&W=>;qsuVAm?kw&tgan1?J1Bq>ym>I~BZK6@4pGid_ynN#Er$(I!`6 zXr$dndUsh$avV-}Powz=Nw(f&SroE$-e}~H6H&tuL?N|}!XsBq3{H9KR9T>6A@DhBpYlRaG3ByKFK}BL6ZlA5>1TDxXj5DKBbl9V-Updm zzXQOZ0MKOsUoe0d4B#&Y@b)S?H!sq8lvH3zl~;dxIQjfkSVtIQSef>Lc%}`uF)bdo z0j*Gb`+U*82u($JEK`HOg6iG#x_hVGSu^7})qw)nb}l~PLz3DIS8&~rxXzQSda21- zqITA{F;6sIg?1+qUaLHgTj5+YZc3V>l^a(sw7o|AC~8%%P3qBBpRja}g>p1zF}L-N zQ;r@R(jWa#Oq&V#dd&oQ=V&oL9_wA+*zmY>)@M; + children?: DocValue[]; +}; + +type DocXmlFragmentValue = { + $xmlFragment: DocValue[]; +}; + +type DocMapValue = { + [key: string]: DocValue; +}; + +type DocArrayValue = DocValue[]; + +type DocValue = + | DocScalar + | DocTextValue + | DocXmlTextValue + | DocXmlElementValue + | DocXmlFragmentValue + | DocMapValue + | DocArrayValue; + +type DocShape = Record; + +type CodecCompareCase = { + expectedMatches: boolean; + left: DocShape; + right: DocShape; + title: string; +}; + +function isPlainObject(value: unknown): value is Record { + return ( + typeof value === 'object' && + value !== null && + !Array.isArray(value) && + !(value instanceof Uint8Array) + ); +} + +function isTextValue(value: DocValue): value is DocTextValue { + return isPlainObject(value) && '$text' in value; +} + +function isXmlTextValue(value: DocValue): value is DocXmlTextValue { + return isPlainObject(value) && '$xmlText' in value; +} + +function isXmlElementValue(value: DocValue): value is DocXmlElementValue { + return isPlainObject(value) && '$xmlElement' in value; +} + +function isXmlFragmentValue(value: DocValue): value is DocXmlFragmentValue { + return isPlainObject(value) && '$xmlFragment' in value; +} + +function buildText(text: string) { + const yText = new Y.Text(); + yText.insert(0, text); + return yText; +} + +function appendXmlNode( + parent: Y.XmlElement | Y.XmlFragment, + value: DocValue +): void { + if (isXmlTextValue(value)) { + parent.push([new Y.XmlText(value.$xmlText)]); + return; + } + + if (!isXmlElementValue(value)) { + throw new Error(`Expected xml node, got ${JSON.stringify(value)}`); + } + + const element = new Y.XmlElement(value.$xmlElement); + parent.push([element]); + for (const [key, attr] of Object.entries(value.attrs ?? {})) { + element.setAttribute(key, String(attr)); + } + for (const child of value.children ?? []) { + appendXmlNode(element, child); + } +} + +function buildArray(values: DocArrayValue) { + const yArray = new Y.Array(); + yArray.push(values.map(value => buildSharedValue(value))); + return yArray; +} + +function buildMap(shape: DocMapValue) { + const yMap = new Y.Map(); + for (const [key, value] of Object.entries(shape)) { + yMap.set(key, buildSharedValue(value)); + } + return yMap; +} + +function buildSharedValue(value: DocValue): unknown { + if ( + value == null || + typeof value === 'boolean' || + typeof value === 'number' || + typeof value === 'string' || + value instanceof Uint8Array + ) { + return value; + } + + if (Array.isArray(value)) { + return buildArray(value); + } + + if (isTextValue(value)) { + return buildText(value.$text); + } + + if ( + isXmlFragmentValue(value) || + isXmlElementValue(value) || + isXmlTextValue(value) + ) { + throw new Error('XML values are only supported at root xmlFragment fields'); + } + + return buildMap(value); +} + +function applyRootValue(doc: Y.Doc, key: string, value: DocValue) { + if (isTextValue(value)) { + doc.getText(key).insert(0, value.$text); + return; + } + + if (isXmlFragmentValue(value)) { + const fragment = doc.getXmlFragment(key); + for (const child of value.$xmlFragment) { + appendXmlNode(fragment, child); + } + return; + } + + if (Array.isArray(value)) { + doc.getArray(key).push(value.map(item => buildSharedValue(item))); + return; + } + + if (isPlainObject(value)) { + const yMap = doc.getMap(key); + for (const [childKey, childValue] of Object.entries(value)) { + yMap.set(childKey, buildSharedValue(childValue as DocValue)); + } + return; + } + + throw new Error( + `Unsupported root value for "${key}": ${JSON.stringify(value)}` + ); +} + +function encodeDocFromShape(shape: DocShape): Buffer { + const doc = new Y.Doc(); + for (const [key, value] of Object.entries(shape)) { + applyRootValue(doc, key, value); + } + return Buffer.from(Y.encodeStateAsUpdate(doc)); +} + +const baseDoc: DocShape = { + blocks: [ + { + id: 'block-1', + meta: { checked: false, weight: 10 }, + type: 'todo', + }, + ], + content: { + $xmlFragment: [ + { + $xmlElement: 'paragraph', + attrs: { kind: 'intro' }, + children: [{ $xmlText: 'hello world' }], + }, + ], + }, + meta: { + info: { + flags: { archived: false, priority: 3 }, + owner: 'alice', + tags: ['alpha', 'beta'], + }, + version: 1, + }, + titles: { + main: { $text: 'Complex document' }, + }, +}; + +const compareCases: CodecCompareCase[] = [ + { + expectedMatches: true, + title: 'ignores map ordering differences', + left: baseDoc, + right: { + titles: { main: { $text: 'Complex document' } }, + content: { + $xmlFragment: [ + { + $xmlElement: 'paragraph', + attrs: { kind: 'intro' }, + children: [{ $xmlText: 'hello world' }], + }, + ], + }, + blocks: [ + { + type: 'todo', + meta: { weight: 10, checked: false }, + id: 'block-1', + }, + ], + meta: { + version: 1, + info: { + tags: ['alpha', 'beta'], + owner: 'alice', + flags: { priority: 3, archived: false }, + }, + }, + }, + }, + { + expectedMatches: false, + title: 'reports nested map text and xml changes', + left: baseDoc, + right: { + ...baseDoc, + blocks: [ + { id: 'block-1', meta: { checked: true, weight: 10 }, type: 'todo' }, + ], + content: { + $xmlFragment: [ + { + $xmlElement: 'paragraph', + attrs: { kind: 'summary' }, + children: [{ $xmlText: 'hello world' }], + }, + ], + }, + meta: { + info: { + flags: { archived: false, priority: 7 }, + owner: 'alice', + tags: ['alpha', 'gamma'], + }, + version: 1, + }, + titles: { main: { $text: 'Complex document updated' } }, + }, + }, + { + expectedMatches: false, + title: 'reports weird mixed arrays nulls binary and nested text changes', + left: { + attachments: [ + { + blob: new Uint8Array([1, 2, 3]), + filename: 'a.bin', + notes: { $text: 'first file' }, + }, + null, + true, + 42, + ], + meta: { nested: [{ labels: ['x', 'y'] }, { labels: ['z'] }] }, + rich: { + $xmlFragment: [ + { + $xmlElement: 'card', + attrs: { mood: 'calm' }, + children: [ + { $xmlElement: 'title', children: [{ $xmlText: 'hello' }] }, + ], + }, + ], + }, + }, + right: { + attachments: [ + { + blob: new Uint8Array([1, 2, 4]), + filename: 'a.bin', + notes: { $text: 'first file updated' }, + }, + null, + false, + 42, + 'tail', + ], + meta: { nested: [{ labels: ['x', 'changed'] }, { labels: ['z'] }] }, + rich: { + $xmlFragment: [ + { + $xmlElement: 'card', + attrs: { mood: 'loud' }, + children: [ + { $xmlElement: 'title', children: [{ $xmlText: 'hello there' }] }, + ], + }, + ], + }, + }, + }, + { + expectedMatches: false, + title: 'reports root text and reordered array content changes', + left: { + logline: { $text: 'affine' }, + queue: ['a', 'b', 'c'], + settings: { enabled: true, retries: 2 }, + }, + right: { + logline: { $text: 'affine cloud' }, + queue: ['a', 'c', 'b'], + settings: { enabled: false, retries: 2 }, + }, + }, +]; + +for (const testCase of compareCases) { + test(`compareCodecResult ${testCase.title}`, t => { + const leftBinary = encodeDocFromShape(testCase.left); + const rightBinary = encodeDocFromShape(testCase.right); + const result = compareCodecResult(leftBinary, rightBinary); + + t.is(result.matches, testCase.expectedMatches); + t.snapshot(result, testCase.title); + }); +} diff --git a/packages/backend/server/src/core/doc/codec-compare.ts b/packages/backend/server/src/core/doc/codec-compare.ts new file mode 100644 index 0000000000..3d30753a4a --- /dev/null +++ b/packages/backend/server/src/core/doc/codec-compare.ts @@ -0,0 +1,363 @@ +import * as Y from 'yjs'; + +const TREE_DIFF_LIMIT = 12; + +type ComparableTree = + | null + | boolean + | number + | string + | ComparableTree[] + | { [key: string]: ComparableTree }; + +export type CodecCompareResult = { + matches: boolean; + treeDiff?: string[]; +}; + +function loadAndReencodeWithYjs(binary: Buffer): Buffer { + const doc = new Y.Doc(); + Y.applyUpdate(doc, binary); + + return Buffer.from(Y.encodeStateAsUpdate(doc)); +} + +function sortObjectEntries( + object: Record +): Array<[string, ComparableTree]> { + return Object.entries(object) + .filter(([, value]) => value !== undefined) + .sort(([left], [right]) => left.localeCompare(right)) + .map(([key, value]) => [key, serializeYValue(value)]); +} + +function isPlainObject(value: object): value is Record { + return Object.getPrototypeOf(value) === Object.prototype; +} + +function getConstructorName(value: unknown): string | undefined { + if (!value || typeof value !== 'object') { + return undefined; + } + + return value.constructor?.name; +} + +function serializeItemContent(content: { + arr?: unknown[]; + embed?: unknown; + str?: string; + type?: unknown; +}): ComparableTree { + if (content.type !== undefined) { + return serializeYValue(content.type); + } + + if (Array.isArray(content.arr)) { + if (content.arr.length === 1) { + return serializeYValue(content.arr[0]); + } + + return content.arr.map(item => serializeYValue(item)); + } + + if (typeof content.str === 'string') { + return content.str; + } + + if (content.embed !== undefined) { + return serializeYValue(content.embed); + } + + return null; +} + +function serializeMapItems(items: Map) { + return Object.fromEntries( + Array.from(items.entries()) + .sort(([left], [right]) => left.localeCompare(right)) + .map(([key, item]) => [ + key, + serializeItemContent( + item.content as Parameters[0] + ), + ]) + ); +} + +function serializeSequenceItems( + start: { + content: unknown; + deleted?: boolean; + right: unknown; + } | null +): ComparableTree[] { + const result: ComparableTree[] = []; + let current = start; + + while (current) { + if (!current.deleted) { + const content = current.content as Parameters< + typeof serializeItemContent + >[0]; + const serialized = serializeItemContent(content); + + if (Array.isArray(serialized) && Array.isArray(content.arr)) { + result.push(...serialized); + } else { + result.push(serialized); + } + } + current = current.right as typeof current; + } + + return result; +} + +function serializeAbstractType(value: { + _map?: Map; + _start?: { + content: unknown; + deleted?: boolean; + right: unknown; + } | null; + nodeName?: string; +}) { + const map = value._map ? serializeMapItems(value._map) : {}; + const children = serializeSequenceItems(value._start ?? null); + const constructorName = getConstructorName(value); + + if (value.nodeName || constructorName === 'YXmlElement') { + return { + nodeName: value.nodeName ?? 'xml', + ...(Object.keys(map).length ? { attrs: map } : null), + ...(children.length ? { children } : null), + }; + } + + if (Object.keys(map).length && !children.length) { + return map; + } + + if (!Object.keys(map).length) { + if (constructorName === 'YArray') { + return children; + } + + if (constructorName === 'YText' || constructorName === 'YXmlText') { + return children.join(''); + } + + if (children.every(child => typeof child === 'string')) { + return children.join(''); + } + + if (children.length === 1) { + return children[0] ?? null; + } + + return children; + } + + return { + ...map, + children, + }; +} + +function serializeYValue(value: unknown): ComparableTree { + if (value == null) { + return null; + } + + if ( + typeof value === 'boolean' || + typeof value === 'number' || + typeof value === 'string' + ) { + return value; + } + + if (value instanceof Uint8Array) { + return `Uint8Array(${value.length})`; + } + + const constructorName = getConstructorName(value); + + // Ignore map field storage order. Compare sorted keys only. + if ( + value instanceof Y.Map || + constructorName === 'YMap' || + value instanceof Y.Array || + constructorName === 'YArray' || + value instanceof Y.Text || + constructorName === 'YText' || + value instanceof Y.XmlText || + constructorName === 'YXmlText' || + value instanceof Y.XmlElement || + constructorName === 'YXmlElement' || + value instanceof Y.XmlFragment || + constructorName === 'YXmlFragment' || + (typeof value === 'object' && + value !== null && + ('_map' in value || '_start' in value)) + ) { + return serializeAbstractType( + value as Parameters[0] + ); + } + + if (Array.isArray(value)) { + return value.map(item => serializeYValue(item)); + } + + if (typeof value === 'object' && isPlainObject(value)) { + return Object.fromEntries(sortObjectEntries(value)); + } + + if (typeof value === 'object') { + return Object.prototype.toString.call(value); + } + + return String(value); +} + +function serializeYDoc(binary: Buffer): ComparableTree { + const doc = new Y.Doc(); + Y.applyUpdate(doc, binary); + + // Ignore top-level share/map ordering too. + return Object.fromEntries( + Array.from(doc.share.entries()) + .sort(([left], [right]) => left.localeCompare(right)) + .map(([key, value]) => [key, serializeYValue(value)]) + ); +} + +export function serializeCodecBinaryForDebug(binary: Buffer): ComparableTree { + return serializeYDoc(binary); +} + +function formatTreeValue(value: ComparableTree): string { + const text = JSON.stringify(value); + if (!text) return String(value); + return text.length > 120 ? `${text.slice(0, 117)}...` : text; +} + +function diffTrees( + yjsTree: ComparableTree, + yoctoTree: ComparableTree, + path = '$', + diffs: string[] = [] +): string[] { + if (diffs.length >= TREE_DIFF_LIMIT) { + return diffs; + } + + if (Object.is(yjsTree, yoctoTree)) { + return diffs; + } + + if (Array.isArray(yjsTree) && Array.isArray(yoctoTree)) { + const maxLength = Math.max(yjsTree.length, yoctoTree.length); + for (let index = 0; index < maxLength; index += 1) { + if (diffs.length >= TREE_DIFF_LIMIT) { + break; + } + const yjsValue = yjsTree[index]; + const yoctoValue = yoctoTree[index]; + if (index >= yjsTree.length) { + diffs.push( + `${path}[${index}] only exists in yocto: ${formatTreeValue(yoctoValue ?? null)}` + ); + continue; + } + if (index >= yoctoTree.length) { + diffs.push( + `${path}[${index}] only exists in yjs: ${formatTreeValue(yjsValue ?? null)}` + ); + continue; + } + diffTrees( + yjsValue ?? null, + yoctoValue ?? null, + `${path}[${index}]`, + diffs + ); + } + + return diffs; + } + + if ( + yjsTree && + yoctoTree && + typeof yjsTree === 'object' && + typeof yoctoTree === 'object' && + !Array.isArray(yjsTree) && + !Array.isArray(yoctoTree) + ) { + const keys = Array.from( + new Set([...Object.keys(yjsTree), ...Object.keys(yoctoTree)]) + ).sort((left, right) => left.localeCompare(right)); + + for (const key of keys) { + if (diffs.length >= TREE_DIFF_LIMIT) { + break; + } + const yjsValue = yjsTree[key]; + const yoctoValue = yoctoTree[key]; + if (!(key in yjsTree)) { + diffs.push( + `${path}.${key} only exists in yocto: ${formatTreeValue(yoctoValue ?? null)}` + ); + continue; + } + if (!(key in yoctoTree)) { + diffs.push( + `${path}.${key} only exists in yjs: ${formatTreeValue(yjsValue ?? null)}` + ); + continue; + } + diffTrees(yjsValue ?? null, yoctoValue ?? null, `${path}.${key}`, diffs); + } + + return diffs; + } + + diffs.push( + `${path}: yjs=${formatTreeValue(yjsTree)} yocto=${formatTreeValue(yoctoTree)}` + ); + + return diffs; +} + +export function compareCodecResult( + yBinary: Buffer, + yoctoBinary: Buffer +): CodecCompareResult { + if (yBinary.equals(yoctoBinary)) { + return { matches: true }; + } + + const normalizedYBinary = loadAndReencodeWithYjs(yBinary); + const normalizedYoctoBinary = loadAndReencodeWithYjs(yoctoBinary); + + if (normalizedYBinary.equals(normalizedYoctoBinary)) { + return { matches: true }; + } + + const treeDiff = diffTrees( + serializeYDoc(normalizedYBinary), + serializeYDoc(normalizedYoctoBinary) + ); + + if (treeDiff.length === 0) { + return { matches: true }; + } + + return { + matches: false, + treeDiff, + }; +} diff --git a/packages/backend/server/src/core/doc/merge-updates.ts b/packages/backend/server/src/core/doc/merge-updates.ts new file mode 100644 index 0000000000..b491dbf4c3 --- /dev/null +++ b/packages/backend/server/src/core/doc/merge-updates.ts @@ -0,0 +1,233 @@ +import type { Logger } from '@nestjs/common'; +import { chunk } from 'lodash-es'; +import * as Y from 'yjs'; +import { mergeUpdates as yjsMergeUpdatesInCore } from 'yjs'; + +import { metrics } from '../../base'; +import { mergeUpdatesInApplyWay as nativeMergeUpdatesInApplyWay } from '../../native'; + +const SLOW_DOC_UPDATE_CODEC_MS = 250; + +type CodecEngine = 'native' | 'yjs'; +type MetricStatus = 'ok' | 'error' | 'partial_error'; +type CodecLogger = Pick; + +function recordCodecMetrics( + operation: 'apply_updates' | 'merge_updates', + engine: CodecEngine, + caller: string, + status: MetricStatus, + duration: number +) { + metrics.doc.histogram(`${operation}_duration`).record(duration, { + engine, + caller, + status, + }); + metrics.doc.counter(`${operation}_calls`).add(1, { + engine, + caller, + status, + }); +} + +function warnSlowCodecOperation( + operation: 'apply_updates' | 'merge_updates', + engine: CodecEngine, + caller: string, + duration: number, + updateCount: number, + logger?: Pick +) { + if (duration < SLOW_DOC_UPDATE_CODEC_MS) { + return; + } + + metrics.doc.counter(`${operation}_slow`).add(1, { + engine, + caller, + }); + logger?.warn( + `Slow ${engine} ${operation} call in ${caller}: ${duration.toFixed(1)}ms for ${updateCount} updates` + ); +} + +function measureSyncCodecOperation( + operation: 'merge_updates', + engine: CodecEngine, + updates: Uint8Array[], + caller: string, + logger: Pick | undefined, + fn: () => T +): T { + const start = performance.now(); + + try { + const result = fn(); + const duration = performance.now() - start; + + recordCodecMetrics(operation, engine, caller, 'ok', duration); + warnSlowCodecOperation( + operation, + engine, + caller, + duration, + updates.length, + logger + ); + + return result; + } catch (error) { + const duration = performance.now() - start; + + recordCodecMetrics(operation, engine, caller, 'error', duration); + throw error; + } +} + +async function measureAsyncCodecOperation( + operation: 'merge_updates', + engine: CodecEngine, + updates: Uint8Array[], + caller: string, + logger: Pick | undefined, + fn: () => Promise +): Promise { + const start = performance.now(); + + try { + const result = await fn(); + const duration = performance.now() - start; + + recordCodecMetrics(operation, engine, caller, 'ok', duration); + warnSlowCodecOperation( + operation, + engine, + caller, + duration, + updates.length, + logger + ); + + return result; + } catch (error) { + const duration = performance.now() - start; + + recordCodecMetrics(operation, engine, caller, 'error', duration); + throw error; + } +} + +export function applyUpdatesWithNative( + updates: Uint8Array[], + caller: string, + logger?: Pick +) { + return measureSyncCodecOperation( + 'merge_updates', + 'native', + updates, + caller, + logger, + () => nativeMergeUpdatesInApplyWay(updates.map(Buffer.from)) + ); +} + +export function mergeUpdatesWithYjs( + updates: Uint8Array[], + caller: string, + logger?: Pick +) { + return measureSyncCodecOperation( + 'merge_updates', + 'yjs', + updates, + caller, + logger, + () => Buffer.from(yjsMergeUpdatesInCore(updates)) + ); +} + +async function recoverDocWithYjs( + updates: Uint8Array[], + caller: string, + logger?: CodecLogger +): Promise { + const doc = new Y.Doc(); + const chunks = chunk(updates, 10); + let i = 0; + let failedUpdates = 0; + const start = performance.now(); + + try { + await new Promise(resolve => { + Y.transact(doc, () => { + const next = () => { + const nextUpdates = chunks.at(i++); + + if (nextUpdates?.length) { + nextUpdates.forEach(update => { + try { + Y.applyUpdate(doc, update); + } catch (error) { + failedUpdates += 1; + metrics.doc.counter('apply_update_failures').add(1, { + engine: 'yjs', + caller, + }); + logger?.error('Failed to apply update', error); + } + }); + + // avoid applying too many updates in single round which will take the whole cpu time like dead lock + setImmediate(() => { + next(); + }); + } else { + resolve(); + } + }; + + next(); + }); + }); + } catch (error) { + const duration = performance.now() - start; + + recordCodecMetrics('apply_updates', 'yjs', caller, 'error', duration); + throw error; + } + + const duration = performance.now() - start; + const status = failedUpdates > 0 ? 'partial_error' : 'ok'; + + recordCodecMetrics('apply_updates', 'yjs', caller, status, duration); + warnSlowCodecOperation( + 'apply_updates', + 'yjs', + caller, + duration, + updates.length, + logger + ); + + return doc; +} + +export async function applyUpdatesWithYjs( + updates: Uint8Array[], + caller: string, + logger?: CodecLogger +) { + return measureAsyncCodecOperation( + 'merge_updates', + 'yjs', + updates, + caller, + logger, + async () => { + const doc = await recoverDocWithYjs(updates, caller, logger); + return Buffer.from(Y.encodeStateAsUpdate(doc)); + } + ); +} diff --git a/packages/backend/server/src/core/doc/options.ts b/packages/backend/server/src/core/doc/options.ts index 75e7879f94..f4377c04a3 100644 --- a/packages/backend/server/src/core/doc/options.ts +++ b/packages/backend/server/src/core/doc/options.ts @@ -1,29 +1,11 @@ import { Injectable, Logger } from '@nestjs/common'; -import { chunk } from 'lodash-es'; -import * as Y from 'yjs'; -import { CallMetric, Config, metrics } from '../../base'; -import { mergeUpdatesInApplyWay as yoctoMergeUpdates } from '../../native'; +import { Config, metrics } from '../../base'; import { QuotaService } from '../quota'; +import { compareCodecResult } from './codec-compare'; +import { applyUpdatesWithNative, applyUpdatesWithYjs } from './merge-updates'; import { DocStorageOptions as IDocStorageOptions } from './storage'; -function compare(yBinary: Buffer, jwstBinary: Buffer, strict = false): boolean { - if (yBinary.equals(jwstBinary)) { - return true; - } - - if (strict) { - return false; - } - - const doc = new Y.Doc(); - Y.applyUpdate(doc, jwstBinary); - - const yBinary2 = Buffer.from(Y.encodeStateAsUpdate(doc)); - - return compare(yBinary, yBinary2, true); -} - @Injectable() export class DocStorageOptions implements IDocStorageOptions { private readonly logger = new Logger('DocStorageOptions'); @@ -34,18 +16,31 @@ export class DocStorageOptions implements IDocStorageOptions { ) {} mergeUpdates = async (updates: Uint8Array[]) => { - const doc = await this.recoverDoc(updates); - const yjsResult = Buffer.from(Y.encodeStateAsUpdate(doc)); + const yjsResult = await applyUpdatesWithYjs( + updates, + 'doc.options.merge_updates', + this.logger + ); if (this.config.doc.experimental.yocto) { metrics.jwst.counter('codec_merge_counter').add(1); let log = false; let yoctoResult: Buffer | null = null; try { - yoctoResult = yoctoMergeUpdates(updates.map(Buffer.from)); - if (!compare(yjsResult, yoctoResult)) { + yoctoResult = applyUpdatesWithNative( + updates, + 'doc.options.yocto_codec_compare', + this.logger + ); + const comparison = compareCodecResult(yjsResult, yoctoResult); + if (!comparison.matches) { metrics.jwst.counter('codec_not_match').add(1); this.logger.warn(`yocto codec result doesn't match yjs codec result`); + if (comparison.treeDiff?.length) { + this.logger.warn( + `yocto codec tree diff:\n${comparison.treeDiff.join('\n')}` + ); + } log = true; if (env.dev) { this.logger.warn(`Expected:\n ${yjsResult.toString('hex')}`); @@ -84,38 +79,4 @@ export class DocStorageOptions implements IDocStorageOptions { historyMinInterval = (_spaceId: string) => { return this.config.doc.history.interval; }; - - @CallMetric('doc', 'yjs_recover_updates_to_doc') - private recoverDoc(updates: Uint8Array[]): Promise { - const doc = new Y.Doc(); - const chunks = chunk(updates, 10); - let i = 0; - - return new Promise(resolve => { - Y.transact(doc, () => { - const next = () => { - const updates = chunks.at(i++); - - if (updates?.length) { - updates.forEach(u => { - try { - Y.applyUpdate(doc, u); - } catch (e) { - this.logger.error('Failed to apply update', e); - } - }); - - // avoid applying too many updates in single round which will take the whole cpu time like dead lock - setImmediate(() => { - next(); - }); - } else { - resolve(doc); - } - }; - - next(); - }); - }); - } } diff --git a/packages/backend/server/src/core/doc/storage/doc.ts b/packages/backend/server/src/core/doc/storage/doc.ts index 882537dc15..fc204d96de 100644 --- a/packages/backend/server/src/core/doc/storage/doc.ts +++ b/packages/backend/server/src/core/doc/storage/doc.ts @@ -8,18 +8,20 @@ import { encodeStateAsUpdate, encodeStateVector, encodeStateVectorFromUpdate, - mergeUpdates, UndoManager, } from 'yjs'; import { CallMetric } from '../../../base'; -import { mergeUpdatesInApplyWay } from '../../../native'; +import { applyUpdatesWithNative, mergeUpdatesWithYjs } from '../merge-updates'; import { Connection } from './connection'; import { SingletonLocker } from './lock'; -async function nativeMergeUpdates(updates: Uint8Array[]): Promise { - // use native module to merge updates - return mergeUpdatesInApplyWay(updates.map(u => Buffer.from(u))); +async function nativeApplyUpdates(updates: Uint8Array[]): Promise { + return applyUpdatesWithNative(updates, 'doc.storage.squash.native'); +} + +async function yjsMergeUpdates(updates: Uint8Array[]): Promise { + return mergeUpdatesWithYjs(updates, 'doc.storage.squash.yjs'); } export interface DocRecord { @@ -62,7 +64,7 @@ export abstract class DocStorageAdapter extends Connection { constructor( protected readonly options: DocStorageOptions = { - mergeUpdates, + mergeUpdates: yjsMergeUpdates, } ) { super(); @@ -114,7 +116,7 @@ export abstract class DocStorageAdapter extends Connection { if (updates.length) { const docUpdate = await this.squash( snapshot ? [snapshot, ...updates] : updates, - nativeMergeUpdates + nativeApplyUpdates ); return docUpdate.bin; } @@ -254,7 +256,7 @@ export abstract class DocStorageAdapter extends Connection { updates: DocUpdate[], merge?: (updates: Uint8Array[]) => Promise ): Promise { - const mergeFn = merge ?? this.options?.mergeUpdates ?? mergeUpdates; + const mergeFn = merge ?? this.options?.mergeUpdates ?? yjsMergeUpdates; const lastUpdate = updates.at(-1); if (!lastUpdate) { throw new Error('No updates to be squashed.'); diff --git a/packages/backend/server/src/core/sync/gateway.ts b/packages/backend/server/src/core/sync/gateway.ts index fec4cd5e2b..5bc14effae 100644 --- a/packages/backend/server/src/core/sync/gateway.ts +++ b/packages/backend/server/src/core/sync/gateway.ts @@ -32,7 +32,6 @@ import { SpaceAccessDenied, } from '../../base'; import { Models } from '../../models'; -import { mergeUpdatesInApplyWay } from '../../native'; import { CurrentUser } from '../auth'; import { DocReader, @@ -40,6 +39,7 @@ import { PgUserspaceDocStorageAdapter, PgWorkspaceDocStorageAdapter, } from '../doc'; +import { applyUpdatesWithNative } from '../doc/merge-updates'; import { AccessController, WorkspaceAction } from '../permission'; import { DocID } from '../utils/doc'; @@ -268,8 +268,10 @@ export class SpaceSyncGateway } try { - const merged = mergeUpdatesInApplyWay( - updates.map(update => Buffer.from(update)) + const merged = applyUpdatesWithNative( + updates, + 'socketio.broadcast', + this.logger ); metrics.socketio.counter('doc_updates_compressed').add(1); return {