feat(recoverable-job): fix files_consistency to check blob chunk consistency
This commit is contained in:
@@ -144,10 +144,11 @@ impl RecoverableJobHandler for DrivesConsistencyCheck {
|
||||
// query. LEFT JOIN via correlated subquery gets us both
|
||||
// sides in one round-trip; the storage_reconcile sweep
|
||||
// uses the same shape.
|
||||
let rows: Vec<(Uuid, i64, i64)> = match sqlx::query_as(
|
||||
let rows: Vec<(Uuid, String, i64, i64)> = match sqlx::query_as(
|
||||
r#"
|
||||
SELECT
|
||||
d.id,
|
||||
d.name,
|
||||
d.used_bytes,
|
||||
COALESCE((
|
||||
SELECT SUM(size)::bigint
|
||||
@@ -189,7 +190,7 @@ impl RecoverableJobHandler for DrivesConsistencyCheck {
|
||||
// Per-row check: cached vs actual. This is the ONE check
|
||||
// in v1 — more per-row branches (quota inversion, kind vs
|
||||
// default_for_user, …) slot in here.
|
||||
for (drive_id, cached, actual) in &rows {
|
||||
for (drive_id, drive_name, cached, actual) in &rows {
|
||||
if *cached != *actual {
|
||||
drift_count += 1;
|
||||
// Persisted finding via the shared helper.
|
||||
@@ -203,9 +204,10 @@ impl RecoverableJobHandler for DrivesConsistencyCheck {
|
||||
"inconsistent",
|
||||
Some(*drive_id),
|
||||
serde_json::json!({
|
||||
"name": drive_name,
|
||||
"cached": cached,
|
||||
"actual": actual,
|
||||
"delta": cached - actual,
|
||||
"delta": cached - actual,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
@@ -213,7 +215,10 @@ impl RecoverableJobHandler for DrivesConsistencyCheck {
|
||||
}
|
||||
|
||||
// Advance cursor to the last row's id + checkpoint.
|
||||
let last_id = rows.last().map(|(id, _, _)| *id).expect("non-empty rows");
|
||||
let last_id = rows
|
||||
.last()
|
||||
.map(|(id, _, _, _)| *id)
|
||||
.expect("non-empty rows");
|
||||
cursor = Some(last_id);
|
||||
let batch_len = rows.len() as u64;
|
||||
if let Err(e) = store
|
||||
|
||||
@@ -95,6 +95,9 @@ impl FilesConsistencyCheck {
|
||||
#[derive(Debug, sqlx::FromRow)]
|
||||
struct FileRow {
|
||||
id: Uuid,
|
||||
/// File name (basename). Captured into finding `detail` so
|
||||
/// operators see a human identifier next to the UUID.
|
||||
name: String,
|
||||
folder_id: Option<Uuid>,
|
||||
is_trashed: bool,
|
||||
size: i64,
|
||||
@@ -102,9 +105,41 @@ struct FileRow {
|
||||
/// `None` when `folder_id IS NULL` (file at drive root) — the
|
||||
/// LEFT JOIN yields no parent row.
|
||||
parent_is_trashed: Option<bool>,
|
||||
/// `None` when the blob row is missing — the LEFT JOIN yields
|
||||
/// no `blobs` side. This IS the `missing_blob` signal.
|
||||
/// Parent folder's materialised `path` (post-D7 files carry no
|
||||
/// path themselves). `None` for root files.
|
||||
parent_path: Option<String>,
|
||||
/// Legacy whole-file blob row size (pre-CDC). `None` when the
|
||||
/// file was ingested via CDC (`chunk_manifests` path) OR when
|
||||
/// the blob is truly missing — disambiguated by `manifest_size`.
|
||||
blob_size: Option<i64>,
|
||||
/// CDC manifest total size. `Some` when the file was ingested
|
||||
/// via FastCDC (its bytes live as chunks referenced by
|
||||
/// `storage.chunk_manifests.chunk_hashes`, not as one
|
||||
/// `storage.blobs` row). `None` when there is no manifest for
|
||||
/// this hash.
|
||||
manifest_size: Option<i64>,
|
||||
/// Total chunks the manifest claims. `None` when the file is
|
||||
/// pre-CDC (whole-file blob path) or has no manifest.
|
||||
manifest_chunk_count: Option<i32>,
|
||||
/// Count of chunks referenced by the manifest that have NO
|
||||
/// matching row in `storage.blobs`. `None` when there's no
|
||||
/// manifest to check. `Some(n)` with `n > 0` means the manifest
|
||||
/// points at reaped chunks — a real data-loss condition, more
|
||||
/// precise than plain `missing_blob` (which only fires when the
|
||||
/// whole-file registry entry is absent). This is a DB-registry
|
||||
/// check; physical backend-existence checks belong in the
|
||||
/// future `storage_consistency` tenant.
|
||||
chunks_missing: Option<i64>,
|
||||
}
|
||||
|
||||
/// Build the file's display path from its folder's `path` and its
|
||||
/// own `name`. Root files just show the name. Trashed folder paths
|
||||
/// still work (ltree keeps them intact under `is_trashed`).
|
||||
fn display_path(folder_path: Option<&str>, name: &str) -> String {
|
||||
match folder_path {
|
||||
Some(p) if !p.is_empty() => format!("{p}/{name}"),
|
||||
_ => name.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -192,19 +227,56 @@ impl RecoverableJobHandler for FilesConsistencyCheck {
|
||||
// row). Left-joining the blob is what lets us detect
|
||||
// `missing_blob` — a matched row has `blob.size`
|
||||
// populated; a miss surfaces as NULL.
|
||||
// Three LEFT JOINs — the blob-existence check has to
|
||||
// handle BOTH storage paths OxiCloud uses:
|
||||
//
|
||||
// * `storage.chunk_manifests` (CDC / FastCDC) — the
|
||||
// dominant path for anything ingested after Apr 2026.
|
||||
// Whole-file hash lives here; actual bytes are chunks
|
||||
// referenced by `chunk_hashes[]`.
|
||||
// * `storage.blobs` (legacy pre-CDC whole-file blob) —
|
||||
// still supported via the read path's fallback for
|
||||
// pre-CDC uploads.
|
||||
//
|
||||
// A file is "missing_blob" ONLY when NEITHER row exists.
|
||||
// Deep chunk validation (every chunk in `chunk_hashes[]`
|
||||
// present in `storage.blobs`) is out of scope here — it
|
||||
// belongs in the future `storage_consistency` tenant that
|
||||
// walks the backend against the blob registry.
|
||||
// Correlated subquery `chunks_missing` runs per-row over
|
||||
// the manifest's chunk_hashes array. `hash` is indexed
|
||||
// (PRIMARY KEY on storage.blobs), so each `NOT EXISTS`
|
||||
// probe is O(log n). NULL (not zero) when the file is
|
||||
// pre-CDC or has no manifest — the LEFT JOIN result on
|
||||
// `m` is NULL and `unnest(NULL::text[])` yields zero rows.
|
||||
let rows: Vec<FileRow> = match sqlx::query_as(
|
||||
r#"
|
||||
SELECT
|
||||
f.id AS id,
|
||||
f.name AS name,
|
||||
f.folder_id AS folder_id,
|
||||
f.is_trashed AS is_trashed,
|
||||
f.size AS size,
|
||||
f.blob_hash AS blob_hash,
|
||||
parent.is_trashed AS parent_is_trashed,
|
||||
b.size AS blob_size
|
||||
parent.path AS parent_path,
|
||||
b.size AS blob_size,
|
||||
m.total_size AS manifest_size,
|
||||
m.chunk_count AS manifest_chunk_count,
|
||||
CASE WHEN m.chunk_hashes IS NULL THEN NULL
|
||||
ELSE (
|
||||
SELECT COUNT(*)::bigint
|
||||
FROM unnest(m.chunk_hashes) AS ch(hash)
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1 FROM storage.blobs bb
|
||||
WHERE bb.hash = ch.hash
|
||||
)
|
||||
)
|
||||
END AS chunks_missing
|
||||
FROM storage.files f
|
||||
LEFT JOIN storage.folders parent ON parent.id = f.folder_id
|
||||
LEFT JOIN storage.blobs b ON b.hash = f.blob_hash
|
||||
LEFT JOIN storage.folders parent ON parent.id = f.folder_id
|
||||
LEFT JOIN storage.blobs b ON b.hash = f.blob_hash
|
||||
LEFT JOIN storage.chunk_manifests m ON m.file_hash = f.blob_hash
|
||||
WHERE ($1::uuid IS NULL OR f.id > $1)
|
||||
ORDER BY f.id
|
||||
LIMIT $2
|
||||
@@ -236,6 +308,12 @@ impl RecoverableJobHandler for FilesConsistencyCheck {
|
||||
}
|
||||
|
||||
for row in &rows {
|
||||
// Human-readable path captured once per row and folded
|
||||
// into every finding on this row. `name` is the raw
|
||||
// basename (useful even when the parent is orphaned
|
||||
// and `parent_path` is None).
|
||||
let path = display_path(row.parent_path.as_deref(), &row.name);
|
||||
|
||||
// (1) parent_folder_trashed: live file under a
|
||||
// soft-deleted folder. Root files (`folder_id IS
|
||||
// NULL`) are exempt — `parent_is_trashed` is None
|
||||
@@ -249,16 +327,28 @@ impl RecoverableJobHandler for FilesConsistencyCheck {
|
||||
"inconsistent",
|
||||
Some(row.id),
|
||||
serde_json::json!({
|
||||
"name": row.name,
|
||||
"path": path,
|
||||
"folder_id": row.folder_id,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
// (2) missing_blob: `blob_hash` has no `storage.blobs`
|
||||
// row. Real data-loss indicator — reading the file
|
||||
// will fail.
|
||||
if row.blob_size.is_none() {
|
||||
// Content-bearing size for this file, in priority
|
||||
// order: CDC manifest (dominant path — every file
|
||||
// uploaded after Apr 2026), then legacy pre-CDC
|
||||
// whole-file blob. `None` = no registry entry on
|
||||
// either path → real `missing_blob`.
|
||||
let content_size = row.manifest_size.or(row.blob_size);
|
||||
|
||||
// (2) missing_blob: NEITHER the CDC manifest nor the
|
||||
// legacy blob row exists for this hash. Real data-loss
|
||||
// indicator — the read path checks manifest first and
|
||||
// falls back to blob; if both are missing, reading
|
||||
// the file will fail. NOT a false positive for CDC
|
||||
// files, because the manifest check catches them.
|
||||
if content_size.is_none() {
|
||||
finding_count += 1;
|
||||
record_or_log(
|
||||
store,
|
||||
@@ -267,19 +357,55 @@ impl RecoverableJobHandler for FilesConsistencyCheck {
|
||||
"data_loss",
|
||||
Some(row.id),
|
||||
serde_json::json!({
|
||||
"name": row.name,
|
||||
"path": path,
|
||||
"blob_hash": row.blob_hash,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
// No point checking size when the blob row is
|
||||
// gone — skip (3) for this row.
|
||||
// No point checking size when neither registry
|
||||
// entry exists — skip (3) for this row.
|
||||
continue;
|
||||
}
|
||||
|
||||
// (2b) chunk_missing: the file's CDC manifest exists
|
||||
// and points at N chunks, but K of them have no row
|
||||
// in `storage.blobs`. Real data-loss condition — the
|
||||
// read path will fail reassembly when it tries to
|
||||
// fetch a reaped chunk. Typically caused by a dedup
|
||||
// GC race (chunk reaped while a manifest still held
|
||||
// a reference) or partial pg_dump/restore that
|
||||
// dropped `storage.blobs` rows.
|
||||
if let Some(missing) = row.chunks_missing
|
||||
&& missing > 0
|
||||
{
|
||||
finding_count += 1;
|
||||
record_or_log(
|
||||
store,
|
||||
FILES_CONSISTENCY_JOB_NAME,
|
||||
"chunk_missing",
|
||||
"data_loss",
|
||||
Some(row.id),
|
||||
serde_json::json!({
|
||||
"name": row.name,
|
||||
"path": path,
|
||||
"blob_hash": row.blob_hash,
|
||||
"chunks_missing": missing,
|
||||
"chunks_total": row.manifest_chunk_count,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
// Deliberately DON'T `continue` — a
|
||||
// chunk_missing finding does not preclude a
|
||||
// size mismatch, and the two are independent
|
||||
// signals worth surfacing separately.
|
||||
}
|
||||
|
||||
// (3) blob_size_mismatch: denormalised size drifted
|
||||
// from the blob's real length. Cheap because we've
|
||||
// already loaded both.
|
||||
if let Some(bs) = row.blob_size
|
||||
// from the content-registry's authoritative size.
|
||||
// Prefers manifest.total_size when present (post-CDC
|
||||
// ingest path); falls back to blob.size (legacy).
|
||||
if let Some(bs) = content_size
|
||||
&& bs != row.size
|
||||
{
|
||||
finding_count += 1;
|
||||
@@ -290,10 +416,13 @@ impl RecoverableJobHandler for FilesConsistencyCheck {
|
||||
"inconsistent",
|
||||
Some(row.id),
|
||||
serde_json::json!({
|
||||
"name": row.name,
|
||||
"path": path,
|
||||
"blob_hash": row.blob_hash,
|
||||
"stored": row.size,
|
||||
"actual": bs,
|
||||
"delta": row.size - bs,
|
||||
"source": if row.manifest_size.is_some() { "manifest" } else { "blob" },
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -100,6 +100,9 @@ impl FoldersConsistencyCheck {
|
||||
#[derive(Debug, sqlx::FromRow)]
|
||||
struct FolderRow {
|
||||
id: Uuid,
|
||||
/// Folder basename — surfaced in finding `detail` so operators
|
||||
/// see a human identifier next to the UUID.
|
||||
name: String,
|
||||
parent_id: Option<Uuid>,
|
||||
is_trashed: bool,
|
||||
path: String,
|
||||
@@ -200,6 +203,7 @@ impl RecoverableJobHandler for FoldersConsistencyCheck {
|
||||
r#"
|
||||
SELECT
|
||||
f.id AS id,
|
||||
f.name AS name,
|
||||
f.parent_id AS parent_id,
|
||||
f.is_trashed AS is_trashed,
|
||||
f.path AS path,
|
||||
@@ -263,6 +267,8 @@ impl RecoverableJobHandler for FoldersConsistencyCheck {
|
||||
"inconsistent",
|
||||
Some(row.id),
|
||||
serde_json::json!({
|
||||
"name": row.name,
|
||||
"path": row.path,
|
||||
"parent_id": row.parent_id,
|
||||
}),
|
||||
)
|
||||
@@ -280,6 +286,7 @@ impl RecoverableJobHandler for FoldersConsistencyCheck {
|
||||
"inconsistent",
|
||||
Some(row.id),
|
||||
serde_json::json!({
|
||||
"name": row.name,
|
||||
"stored": row.path,
|
||||
"expected": row.expected_path,
|
||||
"parent_path": row.parent_path,
|
||||
@@ -301,6 +308,8 @@ impl RecoverableJobHandler for FoldersConsistencyCheck {
|
||||
"inconsistent",
|
||||
Some(row.id),
|
||||
serde_json::json!({
|
||||
"name": row.name,
|
||||
"path": row.path,
|
||||
"stored": row.lpath_text,
|
||||
"expected": row.expected_lpath_text,
|
||||
"parent_lpath": row.parent_lpath_text,
|
||||
|
||||
Reference in New Issue
Block a user