feat(server): add perf metrics for apply update (#14736)

#### PR Dependency Tree


* **PR #14736** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## 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.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

.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 }}

.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: []

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',
+      ],
+    }

packages/backend/server/src/core/doc/__tests__/codec-compare.spec.ts

@@ -0,0 +1,358 @@
+import test from 'ava';
+import * as Y from 'yjs';
+
+import { compareCodecResult } from '../codec-compare';
+
+type DocScalar = null | boolean | number | string | Uint8Array;
+
+type DocTextValue = {
+  $text: string;
+};
+
+type DocXmlTextValue = {
+  $xmlText: string;
+};
+
+type DocXmlElementValue = {
+  $xmlElement: string;
+  attrs?: Record<string, boolean | number | string>;
+  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<string, DocValue>;
+
+type CodecCompareCase = {
+  expectedMatches: boolean;
+  left: DocShape;
+  right: DocShape;
+  title: string;
+};
+
+function isPlainObject(value: unknown): value is Record<string, unknown> {
+  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<unknown>();
+  yArray.push(values.map(value => buildSharedValue(value)));
+  return yArray;
+}
+
+function buildMap(shape: DocMapValue) {
+  const yMap = new Y.Map<unknown>();
+  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);
+  });
+}

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<string, unknown>
+): 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<string, unknown> {
+  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<string, { content: unknown }>) {
+  return Object.fromEntries(
+    Array.from(items.entries())
+      .sort(([left], [right]) => left.localeCompare(right))
+      .map(([key, item]) => [
+        key,
+        serializeItemContent(
+          item.content as Parameters<typeof serializeItemContent>[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<string, { content: unknown }>;
+  _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<typeof serializeAbstractType>[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,
+  };
+}

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<Logger, 'warn' | 'error'>;
+
+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<Logger, 'warn'>
+) {
+  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<T>(
+  operation: 'merge_updates',
+  engine: CodecEngine,
+  updates: Uint8Array[],
+  caller: string,
+  logger: Pick<Logger, 'warn'> | 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<T>(
+  operation: 'merge_updates',
+  engine: CodecEngine,
+  updates: Uint8Array[],
+  caller: string,
+  logger: Pick<Logger, 'warn'> | undefined,
+  fn: () => Promise<T>
+): Promise<T> {
+  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<Logger, 'warn'>
+) {
+  return measureSyncCodecOperation(
+    'merge_updates',
+    'native',
+    updates,
+    caller,
+    logger,
+    () => nativeMergeUpdatesInApplyWay(updates.map(Buffer.from))
+  );
+}
+
+export function mergeUpdatesWithYjs(
+  updates: Uint8Array[],
+  caller: string,
+  logger?: Pick<Logger, 'warn'>
+) {
+  return measureSyncCodecOperation(
+    'merge_updates',
+    'yjs',
+    updates,
+    caller,
+    logger,
+    () => Buffer.from(yjsMergeUpdatesInCore(updates))
+  );
+}
+
+async function recoverDocWithYjs(
+  updates: Uint8Array[],
+  caller: string,
+  logger?: CodecLogger
+): Promise<Y.Doc> {
+  const doc = new Y.Doc();
+  const chunks = chunk(updates, 10);
+  let i = 0;
+  let failedUpdates = 0;
+  const start = performance.now();
+
+  try {
+    await new Promise<void>(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));
+    }
+  );
+}

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<Y.Doc> {
-    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();
-      });
-    });
-  }
 }

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<Uint8Array> {
-  // use native module to merge updates
-  return mergeUpdatesInApplyWay(updates.map(u => Buffer.from(u)));
+async function nativeApplyUpdates(updates: Uint8Array[]): Promise<Uint8Array> {
+  return applyUpdatesWithNative(updates, 'doc.storage.squash.native');
+}
+
+async function yjsMergeUpdates(updates: Uint8Array[]): Promise<Uint8Array> {
+  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<Uint8Array>
   ): Promise<DocUpdate> {
-    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.');

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 {