feat(server): impl doc gc (#15282)

#### PR Dependency Tree


* **PR #15282** 👈

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 automated document cleanup to reconcile missing workspace docs,
delete related stored data, and recover if the doc returns.
* Added effect-based follow-up reconciliation for search indexing,
Copilot embeddings, and comment attachment cleanup with explicit
acknowledgements.
* **Bug Fixes**
* Deleted-document references now persist as dangling references rather
than disappearing.
* Improved document deletion flow to enforce permissions and ensure
authorized deletions succeed.
* **Tests**
* Expanded coverage for cleanup recovery, indexing/embedding
reconciliation, permissions, and reference semantics.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/backend/native/src/runtime/backend_runtime/doc_storage.rs

@@ -126,37 +126,4 @@ impl BackendRuntime {
 
     Ok(true)
   }
-
-  #[napi]
-  pub async fn delete_doc_storage(&self, workspace_id: String, doc_id: String) -> napi::Result<()> {
-    let pool = self.pool().await?;
-    let mut tx = pool
-      .begin()
-      .await
-      .map_err(|err| RuntimeError::database("DocStorage delete begin transaction failed", err))?;
-
-    sqlx::query("DELETE FROM snapshots WHERE workspace_id = $1 AND guid = $2")
-      .bind(&workspace_id)
-      .bind(&doc_id)
-      .execute(&mut *tx)
-      .await
-      .map_err(|err| RuntimeError::database("DocStorage delete snapshot failed", err))?;
-    sqlx::query("DELETE FROM updates WHERE workspace_id = $1 AND guid = $2")
-      .bind(&workspace_id)
-      .bind(&doc_id)
-      .execute(&mut *tx)
-      .await
-      .map_err(|err| RuntimeError::database("DocStorage delete updates failed", err))?;
-    sqlx::query("DELETE FROM snapshot_histories WHERE workspace_id = $1 AND guid = $2")
-      .bind(&workspace_id)
-      .bind(&doc_id)
-      .execute(&mut *tx)
-      .await
-      .map_err(|err| RuntimeError::database("DocStorage delete histories failed", err))?;
-
-    tx.commit()
-      .await
-      .map_err(|err| RuntimeError::database("DocStorage delete commit failed", err))?;
-    Ok(())
-  }
 }

packages/backend/native/src/runtime/backend_runtime/tests.rs

