Skip to content

Commit bb55d6f

Browse files
darkskygitdarkskygit
authored
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 -->
1 parent 81df475 commit bb55d6f

30 files changed

Lines changed: 2656 additions & 336 deletions

File tree

‎blocksuite/integration-test/src/__tests__/main/editor-semantics.spec.ts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { LinkExtension } from '@blocksuite/affine-inline-link';
22
import { textKeymap } from '@blocksuite/affine-inline-preset';
3+
import type { AffineReference } from '@blocksuite/affine-inline-reference';
34
import type {
45
ListBlockModel,
56
ParagraphBlockModel,
@@ -312,6 +313,16 @@ describe('hotkey/bracket/linked-page', () => {
312313
const richText = getRichTextByBlockId(paragraphId);
313314
expect(richText.querySelectorAll('affine-reference').length).toBe(2);
314315
expect(richText.inlineEditor.yTextString.length).toBe(2);
316+
317+
collection.removeDoc(linkedDoc.id);
318+
await wait();
319+
expect(collection.docs.has(linkedDoc.id)).toBe(false);
320+
const danglingReferences =
321+
richText.querySelectorAll<AffineReference>('affine-reference');
322+
expect(danglingReferences.length).toBe(2);
323+
expect([...danglingReferences].every(reference => !reference.refMeta)).toBe(
324+
true
325+
);
315326
});
316327
});
317328

‎packages/backend/native/index.d.ts‎

