Our Personal Data Server from scratch!
feat(pds): selfhealing repo writing by detecting corruption & retrying
Lewis: May this revision serve well! <lu5a@proton.me>
+523
-118
+56
-44
crates/tranquil-api/src/repo/record/batch.rs
+56
-44
crates/tranquil-api/src/repo/record/batch.rs
···
14
14
};
15
15
use tranquil_pds::repo::TrackingBlockStore;
16
16
use tranquil_pds::repo_ops::{
17
-
FinalizeParams, RecordOp, begin_repo_write, extract_backlinks, extract_blob_cids,
18
-
finalize_repo_write,
17
+
CommitResult, FinalizeParams, RecordOp, begin_repo_write, extract_backlinks, extract_blob_cids,
18
+
finalize_repo_write, with_repair_retry,
19
19
};
20
20
use tranquil_pds::state::AppState;
21
21
use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey};
···
74
74
if mst
75
75
.get(&key)
76
76
.await
77
-
.map_err(|e| ApiError::InternalError(Some(format!("Failed to read MST: {e}"))))?
77
+
.map_err(|e| ApiError::from_mst_error("read MST for applyWrites create", &e))?
78
78
.is_some()
79
79
{
80
80
return Err(ApiError::InvalidRequest(format!(
···
92
92
let new_mst = mst
93
93
.add(&key, record_cid)
94
94
.await
95
-
.map_err(|_| ApiError::InternalError(Some("Failed to add to MST".into())))?;
95
+
.map_err(|e| ApiError::from_mst_error("add record to MST", &e))?;
96
96
let uri = AtUri::from_parts(did, collection, &rkey);
97
97
backlinks_to_add.extend(extract_backlinks(&uri, value));
98
98
results.push(WriteResult::CreateResult {
···
138
138
let prev_record_cid = mst
139
139
.get(&key)
140
140
.await
141
-
.map_err(|e| {
142
-
ApiError::InternalError(Some(format!("Failed to read prev record: {}", e)))
143
-
})?
141
+
.map_err(|e| ApiError::from_mst_error("read update target from MST", &e))?
144
142
.ok_or_else(|| {
145
143
ApiError::InvalidRequest("Update target record does not exist".into())
146
144
})?;
···
155
153
let new_mst = mst
156
154
.update(&key, record_cid)
157
155
.await
158
-
.map_err(|_| ApiError::InternalError(Some("Failed to update MST".into())))?;
156
+
.map_err(|e| ApiError::from_mst_error("update record in MST", &e))?;
159
157
let uri = AtUri::from_parts(did, collection, rkey);
160
158
backlinks_to_remove.push(uri.clone());
161
159
backlinks_to_add.extend(extract_backlinks(&uri, value));
···
184
182
let prev_record_cid = mst
185
183
.get(&key)
186
184
.await
187
-
.map_err(|e| {
188
-
ApiError::InternalError(Some(format!("Failed to read prev record: {}", e)))
189
-
})?
185
+
.map_err(|e| ApiError::from_mst_error("read delete target from MST", &e))?
190
186
.ok_or_else(|| {
191
187
ApiError::InvalidRequest("Delete target record does not exist".into())
192
188
})?;
193
189
let new_mst = mst
194
190
.delete(&key)
195
191
.await
196
-
.map_err(|_| ApiError::InternalError(Some("Failed to delete from MST".into())))?;
192
+
.map_err(|e| ApiError::from_mst_error("delete record from MST", &e))?;
197
193
backlinks_to_remove.push(AtUri::from_parts(did, collection, rkey));
198
194
results.push(WriteResult::DeleteResult {});
199
195
ops.push(RecordOp::Delete {
···
236
232
.await
237
233
}
238
234
235
+
async fn execute_apply_writes(
236
+
state: &AppState,
237
+
user_id: uuid::Uuid,
238
+
did: &Did,
239
+
input: &ApplyWritesInput,
240
+
controller_did: Option<&Did>,
241
+
write_summary: Option<serde_json::Value>,
242
+
) -> Result<(CommitResult, Vec<WriteResult>), ApiError> {
243
+
let (ctx, mst) = begin_repo_write(state, user_id, input.swap_commit.as_deref()).await?;
244
+
245
+
let WriteAccumulator {
246
+
mst: final_mst,
247
+
results,
248
+
ops,
249
+
all_blob_cids,
250
+
backlinks_to_add,
251
+
backlinks_to_remove,
252
+
} = process_writes(&input.writes, mst, did, input.validate, &ctx.tracking_store).await?;
253
+
254
+
let commit_result = finalize_repo_write(
255
+
state,
256
+
ctx,
257
+
final_mst,
258
+
FinalizeParams {
259
+
did,
260
+
user_id,
261
+
controller_did,
262
+
delegation_detail: write_summary,
263
+
ops,
264
+
blob_cids: &all_blob_cids,
265
+
backlinks_to_add,
266
+
backlinks_to_remove,
267
+
},
268
+
)
269
+
.await?;
270
+
271
+
Ok((commit_result, results))
272
+
}
273
+
239
274
#[derive(Deserialize)]
240
275
#[serde(tag = "$type")]
241
276
pub enum WriteOp {
···
350
385
.log_db_err("fetching user for batch write")?
351
386
.ok_or(ApiError::InternalError(Some("User not found".into())))?;
352
387
353
-
let (ctx, mst) = begin_repo_write(&state, user_id, input.swap_commit.as_deref()).await?;
354
-
355
-
let WriteAccumulator {
356
-
mst: final_mst,
357
-
results,
358
-
ops,
359
-
all_blob_cids,
360
-
backlinks_to_add,
361
-
backlinks_to_remove,
362
-
} = process_writes(
363
-
&input.writes,
364
-
mst,
365
-
&did,
366
-
input.validate,
367
-
&ctx.tracking_store,
368
-
)
369
-
.await?;
370
-
371
388
let write_summary: Option<serde_json::Value> = controller_did.as_ref().map(|_| {
372
389
let writes: Vec<serde_json::Value> = input
373
390
.writes
···
401
418
})
402
419
});
403
420
404
-
let commit_result = finalize_repo_write(
405
-
&state,
406
-
ctx,
407
-
final_mst,
408
-
FinalizeParams {
409
-
did: &did,
421
+
let (commit_result, results) = with_repair_retry(&state, user_id, || {
422
+
execute_apply_writes(
423
+
&state,
410
424
user_id,
411
-
controller_did: controller_did.as_ref(),
412
-
delegation_detail: write_summary,
413
-
ops,
414
-
blob_cids: &all_blob_cids,
415
-
backlinks_to_add,
416
-
backlinks_to_remove,
417
-
},
418
-
)
425
+
&did,
426
+
&input,
427
+
controller_did.as_ref(),
428
+
write_summary.clone(),
429
+
)
430
+
})
419
431
.await?;
420
432
421
433
Ok(Json(ApplyWritesOutput {
+40
-21
crates/tranquil-api/src/repo/record/delete.rs
+40
-21
crates/tranquil-api/src/repo/record/delete.rs
···
4
4
use serde::{Deserialize, Serialize};
5
5
use serde_json::json;
6
6
use std::str::FromStr;
7
-
use tracing::error;
8
7
use tranquil_pds::api::error::ApiError;
9
8
use tranquil_pds::auth::{Active, Auth, VerifyScope};
10
9
use tranquil_pds::cid_types::RecordCid;
11
-
use tranquil_pds::repo_ops::{FinalizeParams, RecordOp, begin_repo_write, finalize_repo_write};
10
+
use tranquil_pds::repo_ops::{
11
+
FinalizeParams, RecordOp, begin_repo_write, finalize_repo_write, with_repair_retry,
12
+
};
12
13
use tranquil_pds::state::AppState;
13
-
use tranquil_pds::types::{AtIdentifier, AtUri, Nsid, Rkey};
14
+
use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey};
15
+
use uuid::Uuid;
14
16
15
17
#[derive(Deserialize)]
16
18
pub struct DeleteRecordInput {
···
41
43
let user_id = repo_auth.user_id;
42
44
let controller_did = repo_auth.controller_did;
43
45
44
-
let (ctx, mst) = begin_repo_write(&state, user_id, input.swap_commit.as_deref()).await?;
46
+
let out = with_repair_retry(&state, user_id, || {
47
+
delete_record_inner(&state, &did, user_id, controller_did.as_ref(), &input)
48
+
})
49
+
.await?;
50
+
Ok(Json(out))
51
+
}
52
+
53
+
async fn delete_record_inner(
54
+
state: &AppState,
55
+
did: &Did,
56
+
user_id: Uuid,
57
+
controller_did: Option<&Did>,
58
+
input: &DeleteRecordInput,
59
+
) -> Result<DeleteRecordOutput, ApiError> {
60
+
let (ctx, mst) = begin_repo_write(state, user_id, input.swap_commit.as_deref()).await?;
45
61
46
62
let key = format!("{}/{}", input.collection, input.rkey);
47
63
48
64
if let Some(swap_record_str) = &input.swap_record {
49
65
let expected_cid = Cid::from_str(swap_record_str).ok();
50
-
let actual_cid = mst.get(&key).await.ok().flatten();
66
+
let actual_cid = mst
67
+
.get(&key)
68
+
.await
69
+
.map_err(|e| ApiError::from_mst_error("read swap target from MST", &e))?;
51
70
if expected_cid != actual_cid {
52
71
return Err(ApiError::InvalidSwap(Some(
53
72
"Record has been modified or does not exist".into(),
···
55
74
}
56
75
}
57
76
58
-
let prev_record_cid = mst.get(&key).await.map_err(|e| {
59
-
error!("Failed to read prev record from MST: {}", e);
60
-
ApiError::InternalError(Some("Failed to read MST".into()))
61
-
})?;
77
+
let prev_record_cid = mst
78
+
.get(&key)
79
+
.await
80
+
.map_err(|e| ApiError::from_mst_error("read prev record from MST", &e))?;
62
81
let Some(prev_record_cid) = prev_record_cid else {
63
-
return Ok(Json(DeleteRecordOutput { commit: None }));
82
+
return Ok(DeleteRecordOutput { commit: None });
64
83
};
65
84
66
-
let new_mst = mst.delete(&key).await.map_err(|e| {
67
-
error!("Failed to delete from MST: {}", e);
68
-
ApiError::InternalError(Some("Failed to delete from MST".into()))
69
-
})?;
85
+
let new_mst = mst
86
+
.delete(&key)
87
+
.await
88
+
.map_err(|e| ApiError::from_mst_error("delete record from MST", &e))?;
70
89
71
90
let op = RecordOp::Delete {
72
91
collection: input.collection.clone(),
···
74
93
prev: RecordCid::from(prev_record_cid),
75
94
};
76
95
77
-
let deleted_uri = AtUri::from_parts(&did, &input.collection, &input.rkey);
96
+
let deleted_uri = AtUri::from_parts(did, &input.collection, &input.rkey);
78
97
79
98
let commit_result = finalize_repo_write(
80
-
&state,
99
+
state,
81
100
ctx,
82
101
new_mst,
83
102
FinalizeParams {
84
-
did: &did,
103
+
did,
85
104
user_id,
86
-
controller_did: controller_did.as_ref(),
87
-
delegation_detail: controller_did.as_ref().map(|_| {
105
+
controller_did,
106
+
delegation_detail: controller_did.map(|_| {
88
107
json!({
89
108
"action": "delete",
90
109
"collection": input.collection,
···
99
118
)
100
119
.await?;
101
120
102
-
Ok(Json(DeleteRecordOutput {
121
+
Ok(DeleteRecordOutput {
103
122
commit: Some(CommitInfo {
104
123
cid: commit_result.commit_cid.to_string(),
105
124
rev: commit_result.rev,
106
125
}),
107
-
}))
126
+
})
108
127
}
+68
-43
crates/tranquil-api/src/repo/record/write.rs
+68
-43
crates/tranquil-api/src/repo/record/write.rs
···
7
7
use serde_json::json;
8
8
use std::borrow::Cow;
9
9
use std::str::FromStr;
10
-
use tracing::error;
11
10
use tranquil_pds::api::error::{ApiError, DbResultExt};
12
11
use tranquil_pds::auth::{
13
12
Active, Auth, AuthSource, RepoScopeAction, ScopeVerified, VerifyScope, require_not_migrated,
···
15
14
};
16
15
use tranquil_pds::repo_ops::{
17
16
FinalizeParams, RecordOp, begin_repo_write, extract_backlinks, extract_blob_cids,
18
-
finalize_repo_write,
17
+
finalize_repo_write, with_repair_retry,
19
18
};
20
19
use tranquil_pds::state::AppState;
21
20
use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey};
···
130
129
let user_id = repo_auth.user_id;
131
130
let controller_did = repo_auth.controller_did;
132
131
133
-
let (ctx, mut mst) = begin_repo_write(&state, user_id, input.swap_commit.as_deref()).await?;
132
+
let out = with_repair_retry(&state, user_id, || {
133
+
create_record_inner(&state, &did, user_id, controller_did.as_ref(), &input)
134
+
})
135
+
.await?;
136
+
Ok(Json(out))
137
+
}
138
+
139
+
async fn create_record_inner(
140
+
state: &AppState,
141
+
did: &Did,
142
+
user_id: Uuid,
143
+
controller_did: Option<&Did>,
144
+
input: &CreateRecordInput,
145
+
) -> Result<CreateRecordOutput, ApiError> {
146
+
let (ctx, mut mst) = begin_repo_write(state, user_id, input.swap_commit.as_deref()).await?;
134
147
135
148
let validation_status = if input.validate.should_skip() {
136
149
None
···
146
159
)
147
160
};
148
161
149
-
let rkey = input.rkey.unwrap_or_else(Rkey::generate);
162
+
let rkey = input.rkey.clone().unwrap_or_else(Rkey::generate);
150
163
let mut ops: Vec<RecordOp> = Vec::new();
151
164
let mut conflict_uris_to_cleanup: Vec<AtUri> = Vec::new();
152
165
153
166
if !input.validate.should_skip() {
154
-
let record_uri = AtUri::from_parts(&did, &input.collection, &rkey);
167
+
let record_uri = AtUri::from_parts(did, &input.collection, &rkey);
155
168
let backlinks = extract_backlinks(&record_uri, &input.record);
156
169
157
170
if !backlinks.is_empty() {
···
176
189
Ok(Some(cid)) => cid,
177
190
Ok(None) => continue,
178
191
Err(e) => {
179
-
error!(
180
-
"Failed to read conflict record from MST {}: {:?}",
181
-
conflict_uri, e
182
-
);
183
-
return Err(ApiError::InternalError(Some(
184
-
"Failed to read conflicting record from MST".into(),
185
-
)));
192
+
return Err(ApiError::from_mst_error(
193
+
&format!("read conflict record from MST {conflict_uri}"),
194
+
&e,
195
+
));
186
196
}
187
197
};
188
198
189
199
mst = mst.delete(&conflict_key).await.map_err(|e| {
190
-
error!(
191
-
"Failed to delete conflict from MST {}: {:?}",
192
-
conflict_uri, e
193
-
);
194
-
ApiError::InternalError(Some(
195
-
"Failed to delete conflicting record from MST".into(),
196
-
))
200
+
ApiError::from_mst_error(
201
+
&format!("delete conflict from MST {conflict_uri}"),
202
+
&e,
203
+
)
197
204
})?;
198
205
199
206
ops.push(RecordOp::Delete {
···
210
217
if mst
211
218
.get(&key)
212
219
.await
213
-
.map_err(|e| ApiError::InternalError(Some(format!("Failed to read MST: {e}"))))?
220
+
.map_err(|e| ApiError::from_mst_error("read MST for create existence check", &e))?
214
221
.is_some()
215
222
{
216
223
return Err(ApiError::InvalidRequest(format!(
···
229
236
mst = mst
230
237
.add(&key, record_cid)
231
238
.await
232
-
.map_err(|_| ApiError::InternalError(Some("Failed to add to MST".into())))?;
239
+
.map_err(|e| ApiError::from_mst_error("add record to MST", &e))?;
233
240
234
241
ops.push(RecordOp::Create {
235
242
collection: input.collection.clone(),
···
239
246
240
247
let blob_cids = extract_blob_cids(&input.record);
241
248
242
-
let created_uri = AtUri::from_parts(&did, &input.collection, &rkey);
249
+
let created_uri = AtUri::from_parts(did, &input.collection, &rkey);
243
250
let backlinks_to_add = extract_backlinks(&created_uri, &input.record);
244
251
245
252
let commit_result = finalize_repo_write(
246
-
&state,
253
+
state,
247
254
ctx,
248
255
mst,
249
256
FinalizeParams {
250
-
did: &did,
257
+
did,
251
258
user_id,
252
-
controller_did: controller_did.as_ref(),
253
-
delegation_detail: controller_did.as_ref().map(|_| {
259
+
controller_did,
260
+
delegation_detail: controller_did.map(|_| {
254
261
json!({
255
262
"action": "create",
256
263
"collection": input.collection,
···
265
272
)
266
273
.await?;
267
274
268
-
Ok(Json(CreateRecordOutput {
275
+
Ok(CreateRecordOutput {
269
276
uri: created_uri,
270
277
cid: record_cid.to_string(),
271
278
commit: CommitInfo {
···
273
280
rev: commit_result.rev,
274
281
},
275
282
validation_status,
276
-
}))
283
+
})
277
284
}
278
285
279
286
#[derive(Deserialize)]
···
316
323
let user_id = repo_auth.user_id;
317
324
let controller_did = repo_auth.controller_did;
318
325
319
-
let (ctx, mst) = begin_repo_write(&state, user_id, input.swap_commit.as_deref()).await?;
326
+
let out = with_repair_retry(&state, user_id, || {
327
+
put_record_inner(&state, &did, user_id, controller_did.as_ref(), &input)
328
+
})
329
+
.await?;
330
+
Ok(Json(out))
331
+
}
332
+
333
+
async fn put_record_inner(
334
+
state: &AppState,
335
+
did: &Did,
336
+
user_id: Uuid,
337
+
controller_did: Option<&Did>,
338
+
input: &PutRecordInput,
339
+
) -> Result<PutRecordOutput, ApiError> {
340
+
let (ctx, mst) = begin_repo_write(state, user_id, input.swap_commit.as_deref()).await?;
320
341
321
342
let validation_status = if input.validate.should_skip() {
322
343
None
···
334
355
335
356
let key = format!("{}/{}", input.collection, input.rkey);
336
357
358
+
let read_cid = |r: Result<Option<Cid>, jacquard_repo::error::RepoError>| {
359
+
r.map_err(|e| ApiError::from_mst_error("read MST for put", &e))
360
+
};
361
+
337
362
if let Some(swap_record_str) = &input.swap_record {
338
363
let expected_cid = Cid::from_str(swap_record_str).ok();
339
-
let actual_cid = mst.get(&key).await.ok().flatten();
364
+
let actual_cid = read_cid(mst.get(&key).await)?;
340
365
if expected_cid != actual_cid {
341
366
return Err(ApiError::InvalidSwap(Some(
342
367
"Record has been modified or does not exist".into(),
···
344
369
}
345
370
}
346
371
347
-
let existing_cid = mst.get(&key).await.ok().flatten();
372
+
let existing_cid = read_cid(mst.get(&key).await)?;
348
373
let record_ipld = tranquil_pds::util::json_to_ipld(&input.record);
349
374
let record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld)
350
375
.map_err(|_| ApiError::InvalidRecord("Failed to serialize record".into()))?;
···
355
380
.map_err(|_| ApiError::InternalError(Some("Failed to save record block".into())))?;
356
381
357
382
if existing_cid == Some(record_cid) {
358
-
return Ok(Json(PutRecordOutput {
359
-
uri: AtUri::from_parts(&did, &input.collection, &input.rkey),
383
+
return Ok(PutRecordOutput {
384
+
uri: AtUri::from_parts(did, &input.collection, &input.rkey),
360
385
cid: record_cid.to_string(),
361
386
commit: None,
362
387
validation_status,
363
-
}));
388
+
});
364
389
}
365
390
366
-
let record_uri = AtUri::from_parts(&did, &input.collection, &input.rkey);
391
+
let record_uri = AtUri::from_parts(did, &input.collection, &input.rkey);
367
392
let (new_mst, op, is_update, backlinks_to_remove) = match existing_cid {
368
393
Some(prev_cid) => {
369
394
let new_mst = mst
370
395
.update(&key, record_cid)
371
396
.await
372
-
.map_err(|_| ApiError::InternalError(Some("Failed to update MST".into())))?;
397
+
.map_err(|e| ApiError::from_mst_error("update record in MST", &e))?;
373
398
let op = RecordOp::Update {
374
399
collection: input.collection.clone(),
375
400
rkey: input.rkey.clone(),
···
382
407
let new_mst = mst
383
408
.add(&key, record_cid)
384
409
.await
385
-
.map_err(|_| ApiError::InternalError(Some("Failed to add to MST".into())))?;
410
+
.map_err(|e| ApiError::from_mst_error("add record to MST", &e))?;
386
411
let op = RecordOp::Create {
387
412
collection: input.collection.clone(),
388
413
rkey: input.rkey.clone(),
···
396
421
let backlinks_to_add = extract_backlinks(&record_uri, &input.record);
397
422
398
423
let commit_result = finalize_repo_write(
399
-
&state,
424
+
state,
400
425
ctx,
401
426
new_mst,
402
427
FinalizeParams {
403
-
did: &did,
428
+
did,
404
429
user_id,
405
-
controller_did: controller_did.as_ref(),
406
-
delegation_detail: controller_did.as_ref().map(|_| {
430
+
controller_did,
431
+
delegation_detail: controller_did.map(|_| {
407
432
json!({
408
433
"action": if is_update { "update" } else { "create" },
409
434
"collection": input.collection,
···
418
443
)
419
444
.await?;
420
445
421
-
Ok(Json(PutRecordOutput {
446
+
Ok(PutRecordOutput {
422
447
uri: record_uri,
423
448
cid: record_cid.to_string(),
424
449
commit: Some(CommitInfo {
···
426
451
rev: commit_result.rev,
427
452
}),
428
453
validation_status,
429
-
}))
454
+
})
430
455
}
+30
-2
crates/tranquil-pds/src/api/error.rs
+30
-2
crates/tranquil-pds/src/api/error.rs
···
15
15
#[derive(Debug)]
16
16
pub enum ApiError {
17
17
InternalError(Option<String>),
18
+
RepoCorruption,
18
19
AuthenticationRequired,
19
20
AuthenticationFailed(Option<String>),
20
21
InvalidRequest(String),
···
121
122
},
122
123
}
123
124
125
+
const MST_NODE_MISSING_MARKER: &str = "MST node not found";
126
+
124
127
impl ApiError {
128
+
pub fn is_repo_corruption(&self) -> bool {
129
+
matches!(self, Self::RepoCorruption)
130
+
}
131
+
132
+
pub fn detail_is_repo_corruption(detail: &str) -> bool {
133
+
detail.contains(tranquil_store::blockstore::BLOCK_CORRUPTION_MARKER)
134
+
|| detail.contains(MST_NODE_MISSING_MARKER)
135
+
}
136
+
137
+
pub fn from_mst_error(context: &str, e: &jacquard_repo::error::RepoError) -> Self {
138
+
let detail = format!("{e:#}");
139
+
if Self::detail_is_repo_corruption(&detail) {
140
+
tracing::warn!("{context}: repairable MST damage: {detail}");
141
+
Self::RepoCorruption
142
+
} else {
143
+
tracing::error!("{context}: {detail}");
144
+
Self::InternalError(None)
145
+
}
146
+
}
147
+
125
148
fn status_code(&self) -> StatusCode {
126
149
match self {
127
-
Self::InternalError(_) | Self::DatabaseError => StatusCode::INTERNAL_SERVER_ERROR,
150
+
Self::InternalError(_) | Self::RepoCorruption | Self::DatabaseError => {
151
+
StatusCode::INTERNAL_SERVER_ERROR
152
+
}
128
153
Self::UpstreamFailure | Self::UpstreamUnavailable(_) | Self::UpstreamErrorMsg(_) => {
129
154
StatusCode::BAD_GATEWAY
130
155
}
···
223
248
}
224
249
fn error_name(&self) -> Cow<'static, str> {
225
250
match self {
226
-
Self::InternalError(_) | Self::DatabaseError => Cow::Borrowed("InternalServerError"),
251
+
Self::InternalError(_) | Self::RepoCorruption | Self::DatabaseError => {
252
+
Cow::Borrowed("InternalServerError")
253
+
}
227
254
Self::UpstreamFailure | Self::UpstreamUnavailable(_) | Self::UpstreamErrorMsg(_) => {
228
255
Cow::Borrowed("UpstreamError")
229
256
}
···
332
359
Self::InternalError(msg) => msg
333
360
.clone()
334
361
.unwrap_or_else(|| "Internal Server Error".into()),
362
+
Self::RepoCorruption => "Internal Server Error".into(),
335
363
Self::AuthenticationFailed(msg) => msg
336
364
.clone()
337
365
.unwrap_or_else(|| "Authentication failed".into()),
+17
-1
crates/tranquil-pds/src/repo/mod.rs
+17
-1
crates/tranquil-pds/src/repo/mod.rs
···
7
7
use jacquard_repo::error::RepoError;
8
8
use jacquard_repo::repo::CommitData;
9
9
use jacquard_repo::storage::BlockStore;
10
-
use tranquil_store::blockstore::TranquilBlockStore;
10
+
use tranquil_store::blockstore::{RepairOutcome, TranquilBlockStore};
11
11
use tranquil_store::{RealIO, SystemClock};
12
12
13
13
#[derive(Clone)]
···
35
35
match self {
36
36
Self::Postgres(_) => Ok(()),
37
37
Self::TranquilStore(s) => s.decrement_refs(cids).await,
38
+
}
39
+
}
40
+
41
+
pub async fn repair_structure(
42
+
&self,
43
+
entries: &[(String, Cid)],
44
+
expected_root: Cid,
45
+
) -> Result<RepairOutcome, RepoError> {
46
+
match self {
47
+
Self::Postgres(_) => Ok(RepairOutcome {
48
+
nodes_total: 0,
49
+
nodes_repaired: 0,
50
+
}),
51
+
Self::TranquilStore(s) => {
52
+
tranquil_store::blockstore::rebuild_and_repair_mst(s, entries, expected_root).await
53
+
}
38
54
}
39
55
}
40
56
}
+298
-5
crates/tranquil-pds/src/repo_ops.rs
+298
-5
crates/tranquil-pds/src/repo_ops.rs
···
16
16
use serde_json::{Value, json};
17
17
use std::collections::{BTreeMap, HashMap, HashSet};
18
18
use std::str::FromStr;
19
-
use std::sync::Arc;
19
+
use std::sync::{Arc, LazyLock};
20
+
use std::time::{Duration, Instant};
20
21
use tokio::sync::OwnedMutexGuard;
21
22
use tracing::{error, warn};
22
23
use uuid::Uuid;
···
230
231
Ok((ctx, mst))
231
232
}
232
233
234
+
pub async fn repair_repo_structure(
235
+
state: &AppState,
236
+
user_id: Uuid,
237
+
) -> Result<tranquil_store::blockstore::RepairOutcome, ApiError> {
238
+
let _write_lock = state.repo_write_locks.lock(user_id).await;
239
+
240
+
let root_cid_str = state
241
+
.repos
242
+
.repo
243
+
.get_repo_root_cid_by_user_id(user_id)
244
+
.await
245
+
.map_err(|e| {
246
+
error!("repair: DB error fetching repo root: {}", e);
247
+
ApiError::InternalError(None)
248
+
})?
249
+
.ok_or_else(|| ApiError::InternalError(Some("Repo root not found".into())))?;
250
+
let current_root_cid = Cid::from_str(root_cid_str.as_str())
251
+
.map_err(|_| ApiError::InternalError(Some("Invalid repo root CID".into())))?;
252
+
253
+
let commit_bytes = state
254
+
.block_store
255
+
.get(¤t_root_cid)
256
+
.await
257
+
.map_err(|e| {
258
+
error!("repair: failed to load commit block: {}", e);
259
+
ApiError::InternalError(None)
260
+
})?
261
+
.ok_or_else(|| ApiError::InternalError(Some("Commit block not found".into())))?;
262
+
let data_root = Commit::from_cbor(&commit_bytes)
263
+
.map_err(|e| {
264
+
error!("repair: failed to parse commit: {}", e);
265
+
ApiError::InternalError(None)
266
+
})?
267
+
.data;
268
+
269
+
let records = state
270
+
.repos
271
+
.repo
272
+
.get_all_records(user_id)
273
+
.await
274
+
.map_err(|e| {
275
+
error!("repair: get_all_records failed: {}", e);
276
+
ApiError::InternalError(None)
277
+
})?;
278
+
let entries: Vec<(String, Cid)> = records
279
+
.into_iter()
280
+
.filter_map(|r| {
281
+
Cid::from_str(r.record_cid.as_str())
282
+
.ok()
283
+
.map(|cid| (format!("{}/{}", r.collection, r.rkey), cid))
284
+
})
285
+
.collect();
286
+
287
+
warn!(
288
+
user_id = %user_id,
289
+
records = entries.len(),
290
+
"repair: rebuilding full MST from record set"
291
+
);
292
+
293
+
state
294
+
.block_store
295
+
.repair_structure(&entries, data_root)
296
+
.await
297
+
.map_err(|e| {
298
+
error!("repair: structural repair failed: {}", e);
299
+
ApiError::InternalError(Some("Structural repair failed".into()))
300
+
})
301
+
}
302
+
303
+
pub async fn with_repair_retry<T, F, Fut>(
304
+
state: &AppState,
305
+
user_id: Uuid,
306
+
mut attempt: F,
307
+
) -> Result<T, ApiError>
308
+
where
309
+
F: FnMut() -> Fut,
310
+
Fut: std::future::Future<Output = Result<T, ApiError>>,
311
+
{
312
+
match attempt().await {
313
+
Err(e) if e.is_repo_corruption() => {
314
+
warn!(
315
+
"structural MST damage during repo write for user {user_id}, repairing and retrying"
316
+
);
317
+
match repair_repo_structure(state, user_id).await {
318
+
Ok(outcome) if outcome.nodes_repaired > 0 => attempt().await,
319
+
Ok(_) => {
320
+
warn!(
321
+
user_id = %user_id,
322
+
"structural repair rewrote no nodes; damage is not in the MST structure, returning original error without retry"
323
+
);
324
+
Err(e)
325
+
}
326
+
Err(repair_err) => {
327
+
error!(user_id = %user_id, "structural repair failed: {repair_err:?}");
328
+
Err(e)
329
+
}
330
+
}
331
+
}
332
+
other => other,
333
+
}
334
+
}
335
+
336
+
const REPAIR_COOLDOWN: Duration = Duration::from_secs(60);
337
+
const REPAIR_NOOP_COOLDOWN: Duration = Duration::from_secs(600);
338
+
339
+
struct RepairSlot {
340
+
in_flight: bool,
341
+
next_allowed: Instant,
342
+
}
343
+
344
+
struct RepairGuard {
345
+
slots: parking_lot::Mutex<HashMap<Uuid, RepairSlot>>,
346
+
}
347
+
348
+
impl RepairGuard {
349
+
fn try_claim(&self, user_id: Uuid, now: Instant) -> bool {
350
+
let mut slots = self.slots.lock();
351
+
let slot = slots.entry(user_id).or_insert(RepairSlot {
352
+
in_flight: false,
353
+
next_allowed: now,
354
+
});
355
+
if slot.in_flight || now < slot.next_allowed {
356
+
return false;
357
+
}
358
+
slot.in_flight = true;
359
+
true
360
+
}
361
+
362
+
fn release(&self, user_id: Uuid, now: Instant, cooldown: Duration) {
363
+
if let Some(slot) = self.slots.lock().get_mut(&user_id) {
364
+
slot.in_flight = false;
365
+
slot.next_allowed = now + cooldown;
366
+
}
367
+
}
368
+
}
369
+
370
+
static REPAIR_GUARD: LazyLock<RepairGuard> = LazyLock::new(|| RepairGuard {
371
+
slots: parking_lot::Mutex::new(HashMap::new()),
372
+
});
373
+
374
+
struct RepairLease {
375
+
user_id: Uuid,
376
+
cooldown: Duration,
377
+
}
378
+
379
+
impl Drop for RepairLease {
380
+
fn drop(&mut self) {
381
+
REPAIR_GUARD.release(self.user_id, Instant::now(), self.cooldown);
382
+
}
383
+
}
384
+
385
+
pub fn schedule_repo_repair(state: &AppState, user_id: Uuid) {
386
+
if !REPAIR_GUARD.try_claim(user_id, Instant::now()) {
387
+
return;
388
+
}
389
+
let state = state.clone();
390
+
tokio::spawn(async move {
391
+
let mut lease = RepairLease {
392
+
user_id,
393
+
cooldown: REPAIR_COOLDOWN,
394
+
};
395
+
match repair_repo_structure(&state, user_id).await {
396
+
Ok(outcome) => {
397
+
if outcome.nodes_repaired == 0 {
398
+
lease.cooldown = REPAIR_NOOP_COOLDOWN;
399
+
}
400
+
warn!(
401
+
user_id = %user_id,
402
+
nodes_repaired = outcome.nodes_repaired,
403
+
nodes_total = outcome.nodes_total,
404
+
"background MST repair complete"
405
+
);
406
+
}
407
+
Err(e) => error!(user_id = %user_id, "background MST repair failed: {e:?}"),
408
+
}
409
+
});
410
+
}
411
+
233
412
pub async fn finalize_repo_write(
234
413
state: &AppState,
235
414
ctx: RepoWriteContext,
236
415
mst: Mst<TrackingBlockStore>,
237
416
params: FinalizeParams<'_>,
238
417
) -> Result<CommitResult, ApiError> {
239
-
let new_mst_root = mst.persist().await.map_err(|e| {
240
-
error!("MST persist failed: {}", e);
241
-
ApiError::InternalError(None)
242
-
})?;
418
+
let new_mst_root = mst
419
+
.persist()
420
+
.await
421
+
.map_err(|e| ApiError::from_mst_error("MST persist", &e))?;
243
422
244
423
let written_bytes = ctx.tracking_store.take_written_blocks();
245
424
let new_tree_cids: Vec<Cid> = written_bytes.keys().copied().collect();
···
861
1040
.await
862
1041
.map_err(|e| CommitError::DatabaseError(format!("genesis commit event: {}", e)))
863
1042
}
1043
+
1044
+
#[cfg(test)]
1045
+
mod repair_guard_tests {
1046
+
use super::*;
1047
+
1048
+
fn guard() -> RepairGuard {
1049
+
RepairGuard {
1050
+
slots: parking_lot::Mutex::new(HashMap::new()),
1051
+
}
1052
+
}
1053
+
1054
+
#[test]
1055
+
fn claim_dedups_in_flight_then_respects_cooldown() {
1056
+
let g = guard();
1057
+
let user = Uuid::from_u128(1);
1058
+
let t0 = Instant::now();
1059
+
1060
+
assert!(g.try_claim(user, t0), "first claim must succeed");
1061
+
assert!(
1062
+
!g.try_claim(user, t0),
1063
+
"second claim while a repair is in flight must be rejected"
1064
+
);
1065
+
1066
+
g.release(user, t0, REPAIR_COOLDOWN);
1067
+
assert!(
1068
+
!g.try_claim(user, t0 + Duration::from_secs(1)),
1069
+
"claim within the cooldown window must be rejected"
1070
+
);
1071
+
assert!(
1072
+
g.try_claim(user, t0 + REPAIR_COOLDOWN + Duration::from_millis(1)),
1073
+
"claim after the cooldown window must succeed"
1074
+
);
1075
+
}
1076
+
1077
+
#[test]
1078
+
fn distinct_users_do_not_block_each_other() {
1079
+
let g = guard();
1080
+
let t0 = Instant::now();
1081
+
assert!(g.try_claim(Uuid::from_u128(1), t0));
1082
+
assert!(g.try_claim(Uuid::from_u128(2), t0));
1083
+
}
1084
+
1085
+
#[test]
1086
+
fn concurrent_claims_for_one_user_admit_exactly_one() {
1087
+
use std::sync::atomic::{AtomicUsize, Ordering};
1088
+
use std::sync::{Arc, Barrier};
1089
+
1090
+
let g = Arc::new(guard());
1091
+
let user = Uuid::from_u128(42);
1092
+
let now = Instant::now();
1093
+
let winners = Arc::new(AtomicUsize::new(0));
1094
+
let gate = Arc::new(Barrier::new(32));
1095
+
1096
+
let handles: Vec<_> = (0..32)
1097
+
.map(|_| {
1098
+
let g = Arc::clone(&g);
1099
+
let winners = Arc::clone(&winners);
1100
+
let gate = Arc::clone(&gate);
1101
+
std::thread::spawn(move || {
1102
+
gate.wait();
1103
+
if g.try_claim(user, now) {
1104
+
winners.fetch_add(1, Ordering::Relaxed);
1105
+
}
1106
+
})
1107
+
})
1108
+
.collect();
1109
+
handles
1110
+
.into_iter()
1111
+
.for_each(|h| h.join().expect("worker thread panicked"));
1112
+
1113
+
assert_eq!(
1114
+
winners.load(Ordering::Relaxed),
1115
+
1,
1116
+
"exactly one concurrent claim must win the dedup race"
1117
+
);
1118
+
}
1119
+
1120
+
#[test]
1121
+
fn release_re_enables_claim_after_cooldown_for_recurring_corruption() {
1122
+
let g = guard();
1123
+
let user = Uuid::from_u128(7);
1124
+
let t0 = Instant::now();
1125
+
1126
+
assert!(g.try_claim(user, t0));
1127
+
g.release(user, t0, REPAIR_COOLDOWN);
1128
+
let after_cooldown = t0 + REPAIR_COOLDOWN + Duration::from_millis(1);
1129
+
assert!(
1130
+
g.try_claim(user, after_cooldown),
1131
+
"a fresh corruption after the cooldown must be repairable again"
1132
+
);
1133
+
assert!(
1134
+
!g.try_claim(user, after_cooldown),
1135
+
"the re-claimed repair must again dedup while in flight"
1136
+
);
1137
+
}
1138
+
1139
+
#[test]
1140
+
fn noop_repair_uses_longer_cooldown() {
1141
+
let g = guard();
1142
+
let user = Uuid::from_u128(9);
1143
+
let t0 = Instant::now();
1144
+
1145
+
assert!(g.try_claim(user, t0));
1146
+
g.release(user, t0, REPAIR_NOOP_COOLDOWN);
1147
+
assert!(
1148
+
!g.try_claim(user, t0 + REPAIR_COOLDOWN + Duration::from_millis(1)),
1149
+
"after a no-op repair the standard cooldown must not re-admit a claim"
1150
+
);
1151
+
assert!(
1152
+
g.try_claim(user, t0 + REPAIR_NOOP_COOLDOWN + Duration::from_millis(1)),
1153
+
"after the longer no-op cooldown a fresh claim must be admitted"
1154
+
);
1155
+
}
1156
+
}
+14
-2
crates/tranquil-sync/src/repo.rs
+14
-2
crates/tranquil-sync/src/repo.rs
···
163
163
{
164
164
Ok(bytes) => bytes,
165
165
Err(e) => {
166
+
if ApiError::detail_is_repo_corruption(&format!("{e:#}")) {
167
+
tranquil_pds::repo_ops::schedule_repo_repair(&state, account.user_id);
168
+
}
166
169
error!("Failed to generate repo CAR: {}", e);
167
170
return ApiError::InternalError(None).into_response();
168
171
}
···
230
233
let blocks = match state.block_store.get_many(chunk).await {
231
234
Ok(b) => b,
232
235
Err(e) => {
236
+
if ApiError::detail_is_repo_corruption(&format!("{e:#}")) {
237
+
tranquil_pds::repo_ops::schedule_repo_repair(state, user_id);
238
+
}
233
239
error!("Block store error in get_repo_since: {:?}", e);
234
240
return ApiError::InternalError(Some("Failed to get blocks".into()))
235
241
.into_response();
···
301
307
Ok(None) => {
302
308
return ApiError::RecordNotFound.into_response();
303
309
}
304
-
Err(_) => {
310
+
Err(e) => {
311
+
if ApiError::detail_is_repo_corruption(&format!("{e:#}")) {
312
+
tranquil_pds::repo_ops::schedule_repo_repair(&state, account.user_id);
313
+
}
305
314
return ApiError::InternalError(Some("Failed to lookup record".into())).into_response();
306
315
}
307
316
};
···
312
321
}
313
322
};
314
323
let mut proof_blocks: BTreeMap<Cid, bytes::Bytes> = BTreeMap::new();
315
-
if mst.blocks_for_path(&key, &mut proof_blocks).await.is_err() {
324
+
if let Err(e) = mst.blocks_for_path(&key, &mut proof_blocks).await {
325
+
if ApiError::detail_is_repo_corruption(&format!("{e:#}")) {
326
+
tranquil_pds::repo_ops::schedule_repo_repair(&state, account.user_id);
327
+
}
316
328
return ApiError::InternalError(Some("Failed to build proof path".into())).into_response();
317
329
}
318
330
let header = match encode_car_header(&commit_cid) {