@@ -18,8 +18,9 @@ fn migrations_include_runtime_tables_without_worker_heartbeats() {
   assert!(RUNTIME_MIGRATIONS.contains("runtime_states"));
   assert!(RUNTIME_MIGRATIONS.contains("runtime_gates"));
   assert!(RUNTIME_MIGRATIONS.contains("runtime_leases"));
-  assert!(RUNTIME_MIGRATIONS.contains("blob_reconciliation_runs"));
-  assert!(RUNTIME_MIGRATIONS.contains("blob_reconciliation_checkpoints"));
+  assert!(RUNTIME_MIGRATIONS.contains("storage_reconciliation_runs"));
+  assert!(RUNTIME_MIGRATIONS.contains("storage_reconciliation_checkpoints"));
+  assert!(RUNTIME_MIGRATIONS.contains("document_cleanup_candidates"));
   assert!(RUNTIME_MIGRATIONS.contains("doc_blob_refs"));
   assert!(RUNTIME_MIGRATIONS.contains("blob_cleanup_candidates"));
   assert!(!RUNTIME_MIGRATIONS.contains("runtime_worker_heartbeats"));
@@ -136,12 +137,10 @@ async fn insert_invite_quota_fixture(
   .bind(email)
   .execute(&pool)
   .await?;
-  sqlx::query(
-    "INSERT INTO workspaces (id, public, created_at) VALUES ($1, false, clock_timestamp() - interval '60 days')",
-  )
-  .bind(&workspace_id)
-  .execute(&pool)
-  .await?;
+  sqlx::query("INSERT INTO workspaces (id, created_at) VALUES ($1, clock_timestamp() - interval '60 days')")
+    .bind(&workspace_id)
+    .execute(&pool)
+    .await?;
   sqlx::query(
     r#"
     INSERT INTO effective_workspace_quota_states (

packages/backend/native/src/runtime/error.rs

@@ -97,6 +97,16 @@ impl RuntimeError {
       _ => false,
     }
   }
+
+  pub(crate) fn is_serialization_failure(&self) -> bool {
+    matches!(
+      self,
+      Self::Database {
+        source: sqlx::Error::Database(source),
+        ..
+      } if source.code().as_deref() == Some("40001")
+    )
+  }
 }
 
 pub(crate) fn to_napi_error(error: RuntimeError) -> Error {

packages/backend/native/src/runtime/migrations.rs

@@ -5,16 +5,10 @@ use super::{RuntimeError, RuntimeResult};
 pub(crate) const RUNTIME_MIGRATIONS: &str = include_str!("sql/runtime_migrations.sql");
 
 pub(crate) async fn migrate_runtime_tables(pool: &PgPool) -> RuntimeResult<()> {
-  for statement in RUNTIME_MIGRATIONS
-    .split(';')
-    .map(str::trim)
-    .filter(|statement| !statement.is_empty())
-  {
-    sqlx::query(statement)
-      .execute(pool)
-      .await
-      .map_err(|err| RuntimeError::database("Runtime migration failed", err))?;
-  }
+  sqlx::raw_sql(RUNTIME_MIGRATIONS)
+    .execute(pool)
+    .await
+    .map_err(|err| RuntimeError::database("Runtime migration failed", err))?;
 
   Ok(())
 }

packages/backend/native/src/runtime/sql/runtime_migrations.sql

@@ -39,7 +39,7 @@ CREATE TABLE IF NOT EXISTS runtime_leases (
 CREATE INDEX IF NOT EXISTS runtime_leases_expires_at_idx
   ON runtime_leases (expires_at);
 
-CREATE TABLE IF NOT EXISTS blob_reconciliation_runs (
+CREATE TABLE IF NOT EXISTS storage_reconciliation_runs (
   id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
   kind TEXT NOT NULL,
   mode TEXT NOT NULL,
@@ -54,10 +54,10 @@ CREATE TABLE IF NOT EXISTS blob_reconciliation_runs (
   metadata JSONB NOT NULL DEFAULT '{}'
 );
 
-CREATE INDEX IF NOT EXISTS blob_reconciliation_runs_workspace_idx
-  ON blob_reconciliation_runs (workspace_id, started_at DESC);
+CREATE INDEX IF NOT EXISTS storage_reconciliation_runs_workspace_idx
+  ON storage_reconciliation_runs (workspace_id, started_at DESC);
 
-CREATE TABLE IF NOT EXISTS blob_reconciliation_checkpoints (
+CREATE TABLE IF NOT EXISTS storage_reconciliation_checkpoints (
   kind TEXT NOT NULL,
   scope TEXT NOT NULL,
   status TEXT NOT NULL,
@@ -70,8 +70,28 @@ CREATE TABLE IF NOT EXISTS blob_reconciliation_checkpoints (
   PRIMARY KEY (kind, scope)
 );
 
-CREATE INDEX IF NOT EXISTS blob_reconciliation_checkpoints_status_idx
-  ON blob_reconciliation_checkpoints (kind, status, updated_at DESC);
+CREATE INDEX IF NOT EXISTS storage_reconciliation_checkpoints_status_idx
+  ON storage_reconciliation_checkpoints (kind, status, updated_at DESC);
+
+CREATE TABLE IF NOT EXISTS document_cleanup_candidates (
+  workspace_id TEXT NOT NULL,
+  doc_id TEXT NOT NULL,
+  status TEXT NOT NULL CHECK (status IN ('marked', 'effects_pending', 'failed')),
+  missing_since TIMESTAMPTZ(3) NOT NULL,
+  last_observed_missing_at TIMESTAMPTZ(3) NOT NULL,
+  last_doc_activity_at TIMESTAMPTZ(3),
+  cleanup_payload JSONB NOT NULL DEFAULT '{}',
+  attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
+  error TEXT,
+  updated_at TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (workspace_id, doc_id)
+);
+
+CREATE INDEX IF NOT EXISTS document_cleanup_candidates_status_missing_idx
+  ON document_cleanup_candidates (status, missing_since);
+
+CREATE INDEX IF NOT EXISTS document_cleanup_candidates_workspace_status_idx
+  ON document_cleanup_candidates (workspace_id, status, updated_at DESC);
 
 CREATE TABLE IF NOT EXISTS doc_blob_refs (
   workspace_id TEXT NOT NULL,

packages/backend/native/src/runtime/storage_runtime/blob_cleanup.rs

@@ -35,7 +35,7 @@ fn push_workspace_once(workspace_ids: &mut Vec<String>, workspace_id: &str) {
 
 async fn checkpoint_completed(pool: &PgPool, kind: &str, scope: &str) -> RuntimeResult<bool> {
   sqlx::query_scalar::<_, bool>(
-    "SELECT EXISTS(SELECT 1 FROM blob_reconciliation_checkpoints WHERE kind = $1 AND scope = $2 AND status = \
+    "SELECT EXISTS(SELECT 1 FROM storage_reconciliation_checkpoints WHERE kind = $1 AND scope = $2 AND status = \
      'completed')",
   )
   .bind(kind)
@@ -46,7 +46,43 @@ async fn checkpoint_completed(pool: &PgPool, kind: &str, scope: &str) -> Runtime
 }
 
 async fn projection_is_stale(pool: &PgPool, workspace_id: &str) -> RuntimeResult<bool> {
-  let checkpoint_fresh = checkpoint_completed(pool, "doc_blob_refs", workspace_id).await?;
+  let checkpoint_completed_at = sqlx::query_scalar::<_, Option<chrono::DateTime<chrono::Utc>>>(
+    r#"
+    SELECT MIN(completed_at)
+    FROM storage_reconciliation_checkpoints
+    WHERE scope = $1
+      AND kind IN ('document_cleanup', 'doc_blob_refs')
+      AND status = 'completed'
+    HAVING COUNT(*) = 2
+    "#,
+  )
+  .bind(workspace_id)
+  .fetch_optional(pool)
+  .await
+  .map_err(|err| RuntimeError::database("Blob cleanup retention checkpoint load failed", err))?
+  .flatten();
+  let Some(checkpoint_completed_at) = checkpoint_completed_at else {
+    return Ok(true);
+  };
+  let activity_after_checkpoint = sqlx::query_scalar::<_, bool>(
+    r#"
+    SELECT EXISTS(
+      SELECT 1 FROM snapshots
+      WHERE workspace_id = $1 AND updated_at > $2
+      UNION ALL
+      SELECT 1 FROM updates
+      WHERE workspace_id = $1 AND created_at > $2
+      UNION ALL
+      SELECT 1 FROM snapshot_histories
+      WHERE workspace_id = $1 AND timestamp > $2
+    )
+    "#,
+  )
+  .bind(workspace_id)
+  .bind(checkpoint_completed_at)
+  .fetch_one(pool)
+  .await
+  .map_err(|err| RuntimeError::database("Blob cleanup retention activity check failed", err))?;
   let has_stale_rows = sqlx::query_scalar::<_, bool>(
     "SELECT EXISTS(SELECT 1 FROM doc_blob_refs WHERE workspace_id = $1 AND status <> 'fresh')",
   )
@@ -54,7 +90,7 @@ async fn projection_is_stale(pool: &PgPool, workspace_id: &str) -> RuntimeResult
   .fetch_one(pool)
   .await
   .map_err(|err| RuntimeError::database("Blob cleanup projection freshness check failed", err))?;
-  Ok(!checkpoint_fresh || has_stale_rows)
+  Ok(activity_after_checkpoint || has_stale_rows)
 }
 
 async fn stale_projection_workspaces(pool: &PgPool, workspace_id: &str) -> RuntimeResult<Vec<String>> {
@@ -170,7 +206,7 @@ async fn load_completed_blobs(
 
 async fn load_plan_cursor(pool: &PgPool, workspace_id: &str) -> RuntimeResult<Option<String>> {
   let row = sqlx::query_as::<_, (String, serde_json::Value)>(
-    "SELECT status, cursor FROM blob_reconciliation_checkpoints WHERE kind = 'blob_cleanup_plan' AND scope = $1",
+    "SELECT status, cursor FROM storage_reconciliation_checkpoints WHERE kind = 'blob_cleanup_plan' AND scope = $1",
   )
   .bind(workspace_id)
   .fetch_optional(pool)
@@ -199,13 +235,13 @@ async fn upsert_plan_checkpoint(
   let status = if completed { "completed" } else { "running" };
   sqlx::query(
     r#"
-    INSERT INTO blob_reconciliation_checkpoints
+    INSERT INTO storage_reconciliation_checkpoints
       (kind, scope, status, cursor, last_key, completed_at)
     VALUES ('blob_cleanup_plan', $1, $2, $3, $4, CASE WHEN $5 THEN CURRENT_TIMESTAMP ELSE NULL END)
     ON CONFLICT (kind, scope) DO UPDATE
       SET status = EXCLUDED.status,
           cursor = EXCLUDED.cursor,
-          last_key = COALESCE(EXCLUDED.last_key, blob_reconciliation_checkpoints.last_key),
+          last_key = COALESCE(EXCLUDED.last_key, storage_reconciliation_checkpoints.last_key),
           completed_at = CASE WHEN $5 THEN CURRENT_TIMESTAMP ELSE NULL END,
           updated_at = CURRENT_TIMESTAMP
     "#,
@@ -224,7 +260,7 @@ async fn upsert_plan_checkpoint(
 async fn create_run(pool: &PgPool, workspace_id: &str) -> RuntimeResult<String> {
   sqlx::query_scalar::<_, String>(
     r#"
-    INSERT INTO blob_reconciliation_runs (kind, mode, status, workspace_id)
+    INSERT INTO storage_reconciliation_runs (kind, mode, status, workspace_id)
     VALUES ('blob_cleanup_plan', 'mark_only', 'running', $1)
     RETURNING id::text
     "#,
@@ -252,7 +288,7 @@ async fn finish_run(
   .unwrap_or(0);
   sqlx::query(
     r#"
-    UPDATE blob_reconciliation_runs
+    UPDATE storage_reconciliation_runs
     SET status = 'finished',
         finished_at = CURRENT_TIMESTAMP,
         scanned = $2,
@@ -318,7 +354,7 @@ async fn finish_execute_run(
 ) -> RuntimeResult<()> {
   sqlx::query(
     r#"
-    UPDATE blob_reconciliation_runs
+    UPDATE storage_reconciliation_runs
     SET status = 'finished',
         finished_at = CURRENT_TIMESTAMP,
         scanned = $2,

packages/backend/native/src/runtime/storage_runtime/blob_reconciliation.rs

@@ -85,7 +85,8 @@ impl BackfillCheckpoint {
 
 async fn load_checkpoint(pool: &PgPool, scope: &str) -> RuntimeResult<Option<BackfillCheckpoint>> {
   sqlx::query_as::<_, BackfillCheckpoint>(
-    "SELECT last_key, cursor FROM blob_reconciliation_checkpoints WHERE kind = 'blob_metadata_backfill' AND scope = $1",
+    "SELECT last_key, cursor FROM storage_reconciliation_checkpoints WHERE kind = 'blob_metadata_backfill' AND scope \
+     = $1",
   )
   .bind(scope)
   .fetch_optional(pool)
@@ -103,13 +104,13 @@ async fn upsert_checkpoint(
   let status = if completed { "completed" } else { "running" };
   sqlx::query(
     r#"
-    INSERT INTO blob_reconciliation_checkpoints
+    INSERT INTO storage_reconciliation_checkpoints
       (kind, scope, status, cursor, last_key, completed_at, metadata)
     VALUES ('blob_metadata_backfill', $1, $2, $3, $4, CASE WHEN $5 THEN CURRENT_TIMESTAMP ELSE NULL END, $6)
     ON CONFLICT (kind, scope) DO UPDATE
       SET status = EXCLUDED.status,
           cursor = EXCLUDED.cursor,
-          last_key = COALESCE(EXCLUDED.last_key, blob_reconciliation_checkpoints.last_key),
+          last_key = COALESCE(EXCLUDED.last_key, storage_reconciliation_checkpoints.last_key),
           completed_at = CASE WHEN $5 THEN CURRENT_TIMESTAMP ELSE NULL END,
           updated_at = CURRENT_TIMESTAMP,
           metadata = EXCLUDED.metadata
@@ -247,7 +248,7 @@ impl StorageRuntime {
 
     sqlx::query(
       r#"
-      INSERT INTO blob_reconciliation_runs
+      INSERT INTO storage_reconciliation_runs
         (kind, mode, status, workspace_id, finished_at, scanned, changed, failed, metadata)
       VALUES ('blob_metadata_backfill', 'execute', 'finished', $1, CURRENT_TIMESTAMP, $2, $3, $4, $5)
       "#,

packages/backend/native/src/runtime/storage_runtime/doc_blob_refs.rs

@@ -1,111 +1,32 @@
 use affine_doc_loader as doc_loader;
-use chrono::{DateTime, Utc};
-use sqlx::{FromRow, PgPool};
-use y_octo::Doc;
+use sqlx::PgPool;
 
-use super::{RuntimeDocBlobRefsResult, RuntimeError, RuntimeResult, StorageRuntime, napi_error};
+use super::{
+  CurrentDoc, RuntimeDocBlobRefsResult, RuntimeError, RuntimeResult, StorageRuntime, load_current_doc,
+  load_workspace_live_doc_ids, napi_error,
+};
 
 const PARSER_VERSION: i32 = 1;
 
-#[derive(FromRow)]
-struct SnapshotRow {
-  workspace_id: String,
-  doc_id: String,
-  blob: Vec<u8>,
-  updated_at: DateTime<Utc>,
-}
-
-#[derive(FromRow)]
-struct UpdateRow {
-  blob: Vec<u8>,
-  created_at: DateTime<Utc>,
-}
-
 struct ExtractedRef {
   blob_key: String,
   block_id: String,
   flavour: String,
 }
 
-async fn load_snapshot(pool: &PgPool, workspace_id: &str, doc_id: &str) -> RuntimeResult<Option<SnapshotRow>> {
-  sqlx::query_as::<_, SnapshotRow>(
-    r#"
-    SELECT workspace_id, guid AS doc_id, blob, updated_at
-    FROM snapshots
-    WHERE workspace_id = $1 AND guid = $2
-    "#,
+async fn load_workspace_doc_ids(pool: &PgPool, workspace_id: &str) -> RuntimeResult<Vec<String>> {
+  let mut ids = load_workspace_live_doc_ids(pool, workspace_id).await?;
+  let retained = sqlx::query_scalar::<_, String>(
+    "SELECT doc_id FROM document_cleanup_candidates WHERE workspace_id = $1 AND status IN ('marked', 'failed') ORDER \
+     BY doc_id",
   )
   .bind(workspace_id)
-  .bind(doc_id)
-  .fetch_optional(pool)
-  .await
-  .map_err(|err| RuntimeError::database("Doc blob refs load snapshot failed", err))
-}
-
-async fn load_updates(pool: &PgPool, workspace_id: &str, doc_id: &str) -> RuntimeResult<Vec<UpdateRow>> {
-  sqlx::query_as::<_, UpdateRow>(
-    r#"
-    SELECT blob, created_at
-    FROM updates
-    WHERE workspace_id = $1 AND guid = $2
-    ORDER BY created_at ASC
-    "#,
-  )
-  .bind(workspace_id)
-  .bind(doc_id)
   .fetch_all(pool)
   .await
-  .map_err(|err| RuntimeError::database("Doc blob refs load updates failed", err))
-}
-
-fn apply_doc_updates(updates: impl IntoIterator<Item = Vec<u8>>) -> RuntimeResult<Vec<u8>> {
-  let mut doc = Doc::default();
-  for update in updates {
-    doc
-      .apply_update_from_binary_v1(&update)
-      .map_err(|err| RuntimeError::invalid_state(format!("Doc blob refs merge failed: {err}")))?;
-  }
-  doc
-    .encode_update_v1()
-    .map_err(|err| RuntimeError::invalid_state(format!("Doc blob refs encode failed: {err}")))
-}
-
-async fn load_current_doc(pool: &PgPool, workspace_id: &str, doc_id: &str) -> RuntimeResult<Option<SnapshotRow>> {
-  let snapshot = load_snapshot(pool, workspace_id, doc_id).await?;
-  let updates = load_updates(pool, workspace_id, doc_id).await?;
-  if snapshot.is_none() && updates.is_empty() {
-    return Ok(None);
-  }
-
-  let mut merge_inputs = Vec::with_capacity(updates.len() + usize::from(snapshot.is_some()));
-  let mut updated_at = snapshot
-    .as_ref()
-    .map(|snapshot| snapshot.updated_at)
-    .unwrap_or_else(Utc::now);
-  if let Some(snapshot) = snapshot {
-    merge_inputs.push(snapshot.blob);
-  }
-  for update in updates {
-    updated_at = update.created_at;
-    merge_inputs.push(update.blob);
-  }
-
-  Ok(Some(SnapshotRow {
-    workspace_id: workspace_id.to_string(),
-    doc_id: doc_id.to_string(),
-    blob: apply_doc_updates(merge_inputs)?,
-    updated_at,
-  }))
-}
-
-async fn load_workspace_doc_ids(pool: &PgPool, workspace_id: &str) -> RuntimeResult<Vec<String>> {
-  let Some(root) = load_current_doc(pool, workspace_id, workspace_id).await? else {
-    return Ok(Vec::new());
-  };
-  let ids = doc_loader::get_doc_ids_from_binary(root.blob, true)
-    .map_err(|err| RuntimeError::invalid_state(format!("Doc blob refs root doc parse failed: {err}")))?;
-  let mut ids = ids;
+  .map_err(|err| RuntimeError::database("Doc blob refs candidate load failed", err))?;
+  ids.extend(retained);
   ids.sort();
+  ids.dedup();
   Ok(ids)
 }
 
@@ -124,7 +45,7 @@ async fn upsert_projection_checkpoint(
   };
   sqlx::query(
     r#"
-    INSERT INTO blob_reconciliation_checkpoints
+    INSERT INTO storage_reconciliation_checkpoints
       (kind, scope, status, cursor, completed_at, metadata)
     VALUES ('doc_blob_refs', $1, $2, $3, CASE WHEN $4 THEN CURRENT_TIMESTAMP ELSE NULL END, $5)
     ON CONFLICT (kind, scope) DO UPDATE
@@ -151,7 +72,7 @@ async fn upsert_projection_checkpoint(
 async fn upsert_projection_failure_checkpoint(pool: &PgPool, workspace_id: &str, error: &str) -> RuntimeResult<()> {
   sqlx::query(
     r#"
-    INSERT INTO blob_reconciliation_checkpoints
+    INSERT INTO storage_reconciliation_checkpoints
       (kind, scope, status, cursor, completed_at, metadata)
     VALUES ('doc_blob_refs', $1, 'failed', '{}', NULL, $2)
     ON CONFLICT (kind, scope) DO UPDATE
@@ -174,14 +95,17 @@ async fn upsert_projection_failure_checkpoint(pool: &PgPool, workspace_id: &str,
 }
 
 async fn load_projection_cursor(pool: &PgPool, workspace_id: &str) -> RuntimeResult<Option<String>> {
-  let cursor = sqlx::query_scalar::<_, serde_json::Value>(
-    "SELECT cursor FROM blob_reconciliation_checkpoints WHERE kind = 'doc_blob_refs' AND scope = $1",
+  let checkpoint = sqlx::query_as::<_, (String, serde_json::Value)>(
+    "SELECT status, cursor FROM storage_reconciliation_checkpoints WHERE kind = 'doc_blob_refs' AND scope = $1",
   )
   .bind(workspace_id)
   .fetch_optional(pool)
   .await
   .map_err(|err| RuntimeError::database("Doc blob refs checkpoint load failed", err))?;
-  Ok(cursor.and_then(|cursor| {
+  Ok(checkpoint.and_then(|(status, cursor)| {
+    if status != "running" {
+      return None;
+    }
     cursor
       .get("lastDocId")
       .and_then(|value| value.as_str())
@@ -205,7 +129,7 @@ async fn purge_removed_doc_refs(pool: &PgPool, workspace_id: &str, current_doc_i
   Ok(result.rows_affected() as i64)
 }
 
-fn extract_refs(snapshot: &SnapshotRow) -> RuntimeResult<Vec<ExtractedRef>> {
+fn extract_refs(snapshot: &CurrentDoc) -> RuntimeResult<Vec<ExtractedRef>> {
   let parsed = doc_loader::parse_doc_from_binary(snapshot.blob.clone(), snapshot.doc_id.clone())
     .map_err(|err| RuntimeError::invalid_state(format!("Doc blob refs parse failed: {err}")))?;
   let mut refs = Vec::new();
@@ -226,6 +150,9 @@ fn extract_refs(snapshot: &SnapshotRow) -> RuntimeResult<Vec<ExtractedRef>> {
 
 #[cfg(test)]
 mod tests {
+  use chrono::Utc;
+  use y_octo::Doc;
+
   use super::*;
 
   #[test]
@@ -233,7 +160,7 @@ mod tests {
     let doc_id = "doc-blob-ref-test".to_string();
     let blob =
       doc_loader::build_full_doc("Doc", "![Alt](blob://image-blob-key)", &doc_id).expect("doc fixture should build");
-    let snapshot = SnapshotRow {
+    let snapshot = CurrentDoc {
       workspace_id: "workspace".to_string(),
       doc_id,
       blob,
@@ -272,9 +199,21 @@ mod tests {
     let ids = doc_loader::get_doc_ids_from_binary(root, true).expect("root doc ids should parse");
     assert_eq!(ids, vec!["active-doc", "trashed-doc"]);
   }
+
+  #[test]
+  fn doc_blob_refs_rejects_corrupt_docs() {
+    let snapshot = CurrentDoc {
+      workspace_id: "workspace".to_string(),
+      doc_id: "corrupt".to_string(),
+      blob: vec![0xff],
+      updated_at: Utc::now(),
+    };
+
+    assert!(extract_refs(&snapshot).is_err());
+  }
 }
 
-async fn replace_doc_refs(pool: &PgPool, snapshot: &SnapshotRow, refs: Vec<ExtractedRef>) -> RuntimeResult<(i64, i64)> {
+async fn replace_doc_refs(pool: &PgPool, snapshot: &CurrentDoc, refs: Vec<ExtractedRef>) -> RuntimeResult<(i64, i64)> {
   let mut tx = pool
     .begin()
     .await

packages/backend/native/src/runtime/storage_runtime/mod.rs

@@ -7,18 +7,21 @@ use std::{
 };
 
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
+use chrono::{DateTime, Utc};
 use napi::bindgen_prelude::Buffer;
 use serde::Deserialize;
 use serde_json::{Map, Value};
 use sha2::{Digest, Sha256};
-use sqlx::{PgPool, Row, postgres::PgPoolOptions};
+use sqlx::{FromRow, PgPool, Row, postgres::PgPoolOptions};
 use tokio::{sync::Mutex, task::JoinSet};
+use y_octo::Doc;
 
 mod assetpack;
 mod blob_cleanup;
 mod blob_reclaimer;
 mod blob_reconciliation;
 mod doc_blob_refs;
+mod document_cleanup;
 pub(crate) mod object_storage;
 
 use self::object_storage::{
@@ -33,9 +36,10 @@ pub(super) use super::{
   napi_error, to_napi_error,
   types::{
     RuntimeBlobCleanupExecuteResult, RuntimeBlobCleanupPlanResult, RuntimeBlobCleanupResult, RuntimeBlobCompleteResult,
-    RuntimeBlobMetadataBackfillResult, RuntimeDocBlobRefsResult, RuntimeMultipartUploadInit,
-    RuntimeMultipartUploadPart, RuntimeObjectGetResult, RuntimeObjectListEntry, RuntimeObjectMetadata,
-    RuntimeObjectStoragePutOptions, RuntimePresignedObjectRequest,
+    RuntimeBlobMetadataBackfillResult, RuntimeDocBlobRefsResult, RuntimeDocumentCleanupAckResult,
+    RuntimeDocumentCleanupEffect, RuntimeDocumentCleanupExecuteResult, RuntimeDocumentCleanupReconcileResult,
+    RuntimeMultipartUploadInit, RuntimeMultipartUploadPart, RuntimeObjectGetResult, RuntimeObjectListEntry,
+    RuntimeObjectMetadata, RuntimeObjectStoragePutOptions, RuntimePresignedObjectRequest,
   },
 };
 
@@ -45,6 +49,100 @@ const OBJECT_DELETE_MANY_CONCURRENCY: usize = 3;
 
 type Result<T> = RuntimeResult<T>;
 
+#[derive(FromRow)]
+struct CurrentDoc {
+  workspace_id: String,
+  doc_id: String,
+  blob: Vec<u8>,
+  updated_at: DateTime<Utc>,
+}
+
+#[derive(FromRow)]
+struct CurrentDocUpdate {
+  blob: Vec<u8>,
+  created_at: DateTime<Utc>,
+}
+
+async fn load_current_doc(pool: &PgPool, workspace_id: &str, doc_id: &str) -> RuntimeResult<Option<CurrentDoc>> {
+  let snapshot = sqlx::query_as::<_, CurrentDoc>(
+    r#"
+    SELECT workspace_id, guid AS doc_id, blob, updated_at
+    FROM snapshots
+    WHERE workspace_id = $1 AND guid = $2
+    "#,
+  )
+  .bind(workspace_id)
+  .bind(doc_id)
+  .fetch_optional(pool)
+  .await
+  .map_err(|err| RuntimeError::database("Current doc snapshot load failed", err))?;
+  let updates = sqlx::query_as::<_, CurrentDocUpdate>(
+    r#"
+    SELECT blob, created_at
+    FROM updates
+    WHERE workspace_id = $1 AND guid = $2
+    ORDER BY created_at ASC
+    "#,
+  )
+  .bind(workspace_id)
+  .bind(doc_id)
+  .fetch_all(pool)
+  .await
+  .map_err(|err| RuntimeError::database("Current doc updates load failed", err))?;
+  merge_current_doc(workspace_id, doc_id, snapshot, updates)
+}
+
+fn merge_current_doc(
+  workspace_id: &str,
+  doc_id: &str,
+  snapshot: Option<CurrentDoc>,
+  updates: Vec<CurrentDocUpdate>,
+) -> RuntimeResult<Option<CurrentDoc>> {
+  if snapshot.is_none() && updates.is_empty() {
+    return Ok(None);
+  }
+  let mut doc = Doc::default();
+  let mut updated_at = snapshot
+    .as_ref()
+    .map(|snapshot| snapshot.updated_at)
+    .or_else(|| updates.first().map(|update| update.created_at))
+    .unwrap_or_else(Utc::now);
+  if let Some(snapshot) = &snapshot {
+    doc
+      .apply_update_from_binary_v1(&snapshot.blob)
+      .map_err(|err| RuntimeError::invalid_state(format!("Current doc snapshot merge failed: {err}")))?;
+  }
+  for update in updates {
+    updated_at = updated_at.max(update.created_at);
+    doc
+      .apply_update_from_binary_v1(&update.blob)
+      .map_err(|err| RuntimeError::invalid_state(format!("Current doc update merge failed: {err}")))?;
+  }
+  let blob = doc
+    .encode_update_v1()
+    .map_err(|err| RuntimeError::invalid_state(format!("Current doc encode failed: {err}")))?;
+
+  Ok(Some(CurrentDoc {
+    workspace_id: workspace_id.to_string(),
+    doc_id: doc_id.to_string(),
+    blob,
+    updated_at,
+  }))
+}
+
+async fn load_workspace_live_doc_ids(pool: &PgPool, workspace_id: &str) -> RuntimeResult<Vec<String>> {
+  workspace_live_doc_ids(load_current_doc(pool, workspace_id, workspace_id).await?)
+}
+
+fn workspace_live_doc_ids(root: Option<CurrentDoc>) -> RuntimeResult<Vec<String>> {
+  let root = root.ok_or_else(|| RuntimeError::invalid_state("Workspace root doc is missing"))?;
+  let mut ids = affine_doc_loader::get_doc_ids_from_binary(root.blob, true)
+    .map_err(|err| RuntimeError::invalid_state(format!("Workspace root doc parse failed: {err}")))?;
+  ids.sort();
+  ids.dedup();
+  Ok(ids)
+}
+
 #[napi_derive::napi(object)]
 pub struct StorageRuntimeHealth {
   pub started: bool,
@@ -1385,6 +1483,81 @@ fn system_time_ms(time: SystemTime) -> Result<i64> {
 mod tests {
   use super::*;
 
+  #[test]
+  fn workspace_live_set_merges_pending_updates_and_includes_trash() {
+    use y_octo::{Any, Value};
+
+    let snapshot = affine_doc_loader::add_doc_to_root_doc(Vec::new(), "live", None).unwrap();
+    let pending = affine_doc_loader::add_doc_to_root_doc(snapshot.clone(), "trash", None).unwrap();
+    let merged = merge_current_doc(
+      "workspace",
+      "workspace",
+      Some(CurrentDoc {
+        workspace_id: "workspace".to_string(),
+        doc_id: "workspace".to_string(),
+        blob: snapshot,
+        updated_at: Utc::now(),
+      }),
+      vec![CurrentDocUpdate {
+        blob: pending,
+        created_at: Utc::now(),
+      }],
+    )
+    .unwrap()
+    .unwrap();
+    let mut root = Doc::default();
+    root.apply_update_from_binary_v1(&merged.blob).unwrap();
+    let meta = root.get_map("meta").unwrap();
+    let mut pages = meta.get("pages").and_then(|value| value.to_array()).unwrap();
+    let mut trash = pages
+      .iter()
+      .find_map(|value| {
+        let page = value.to_map()?;
+        (page.get("id")?.to_any()? == Any::String("trash".to_string())).then_some(page)
+      })
+      .unwrap();
+    trash.insert("trash".to_string(), Value::Any(Any::True)).unwrap();
+
+    let ids = workspace_live_doc_ids(Some(CurrentDoc {
+      workspace_id: "workspace".to_string(),
+      doc_id: "workspace".to_string(),
+      blob: root.encode_update_v1().unwrap(),
+      updated_at: Utc::now(),
+    }))
+    .unwrap();
+    assert_eq!(ids, ["live", "trash"]);
+
+    let trash_index = pages
+      .iter()
+      .position(|value| {
+        value.to_map().and_then(|page| page.get("id")) == Some(Value::Any(Any::String("trash".to_string())))
+      })
+      .unwrap();
+    pages.remove(trash_index as u64, 1).unwrap();
+    let ids = workspace_live_doc_ids(Some(CurrentDoc {
+      workspace_id: "workspace".to_string(),
+      doc_id: "workspace".to_string(),
+      blob: root.encode_update_v1().unwrap(),
+      updated_at: Utc::now(),
+    }))
+    .unwrap();
+    assert_eq!(ids, ["live"]);
+  }
+
+  #[test]
+  fn workspace_live_set_fails_closed_for_missing_or_corrupt_root() {
+    assert!(workspace_live_doc_ids(None).is_err());
+    assert!(
+      workspace_live_doc_ids(Some(CurrentDoc {
+        workspace_id: "workspace".to_string(),
+        doc_id: "workspace".to_string(),
+        blob: vec![0xff],
+        updated_at: Utc::now(),
+      }))
+      .is_err()
+    );
+  }
+
   #[test]
   fn fs_key_normalization_rejects_traversal() {
     for (key, valid) in [

packages/backend/native/src/runtime/types.rs

@@ -243,6 +243,41 @@ pub struct RuntimeDocBlobRefsResult {
   pub next_cursor: Option<String>,
 }
 
+#[napi_derive::napi(object)]
+pub struct RuntimeDocumentCleanupReconcileResult {
+  pub scanned_docs: i64,
+  pub marked: i64,
+  pub reset: i64,
+  pub recovered: i64,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeDocumentCleanupEffect {
+  pub workspace_id: String,
+  pub doc_id: String,
+  pub cleanup_version: String,
+  pub comment_objects_done: bool,
+  pub search_done: bool,
+  pub copilot_done: bool,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeDocumentCleanupExecuteResult {
+  pub scanned_candidates: i64,
+  pub serialization_retries: i64,
+  pub executed: i64,
+  pub recovered: i64,
+  pub reset: i64,
+  pub failed: i64,
+  pub deleted_rows: i64,
+  pub effects: Vec<RuntimeDocumentCleanupEffect>,
+}
+
+#[napi_derive::napi(object)]
+pub struct RuntimeDocumentCleanupAckResult {
+  pub completed: bool,
+}
+
 #[napi_derive::napi(object)]
 pub struct RuntimeBlobCleanupPlanResult {
   pub run_id: Option<String>,