cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(manifest_consistency): add missing derived_blob to repair

Browse Source
This commit is contained in:
Edouard Vanbelle
2026-09-02 21:58:38 +02:00
parent 569d3ec526
commit 8a63663209
1 changed files with 99 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/services/manifests_consistency_service.rs
+99 -40
View File
@@ -93,6 +93,30 @@ fn manifest_page_sql(registry: &BlobReferenceRegistry) -> String {
) )
} }
/// Repair statement targeting one manifest by `file_hash`. Uses the
/// SAME registry-derived expression as [`manifest_page_sql`] so
/// detection and repair agree on what "actual" means — any future
/// manifest-level ref source added to the registry flows into both
/// queries with no code change here.
///
/// The `<> (subquery)` guard makes the UPDATE a no-op when the value
/// is already correct — so this is idempotent under concurrent-repair
/// races AND under retry.
///
/// The subquery re-reads inside the same statement, so a concurrent
/// insert/delete between page fetch and this UPDATE can't leave a
/// stale value: PG's snapshot for the UPDATE sees the up-to-date row
/// counts.
fn manifest_repair_sql(registry: &BlobReferenceRegistry) -> String {
let expected = registry.ref_count_expr(RefLevel::Manifest, "m.file_hash");
format!(
"UPDATE storage.chunk_manifests m
SET ref_count = ({expected})::bigint
WHERE m.file_hash = $1
AND m.ref_count <> ({expected})::bigint"
)
}
pub struct ManifestsConsistencyCheck { pub struct ManifestsConsistencyCheck {
pool: Arc<PgPool>, pool: Arc<PgPool>,
/// Built once from the blob-reference registry so this recompute and /// Built once from the blob-reference registry so this recompute and
@@ -100,6 +124,24 @@ pub struct ManifestsConsistencyCheck {
/// identically. Assembled at construction rather than per page so the /// identically. Assembled at construction rather than per page so the
/// sweep runs a fixed statement. /// sweep runs a fixed statement.
page_sql: String, page_sql: String,
/// Repair statement — built from the SAME registry as `page_sql` so
/// detection and repair use identical formulas by construction. Any
/// future 4th manifest-level ref source added to the registry
/// automatically flows into both queries with no code change here.
///
/// Previously the repair query was inlined with the files-only
/// formula, which meant drift from `content_derived_blobs` or
/// `file_attached_blobs` would be DETECTED but NOT repaired even
/// under `?repair=true`. Operators who added those tables saw
/// findings that couldn't be cleared by the repair path — bug fixed
/// 2026-09-02.
///
/// The `?repair=true` gate on the trigger endpoint still stands as
/// the operator's explicit opt-in — this fix only widens what
/// repair CAN do when the operator chooses to run it. Discovery-
/// only remains the default so leaks in insert paths still surface
/// via findings between repair invocations.
repair_sql: String,
} }
impl ManifestsConsistencyCheck { impl ManifestsConsistencyCheck {
@@ -107,6 +149,7 @@ impl ManifestsConsistencyCheck {
Self { Self {
pool, pool,
page_sql: manifest_page_sql(&reference_registry), page_sql: manifest_page_sql(&reference_registry),
repair_sql: manifest_repair_sql(&reference_registry),
} }
} }
@@ -308,25 +351,17 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck {
} }
finding_count += 1; finding_count += 1;
let delta = row.actual_ref_count - row.ref_count as i64; let delta = row.actual_ref_count - row.ref_count as i64;
record_or_log( let detail = serde_json::json!({
store, "file_hash": row.file_hash,
MANIFESTS_CONSISTENCY_JOB_NAME, "stored": row.ref_count,
"manifest_refcount_mismatch", "actual": row.actual_ref_count,
"inconsistent", "delta": delta,
None, // a hash isn't a UUID; the identifier lives in detail "total_size": row.total_size,
serde_json::json!({ "chunk_count": row.chunk_count,
"file_hash": row.file_hash, // Under-count is the dangerous direction: GC reaps
"stored": row.ref_count, // a manifest whose content is still reachable.
"actual": row.actual_ref_count, "reap_risk": delta > 0,
"delta": delta, });
"total_size": row.total_size,
"chunk_count": row.chunk_count,
// Under-count is the dangerous direction: GC reaps a
// manifest whose content is still reachable.
"reap_risk": delta > 0,
}),
)
.await;
// Repair pass — content-safe corrective UPDATE. The // Repair pass — content-safe corrective UPDATE. The
// stored counter is set to what the auditor formula // stored counter is set to what the auditor formula
@@ -338,22 +373,30 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck {
// The `<> (subquery)` guard makes the UPDATE a no-op // The `<> (subquery)` guard makes the UPDATE a no-op
// if the value is already correct, so this is // if the value is already correct, so this is
// idempotent under retry. // idempotent under retry.
if repair { //
match sqlx::query( // `self.repair_sql` is built once at construction from
"UPDATE storage.chunk_manifests m \ // the same `BlobReferenceRegistry` as the page query —
SET ref_count = ( \ // detection and repair use identical formulas by
SELECT COUNT(*) FROM storage.files \ // construction. See `manifest_repair_sql` for the SQL.
WHERE blob_hash = m.file_hash \ //
) \ // Attempt repair FIRST, then record the finding with
WHERE m.file_hash = $1 \ // severity/kind reflecting the final state:
AND m.ref_count <> ( \ // * repair succeeded → severity "info", kind "manifest_refcount_repaired"
SELECT COUNT(*) FROM storage.files \ // * repair no-op → severity "info", kind "manifest_refcount_resolved"
WHERE blob_hash = m.file_hash \ // * repair failed → severity "inconsistent", kind "manifest_refcount_mismatch"
)", // * no repair requested → severity "inconsistent", kind "manifest_refcount_mismatch"
) //
.bind(&row.file_hash) // Parallels the WARN-then-INFO sequence in logs: an
.execute(self.pool.as_ref()) // unresolved drift raises attention ("inconsistent"),
.await // a repaired one records the fix at info level without
// inflating the "needs action" tally the outcome UI
// shows. The detail JSON still carries `stored/actual/
// delta` so the audit trail is complete either way.
let (kind, severity) = if repair {
match sqlx::query(&self.repair_sql)
.bind(&row.file_hash)
.execute(self.pool.as_ref())
.await
{ {
Ok(res) if res.rows_affected() > 0 => { Ok(res) if res.rows_affected() > 0 => {
repaired_count += 1; repaired_count += 1;
@@ -366,12 +409,15 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck {
actual = row.actual_ref_count, actual = row.actual_ref_count,
"🩹 manifest ref_count repaired" "🩹 manifest ref_count repaired"
); );
("manifest_refcount_repaired", "info")
} }
Ok(_) => { Ok(_) => {
// Row not touched — either another concurrent // Row not touched — either another
// repair fixed it first, or the drift healed // concurrent repair fixed it first, or the
// itself between page fetch and UPDATE. // drift healed itself between page fetch
// Silent no-op. // and UPDATE. Either way, current state
// is correct — record as info.
("manifest_refcount_resolved", "info")
} }
Err(e) => { Err(e) => {
tracing::warn!( tracing::warn!(
@@ -382,9 +428,22 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck {
error = %e, error = %e,
"manifest ref_count repair UPDATE failed — finding stays" "manifest ref_count repair UPDATE failed — finding stays"
); );
("manifest_refcount_mismatch", "inconsistent")
} }
} }
} } else {
("manifest_refcount_mismatch", "inconsistent")
};
record_or_log(
store,
MANIFESTS_CONSISTENCY_JOB_NAME,
kind,
severity,
None, // a hash isn't a UUID; the identifier lives in detail
detail,
)
.await;
} }
// Advance cursor + checkpoint. // Advance cursor + checkpoint.