Lines changed: 34 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ export declare class BackendRuntime {
2222
compactPendingDocUpdates(workspaceId: string, docId: string, batchLimit: number, historyMinIntervalMs: number, historyMaxAgeSeconds: number, owner: string, leaseTtlMs: number): Promise<RuntimeDocCompactionResult>
2323
upsertDocSnapshot(workspaceId: string, docId: string, blob: Buffer, timestampMs: number, editorId?: string | undefined | null): Promise<boolean>
2424
createDocHistory(input: RuntimeDocHistoryInput): Promise<boolean>
25-
deleteDocStorage(workspaceId: string, docId: string): Promise<void>
2625
putRuntimeGateIfAbsent(key: string, ttlMs: number): Promise<boolean>
2726
cleanupExpiredRuntimeGates(limit: number): Promise<number>
2827
cleanupExpiredUserSessions(limit: number): Promise<number>
@@ -78,6 +77,9 @@ export declare class StorageRuntime {
7877
backfillMissingBlobMetadata(workspaceId: string | undefined | null, limit: number): Promise<RuntimeBlobMetadataBackfillResult>
7978
rebuildDocBlobRefs(workspaceId: string, docId: string): Promise<RuntimeDocBlobRefsResult>
8079
rebuildWorkspaceDocBlobRefs(workspaceId: string, limit: number): Promise<RuntimeDocBlobRefsResult>
80+
reconcileWorkspaceDocuments(workspaceId: string): Promise<RuntimeDocumentCleanupReconcileResult>
81+
executeDocumentCleanupCandidates(workspaceId: string | undefined | null, gracePeriodDays: number, limit: number): Promise<RuntimeDocumentCleanupExecuteResult>
82+
ackDocumentCleanupEffect(workspaceId: string, docId: string, cleanupVersion: string, effect: string): Promise<RuntimeDocumentCleanupAckResult>
8183
constructor()
8284
start(): Promise<void>
8385
configure(configJson: string): void
@@ -976,6 +978,37 @@ export interface RuntimeDocHistoryInput {
976978
historyMaxAgeMs: number
977979
}
978980

981+
export interface RuntimeDocumentCleanupAckResult {
982+
completed: boolean
983+
}
984+
985+
export interface RuntimeDocumentCleanupEffect {
986+
workspaceId: string
987+
docId: string
988+
cleanupVersion: string
989+
commentObjectsDone: boolean
990+
searchDone: boolean
991+
copilotDone: boolean
992+
}
993+
994+
export interface RuntimeDocumentCleanupExecuteResult {
995+
scannedCandidates: number
996+
serializationRetries: number
997+
executed: number
998+
recovered: number
999+
reset: number
1000+
failed: number
1001+
deletedRows: number
1002+
effects: Array<RuntimeDocumentCleanupEffect>
1003+
}
1004+
1005+
export interface RuntimeDocumentCleanupReconcileResult {
1006+
scannedDocs: number
1007+
marked: number
1008+
reset: number
1009+
recovered: number
1010+
}
1011+
9791012
export interface RuntimeInviteAbuseActionRequired {
9801013
action: string
9811014
subjectKey: string

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

Lines changed: 0 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -126,37 +126,4 @@ impl BackendRuntime {
126126

127127
Ok(true)
128128
}
129-
130-
#[napi]
131-
pub async fn delete_doc_storage(&self, workspace_id: String, doc_id: String) -> napi::Result<()> {
132-
let pool = self.pool().await?;
133-
let mut tx = pool
134-
.begin()
135-
.await
136-
.map_err(|err| RuntimeError::database("DocStorage delete begin transaction failed", err))?;
137-
138-
sqlx::query("DELETE FROM snapshots WHERE workspace_id = $1 AND guid = $2")
139-
.bind(&workspace_id)
140-
.bind(&doc_id)
141-
.execute(&mut *tx)
142-
.await
143-
.map_err(|err| RuntimeError::database("DocStorage delete snapshot failed", err))?;
144-
sqlx::query("DELETE FROM updates WHERE workspace_id = $1 AND guid = $2")
145-
.bind(&workspace_id)
146-
.bind(&doc_id)
147-
.execute(&mut *tx)
148-
.await
149-
.map_err(|err| RuntimeError::database("DocStorage delete updates failed", err))?;
150-
sqlx::query("DELETE FROM snapshot_histories WHERE workspace_id = $1 AND guid = $2")
151-
.bind(&workspace_id)
152-
.bind(&doc_id)
153-
.execute(&mut *tx)
154-
.await
155-
.map_err(|err| RuntimeError::database("DocStorage delete histories failed", err))?;
156-
157-
tx.commit()
158-
.await
159-
.map_err(|err| RuntimeError::database("DocStorage delete commit failed", err))?;
160-
Ok(())
161-
}
162129
}

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

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,9 @@ fn migrations_include_runtime_tables_without_worker_heartbeats() {
1818
assert!(RUNTIME_MIGRATIONS.contains("runtime_states"));
1919
assert!(RUNTIME_MIGRATIONS.contains("runtime_gates"));
2020
assert!(RUNTIME_MIGRATIONS.contains("runtime_leases"));
21-
assert!(RUNTIME_MIGRATIONS.contains("blob_reconciliation_runs"));
22-
assert!(RUNTIME_MIGRATIONS.contains("blob_reconciliation_checkpoints"));
21+
assert!(RUNTIME_MIGRATIONS.contains("storage_reconciliation_runs"));
22+
assert!(RUNTIME_MIGRATIONS.contains("storage_reconciliation_checkpoints"));
23+
assert!(RUNTIME_MIGRATIONS.contains("document_cleanup_candidates"));
2324
assert!(RUNTIME_MIGRATIONS.contains("doc_blob_refs"));
2425
assert!(RUNTIME_MIGRATIONS.contains("blob_cleanup_candidates"));
2526
assert!(!RUNTIME_MIGRATIONS.contains("runtime_worker_heartbeats"));
@@ -136,12 +137,10 @@ async fn insert_invite_quota_fixture(
136137
.bind(email)
137138
.execute(&pool)
138139
.await?;
139-
sqlx::query(
140-
"INSERT INTO workspaces (id, public, created_at) VALUES ($1, false, clock_timestamp() - interval '60 days')",
141-
)
142-
.bind(&workspace_id)
143-
.execute(&pool)
144-
.await?;
140+
sqlx::query("INSERT INTO workspaces (id, created_at) VALUES ($1, clock_timestamp() - interval '60 days')")
141+
.bind(&workspace_id)
142+
.execute(&pool)
143+
.await?;
145144
sqlx::query(
146145
r#"
147146
INSERT INTO effective_workspace_quota_states (

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,16 @@ impl RuntimeError {
9797
_ => false,
9898
}
9999
}
100+
101+
pub(crate) fn is_serialization_failure(&self) -> bool {
102+
matches!(
103+
self,
104+
Self::Database {
105+
source: sqlx::Error::Database(source),
106+
..
107+
} if source.code().as_deref() == Some("40001")
108+
)
109+
}
100110
}
101111

102112
pub(crate) fn to_napi_error(error: RuntimeError) -> Error {

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

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -5,16 +5,10 @@ use super::{RuntimeError, RuntimeResult};
55
pub(crate) const RUNTIME_MIGRATIONS: &str = include_str!("sql/runtime_migrations.sql");
66

77
pub(crate) async fn migrate_runtime_tables(pool: &PgPool) -> RuntimeResult<()> {
8-
for statement in RUNTIME_MIGRATIONS
9-
.split(';')
10-
.map(str::trim)
11-
.filter(|statement| !statement.is_empty())
12-
{
13-
sqlx::query(statement)
14-
.execute(pool)
15-
.await
16-
.map_err(|err| RuntimeError::database("Runtime migration failed", err))?;
17-
}
8+
sqlx::raw_sql(RUNTIME_MIGRATIONS)
9+
.execute(pool)
10+
.await
11+
.map_err(|err| RuntimeError::database("Runtime migration failed", err))?;
1812

1913
Ok(())
2014
}

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

Lines changed: 26 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ CREATE TABLE IF NOT EXISTS runtime_leases (
3939
CREATE INDEX IF NOT EXISTS runtime_leases_expires_at_idx
4040
ON runtime_leases (expires_at);
4141

42-
CREATE TABLE IF NOT EXISTS blob_reconciliation_runs (
42+
CREATE TABLE IF NOT EXISTS storage_reconciliation_runs (
4343
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
4444
kind TEXT NOT NULL,
4545
mode TEXT NOT NULL,
@@ -54,10 +54,10 @@ CREATE TABLE IF NOT EXISTS blob_reconciliation_runs (
5454
metadata JSONB NOT NULL DEFAULT '{}'
5555
);
5656

57-
CREATE INDEX IF NOT EXISTS blob_reconciliation_runs_workspace_idx
58-
ON blob_reconciliation_runs (workspace_id, started_at DESC);
57+
CREATE INDEX IF NOT EXISTS storage_reconciliation_runs_workspace_idx
58+
ON storage_reconciliation_runs (workspace_id, started_at DESC);
5959

60-
CREATE TABLE IF NOT EXISTS blob_reconciliation_checkpoints (
60+
CREATE TABLE IF NOT EXISTS storage_reconciliation_checkpoints (
6161
kind TEXT NOT NULL,
6262
scope TEXT NOT NULL,
6363
status TEXT NOT NULL,
@@ -70,8 +70,28 @@ CREATE TABLE IF NOT EXISTS blob_reconciliation_checkpoints (
7070
PRIMARY KEY (kind, scope)
7171
);
7272

73-
CREATE INDEX IF NOT EXISTS blob_reconciliation_checkpoints_status_idx
74-
ON blob_reconciliation_checkpoints (kind, status, updated_at DESC);
73+
CREATE INDEX IF NOT EXISTS storage_reconciliation_checkpoints_status_idx
74+
ON storage_reconciliation_checkpoints (kind, status, updated_at DESC);
75+
76+
CREATE TABLE IF NOT EXISTS document_cleanup_candidates (
77+
workspace_id TEXT NOT NULL,
78+
doc_id TEXT NOT NULL,
79+
status TEXT NOT NULL CHECK (status IN ('marked', 'effects_pending', 'failed')),
80+
missing_since TIMESTAMPTZ(3) NOT NULL,
81+
last_observed_missing_at TIMESTAMPTZ(3) NOT NULL,
82+
last_doc_activity_at TIMESTAMPTZ(3),
83+
cleanup_payload JSONB NOT NULL DEFAULT '{}',
84+
attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
85+
error TEXT,
86+
updated_at TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
87+
PRIMARY KEY (workspace_id, doc_id)
88+
);
89+
90+
CREATE INDEX IF NOT EXISTS document_cleanup_candidates_status_missing_idx
91+
ON document_cleanup_candidates (status, missing_since);
92+
93+
CREATE INDEX IF NOT EXISTS document_cleanup_candidates_workspace_status_idx
94+
ON document_cleanup_candidates (workspace_id, status, updated_at DESC);
7595

7696
CREATE TABLE IF NOT EXISTS doc_blob_refs (
7797
workspace_id TEXT NOT NULL,

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

Lines changed: 45 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ fn push_workspace_once(workspace_ids: &mut Vec<String>, workspace_id: &str) {
3535

3636
async fn checkpoint_completed(pool: &PgPool, kind: &str, scope: &str) -> RuntimeResult<bool> {
3737
sqlx::query_scalar::<_, bool>(
38-
"SELECT EXISTS(SELECT 1 FROM blob_reconciliation_checkpoints WHERE kind = $1 AND scope = $2 AND status = \
38+
"SELECT EXISTS(SELECT 1 FROM storage_reconciliation_checkpoints WHERE kind = $1 AND scope = $2 AND status = \
3939
'completed')",
4040
)
4141
.bind(kind)
@@ -46,15 +46,51 @@ async fn checkpoint_completed(pool: &PgPool, kind: &str, scope: &str) -> Runtime
4646
}
4747

4848
async fn projection_is_stale(pool: &PgPool, workspace_id: &str) -> RuntimeResult<bool> {
49-
let checkpoint_fresh = checkpoint_completed(pool, "doc_blob_refs", workspace_id).await?;
49+
let checkpoint_completed_at = sqlx::query_scalar::<_, Option<chrono::DateTime<chrono::Utc>>>(
50+
r#"
51+
SELECT MIN(completed_at)
52+
FROM storage_reconciliation_checkpoints
53+
WHERE scope = $1
54+
AND kind IN ('document_cleanup', 'doc_blob_refs')
55+
AND status = 'completed'
56+
HAVING COUNT(*) = 2
57+
"#,
58+
)
59+
.bind(workspace_id)
60+
.fetch_optional(pool)
61+
.await
62+
.map_err(|err| RuntimeError::database("Blob cleanup retention checkpoint load failed", err))?
63+
.flatten();
64+
let Some(checkpoint_completed_at) = checkpoint_completed_at else {
65+
return Ok(true);
66+
};
67+
let activity_after_checkpoint = sqlx::query_scalar::<_, bool>(
68+
r#"
69+
SELECT EXISTS(
70+
SELECT 1 FROM snapshots
71+
WHERE workspace_id = $1 AND updated_at > $2
72+
UNION ALL
73+
SELECT 1 FROM updates
74+
WHERE workspace_id = $1 AND created_at > $2
75+
UNION ALL
76+
SELECT 1 FROM snapshot_histories
77+
WHERE workspace_id = $1 AND timestamp > $2
78+
)
79+
"#,
80+
)
81+
.bind(workspace_id)
82+
.bind(checkpoint_completed_at)
83+
.fetch_one(pool)
84+
.await
85+
.map_err(|err| RuntimeError::database("Blob cleanup retention activity check failed", err))?;
5086
let has_stale_rows = sqlx::query_scalar::<_, bool>(
5187
"SELECT EXISTS(SELECT 1 FROM doc_blob_refs WHERE workspace_id = $1 AND status <> 'fresh')",
5288
)
5389
.bind(workspace_id)
5490
.fetch_one(pool)
5591
.await
5692
.map_err(|err| RuntimeError::database("Blob cleanup projection freshness check failed", err))?;
57-
Ok(!checkpoint_fresh || has_stale_rows)
93+
Ok(activity_after_checkpoint || has_stale_rows)
5894
}
5995

6096
async fn stale_projection_workspaces(pool: &PgPool, workspace_id: &str) -> RuntimeResult<Vec<String>> {
@@ -170,7 +206,7 @@ async fn load_completed_blobs(
170206

171207
async fn load_plan_cursor(pool: &PgPool, workspace_id: &str) -> RuntimeResult<Option<String>> {
172208
let row = sqlx::query_as::<_, (String, serde_json::Value)>(
173-
"SELECT status, cursor FROM blob_reconciliation_checkpoints WHERE kind = 'blob_cleanup_plan' AND scope = $1",
209+
"SELECT status, cursor FROM storage_reconciliation_checkpoints WHERE kind = 'blob_cleanup_plan' AND scope = $1",
174210
)
175211
.bind(workspace_id)
176212
.fetch_optional(pool)
@@ -199,13 +235,13 @@ async fn upsert_plan_checkpoint(
199235
let status = if completed { "completed" } else { "running" };
200236
sqlx::query(
201237
r#"
202-
INSERT INTO blob_reconciliation_checkpoints
238+
INSERT INTO storage_reconciliation_checkpoints
203239
(kind, scope, status, cursor, last_key, completed_at)
204240
VALUES ('blob_cleanup_plan', $1, $2, $3, $4, CASE WHEN $5 THEN CURRENT_TIMESTAMP ELSE NULL END)
205241
ON CONFLICT (kind, scope) DO UPDATE
206242
SET status = EXCLUDED.status,
207243
cursor = EXCLUDED.cursor,
208-
last_key = COALESCE(EXCLUDED.last_key, blob_reconciliation_checkpoints.last_key),
244+
last_key = COALESCE(EXCLUDED.last_key, storage_reconciliation_checkpoints.last_key),
209245
completed_at = CASE WHEN $5 THEN CURRENT_TIMESTAMP ELSE NULL END,
210246
updated_at = CURRENT_TIMESTAMP
211247
"#,
@@ -224,7 +260,7 @@ async fn upsert_plan_checkpoint(
224260
async fn create_run(pool: &PgPool, workspace_id: &str) -> RuntimeResult<String> {
225261
sqlx::query_scalar::<_, String>(
226262
r#"
227-
INSERT INTO blob_reconciliation_runs (kind, mode, status, workspace_id)
263+
INSERT INTO storage_reconciliation_runs (kind, mode, status, workspace_id)
228264
VALUES ('blob_cleanup_plan', 'mark_only', 'running', $1)
229265
RETURNING id::text
230266
"#,
@@ -252,7 +288,7 @@ async fn finish_run(
252288
.unwrap_or(0);
253289
sqlx::query(
254290
r#"
255-
UPDATE blob_reconciliation_runs
291+
UPDATE storage_reconciliation_runs
256292
SET status = 'finished',
257293
finished_at = CURRENT_TIMESTAMP,
258294
scanned = $2,
@@ -318,7 +354,7 @@ async fn finish_execute_run(
318354
) -> RuntimeResult<()> {
319355
sqlx::query(
320356
r#"
321-
UPDATE blob_reconciliation_runs
357+
UPDATE storage_reconciliation_runs
322358
SET status = 'finished',
323359
finished_at = CURRENT_TIMESTAMP,
324360
scanned = $2,

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

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,8 @@ impl BackfillCheckpoint {
8585

8686
async fn load_checkpoint(pool: &PgPool, scope: &str) -> RuntimeResult<Option<BackfillCheckpoint>> {
8787
sqlx::query_as::<_, BackfillCheckpoint>(
88-
"SELECT last_key, cursor FROM blob_reconciliation_checkpoints WHERE kind = 'blob_metadata_backfill' AND scope = $1",
88+
"SELECT last_key, cursor FROM storage_reconciliation_checkpoints WHERE kind = 'blob_metadata_backfill' AND scope \
89+
= $1",
8990
)
9091
.bind(scope)
9192
.fetch_optional(pool)
@@ -103,13 +104,13 @@ async fn upsert_checkpoint(
103104
let status = if completed { "completed" } else { "running" };
104105
sqlx::query(
105106
r#"
106-
INSERT INTO blob_reconciliation_checkpoints
107+
INSERT INTO storage_reconciliation_checkpoints
107108
(kind, scope, status, cursor, last_key, completed_at, metadata)
108109
VALUES ('blob_metadata_backfill', $1, $2, $3, $4, CASE WHEN $5 THEN CURRENT_TIMESTAMP ELSE NULL END, $6)
109110
ON CONFLICT (kind, scope) DO UPDATE
110111
SET status = EXCLUDED.status,
111112
cursor = EXCLUDED.cursor,
112-
last_key = COALESCE(EXCLUDED.last_key, blob_reconciliation_checkpoints.last_key),
113+
last_key = COALESCE(EXCLUDED.last_key, storage_reconciliation_checkpoints.last_key),
113114
completed_at = CASE WHEN $5 THEN CURRENT_TIMESTAMP ELSE NULL END,
114115
updated_at = CURRENT_TIMESTAMP,
115116
metadata = EXCLUDED.metadata
@@ -247,7 +248,7 @@ impl StorageRuntime {
247248

248249
sqlx::query(
249250
r#"
250-
INSERT INTO blob_reconciliation_runs
251+
INSERT INTO storage_reconciliation_runs
251252
(kind, mode, status, workspace_id, finished_at, scanned, changed, failed, metadata)
252253
VALUES ('blob_metadata_backfill', 'execute', 'finished', $1, CURRENT_TIMESTAMP, $2, $3, $4, $5)
253254
"#,

0 commit comments

Comments
 (0)