fix(server): config & update handle (#15173)

#### PR Dependency Tree


* **PR #15173** 👈

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

* **New Features**
* Added native document update validation to check incoming Yjs updates
for decodability before applying them.
* Introduced support for validation timeouts and cancellation during
update checks.
* Blob maintenance jobs now detect when object storage is unavailable
and skip related work gracefully.

* **Bug Fixes**
* Invalid (and oversized) updates are now filtered out earlier during
document ingestion.
* Background blob maintenance continues processing other work even if
one workspace fails.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/backend/server/src/core/backend-runtime/__tests__/blob-job.spec.ts

@@ -0,0 +1,156 @@
+import ava, { TestFn } from 'ava';
+import Sinon from 'sinon';
+
+import { BackendRuntimeBlobJob } from '../blob-job';
+
+interface Context {
+  runtime: {
+    health: Sinon.SinonStub;
+    backfillMissingBlobMetadata: Sinon.SinonStub;
+    rebuildWorkspaceDocBlobRefs: Sinon.SinonStub;
+    planUnreferencedWorkspaceBlobs: Sinon.SinonStub;
+    executeBlobCleanupCandidates: Sinon.SinonStub;
+  };
+  event: {
+    emitAsync: Sinon.SinonStub;
+  };
+  queue: {
+    add: Sinon.SinonStub;
+  };
+  db: {
+    workspace: {
+      findMany: Sinon.SinonStub;
+    };
+  };
+  job: BackendRuntimeBlobJob;
+}
+
+const test = ava as TestFn<Context>;
+
+test.beforeEach(t => {
+  t.context.runtime = {
+    health: Sinon.stub().resolves({
+      databaseConnected: true,
+      objectStorageConfigured: true,
+    }),
+    backfillMissingBlobMetadata: Sinon.stub(),
+    rebuildWorkspaceDocBlobRefs: Sinon.stub(),
+    planUnreferencedWorkspaceBlobs: Sinon.stub(),
+    executeBlobCleanupCandidates: Sinon.stub(),
+  };
+  t.context.event = {
+    emitAsync: Sinon.stub().resolves(undefined),
+  };
+  t.context.queue = {
+    add: Sinon.stub().resolves(undefined),
+  };
+  t.context.db = {
+    workspace: {
+      findMany: Sinon.stub(),
+    },
+  };
+  t.context.job = new BackendRuntimeBlobJob(
+    t.context.runtime as any,
+    t.context.event as any,
+    t.context.queue as any,
+    t.context.db as any
+  );
+});
+
+const objectStorageRequiredCases: {
+  name: string;
+  run: (context: Context) => Promise<unknown>;
+  untouched: (context: Context) => Sinon.SinonStub[];
+}[] = [
+  {
+    name: 'blob metadata backfill sweep',
+    run: context => context.job.backfillMissingBlobMetadataBySid({}),
+    untouched: context => [
+      context.db.workspace.findMany,
+      context.runtime.backfillMissingBlobMetadata,
+      context.queue.add,
+    ],
+  },
+  {
+    name: 'blob cleanup execution',
+    run: context =>
+      context.job.executeBlobCleanupCandidates({ runId: 'run-1' }),
+    untouched: context => [
+      context.runtime.executeBlobCleanupCandidates,
+      context.event.emitAsync,
+    ],
+  },
+  {
+    name: 'blob cleanup planning sweep',
+    run: context => context.job.planUnreferencedWorkspaceBlobsBySid({}),
+    untouched: context => [
+      context.db.workspace.findMany,
+      context.runtime.planUnreferencedWorkspaceBlobs,
+      context.queue.add,
+    ],
+  },
+  {
+    name: 'blob cleanup planning',
+    run: context =>
+      context.job.planUnreferencedWorkspaceBlobs({
+        workspaceId: 'workspace-1',
+      }),
+    untouched: context => [context.runtime.planUnreferencedWorkspaceBlobs],
+  },
+];
+
+for (const scenario of objectStorageRequiredCases) {
+  test(`${scenario.name} skips when object storage is not configured`, async t => {
+    t.context.runtime.health.resolves({
+      databaseConnected: true,
+      objectStorageConfigured: false,
+    });
+
+    await scenario.run(t.context);
+
+    t.true(t.context.runtime.health.calledOnce);
+    for (const stub of scenario.untouched(t.context)) {
+      t.false(stub.called);
+    }
+  });
+}
+
+test('doc blob refs sweep continues after one workspace fails', async t => {
+  t.context.db.workspace.findMany.resolves([
+    { id: 'workspace-1', sid: 1 },
+    { id: 'workspace-2', sid: 2 },
+  ]);
+  t.context.runtime.rebuildWorkspaceDocBlobRefs
+    .onFirstCall()
+    .rejects(new Error('bad root doc'))
+    .onSecondCall()
+    .resolves({
+      scannedDocs: 1,
+      parsedDocs: 1,
+      refsWritten: 0,
+      refsDeleted: 0,
+      failedDocs: 0,
+      nextCursor: null,
+    });
+
+  await t.context.job.rebuildWorkspaceDocBlobRefsBySid({
+    workspaceLimit: 2,
+    docLimit: 100,
+  });
+
+  t.is(t.context.runtime.rebuildWorkspaceDocBlobRefs.callCount, 2);
+  t.deepEqual(t.context.runtime.rebuildWorkspaceDocBlobRefs.firstCall.args, [
+    'workspace-1',
+    100,
+  ]);
+  t.deepEqual(t.context.runtime.rebuildWorkspaceDocBlobRefs.secondCall.args, [
+    'workspace-2',
+    100,
+  ]);
+  t.true(
+    t.context.queue.add.calledWith(
+      'backendRuntime.rebuildWorkspaceDocBlobRefsBySid',
+      { lastSid: 2, workspaceLimit: 2, docLimit: 100 }
+    )
+  );
+});

packages/backend/server/src/core/backend-runtime/blob-job.ts

@@ -99,6 +99,10 @@ export class BackendRuntimeBlobJob {
     workspaceLimit = 100,
     objectLimit = 1000,
   }: Jobs['backendRuntime.backfillMissingBlobMetadataBySid']) {
+    if (!(await this.hasObjectStorage('blob metadata backfill sweep'))) {
+      return;
+    }
+
     const workspaces = await this.db.workspace.findMany({
       where: { sid: { gt: lastSid } },
       orderBy: { sid: 'asc' },
@@ -107,9 +111,16 @@ export class BackendRuntimeBlobJob {
     });
 
     for (const workspace of workspaces) {
-      await this.drainBlobMetadataBackfill(workspace.id, objectLimit, {
-        sid: workspace.sid,
-      });
+      try {
+        await this.drainBlobMetadataBackfill(workspace.id, objectLimit, {
+          sid: workspace.sid,
+        });
+      } catch (err) {
+        this.logger.error(
+          `blob metadata backfill failed workspace=${workspace.id} sid=${workspace.sid}`,
+          err
+        );
+      }
     }
 
     const nextSid = workspaces.at(-1)?.sid;
@@ -192,6 +203,10 @@ export class BackendRuntimeBlobJob {
     workspaceId,
     limit = 1000,
   }: Jobs['backendRuntime.backfillMissingBlobMetadata']) {
+    if (!(await this.hasObjectStorage('blob metadata backfill'))) {
+      return;
+    }
+
     await this.drainBlobMetadataBackfill(workspaceId, limit);
   }
 
@@ -226,9 +241,16 @@ export class BackendRuntimeBlobJob {
     });
 
     for (const workspace of workspaces) {
-      await this.drainWorkspaceDocBlobRefs(workspace.id, docLimit, {
-        sid: workspace.sid,
-      });
+      try {
+        await this.drainWorkspaceDocBlobRefs(workspace.id, docLimit, {
+          sid: workspace.sid,
+        });
+      } catch (err) {
+        this.logger.error(
+          `doc blob refs rebuild failed workspace=${workspace.id} sid=${workspace.sid}`,
+          err
+        );
+      }
     }
 
     const nextSid = workspaces.at(-1)?.sid;
@@ -247,6 +269,10 @@ export class BackendRuntimeBlobJob {
     gracePeriodDays = 30,
     limit = 1000,
   }: Jobs['backendRuntime.planUnreferencedWorkspaceBlobs']) {
+    if (!(await this.hasObjectStorage('blob cleanup planning'))) {
+      return;
+    }
+
     const result = await this.rt.planUnreferencedWorkspaceBlobs(
       workspaceId,
       gracePeriodDays,
@@ -264,6 +290,10 @@ export class BackendRuntimeBlobJob {
     gracePeriodDays = 30,
     limit = 1000,
   }: Jobs['backendRuntime.planUnreferencedWorkspaceBlobsBySid']) {
+    if (!(await this.hasObjectStorage('blob cleanup planning sweep'))) {
+      return;
+    }
+
     const workspaces = await this.db.workspace.findMany({
       where: {
         sid: {
@@ -281,14 +311,21 @@ export class BackendRuntimeBlobJob {
     });
 
     for (const workspace of workspaces) {
-      const result = await this.rt.planUnreferencedWorkspaceBlobs(
-        workspace.id,
-        gracePeriodDays,
-        limit
-      );
-      this.logger.log(
-        `planned blob cleanup workspace=${workspace.id} sid=${workspace.sid} run=${result.runId} candidates=${result.candidatesMarked} scanned=${result.scannedBlobs}`
-      );
+      try {
+        const result = await this.rt.planUnreferencedWorkspaceBlobs(
+          workspace.id,
+          gracePeriodDays,
+          limit
+        );
+        this.logger.log(
+          `planned blob cleanup workspace=${workspace.id} sid=${workspace.sid} run=${result.runId} candidates=${result.candidatesMarked} scanned=${result.scannedBlobs}`
+        );
+      } catch (err) {
+        this.logger.error(
+          `blob cleanup planning failed workspace=${workspace.id} sid=${workspace.sid}`,
+          err
+        );
+      }
     }
 
     const nextSid = workspaces.at(-1)?.sid;
@@ -308,6 +345,10 @@ export class BackendRuntimeBlobJob {
     gracePeriodDays = 30,
     limit = 1000,
   }: Jobs['backendRuntime.executeBlobCleanupCandidates']) {
+    if (!(await this.hasObjectStorage('blob cleanup execution'))) {
+      return;
+    }
+
     const result = await this.rt.executeBlobCleanupCandidates(
       runId,
       gracePeriodDays,
@@ -365,4 +406,16 @@ export class BackendRuntimeBlobJob {
       }
     }
   }
+
+  private async hasObjectStorage(operation: string) {
+    const health = await this.rt.health();
+    if (health.objectStorageConfigured) {
+      return true;
+    }
+
+    this.logger.warn(
+      `skip ${operation}: BackendRuntime object storage is not configured`
+    );
+    return false;
+  }
 }

packages/backend/server/src/core/doc/adapters/userspace.ts

@@ -55,6 +55,9 @@ export class PgUserspaceDocStorageAdapter extends DocStorageAdapter {
       return 0;
     }
 
+    updates = await this.filterValidDocUpdates(userId, docId, updates);
+    if (!updates.length) return 0;
+
     await using _lock = await this.lockDocForUpdate(userId, docId);
     const snapshot = await this.getDocSnapshot(userId, docId);
     const now = Date.now();

packages/backend/server/src/core/doc/adapters/workspace.ts

@@ -61,6 +61,9 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
       return 0;
     }
 
+    updates = await this.filterValidDocUpdates(workspaceId, docId, updates);
+    if (!updates.length) return 0;
+
     const isNewDoc = !(await this.models.doc.exists(workspaceId, docId));
 
     let pendings = updates;

packages/backend/server/src/core/doc/storage/doc.ts

@@ -11,11 +11,15 @@ import {
   UndoManager,
 } from 'yjs';
 
-import { CallMetric } from '../../../base';
+import { CallMetric, metrics } from '../../../base';
+import { validateDocUpdate } from '../../../native';
 import { applyUpdatesWithNative, mergeUpdatesWithYjs } from '../merge-updates';
 import { Connection } from './connection';
 import { SingletonLocker } from './lock';
 
+const DOC_UPDATE_VALIDATE_TIMEOUT_MS = 1000;
+const DOC_UPDATE_VALIDATE_MAX_BYTES = 32 * 1024 * 1024;
+
 async function nativeApplyUpdates(updates: Uint8Array[]): Promise<Uint8Array> {
   return applyUpdatesWithNative(updates, 'doc.storage.squash.native');
 }
@@ -81,6 +85,47 @@ export abstract class DocStorageAdapter extends Connection {
     );
   }
 
+  protected async filterValidDocUpdates(
+    spaceId: string,
+    docId: string,
+    updates: Uint8Array[]
+  ) {
+    const valid: Uint8Array[] = [];
+    for (const update of updates) {
+      const reason = await this.invalidDocUpdateReason(update);
+      if (reason) {
+        metrics.doc.counter('doc_update_rejected').add(1, { reason });
+        this.logger.warn(
+          `Dropped invalid doc update, spaceId: ${spaceId}, docId: ${docId}, reason: ${reason}, size: ${update.length}`
+        );
+        continue;
+      }
+      valid.push(update);
+    }
+    return valid;
+  }
+
+  private async invalidDocUpdateReason(update: Uint8Array) {
+    if (update.length === 2 && update[0] === 0 && update[1] === 0) {
+      return null;
+    }
+    if (update.length > DOC_UPDATE_VALIDATE_MAX_BYTES) {
+      return 'oversized';
+    }
+
+    try {
+      return (await validateDocUpdate(Buffer.from(update), {
+        timeoutMs: DOC_UPDATE_VALIDATE_TIMEOUT_MS,
+      }))
+        ? null
+        : 'invalid';
+    } catch (err) {
+      this.logger.warn('Doc update validation failed', err);
+      metrics.doc.counter('doc_update_validation_failed').add(1);
+      return null;
+    }
+  }
+
   async getDoc(spaceId: string, docId: string): Promise<DocRecord | null> {
     await using _lock = await this.lockDocForUpdate(spaceId, docId);