feat(jobs/ui): blobs_consistency add b3sum check
This commit is contained in:
@@ -393,11 +393,16 @@
|
|||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
// The `consistency_batch` coordinator gets a "Run deep" variant —
|
// Jobs that respect `?deep=true`:
|
||||||
// the only job that respects `?deep=true` today (via propagation to
|
// * `consistency_batch` — propagates deep to every child that
|
||||||
// `storage_consistency` once it lands).
|
// understands it
|
||||||
|
// * `blobs_consistency` — deep mode re-reads + re-hashes every
|
||||||
|
// blob for silent bit-rot detection (severity `data_loss`).
|
||||||
|
// Full read of storage; can take hours on big installs — the
|
||||||
|
// "Run" (normal) button on the same row does the cheap
|
||||||
|
// existence probes only.
|
||||||
function supportsDeep(name: string): boolean {
|
function supportsDeep(name: string): boolean {
|
||||||
return name === 'consistency_batch';
|
return name === 'consistency_batch' || name === 'blobs_consistency';
|
||||||
}
|
}
|
||||||
|
|
||||||
function isRunning(job: JobSummary): boolean {
|
function isRunning(job: JobSummary): boolean {
|
||||||
|
|||||||
@@ -357,10 +357,21 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
|
|||||||
// (3) blob_corrupted — DEEP MODE only. Read the
|
// (3) blob_corrupted — DEEP MODE only. Read the
|
||||||
// whole blob, recompute BLAKE3, compare to the hash
|
// whole blob, recompute BLAKE3, compare to the hash
|
||||||
// it's indexed under. Any mismatch = silent bit-rot.
|
// it's indexed under. Any mismatch = silent bit-rot.
|
||||||
|
//
|
||||||
|
// Finding fields:
|
||||||
|
// * `hash` — expected hash (the key the blob is
|
||||||
|
// indexed under in `storage.blobs`).
|
||||||
|
// * `computed_hash` — what BLAKE3 of the current
|
||||||
|
// bytes actually produces. Diagnostic: a
|
||||||
|
// one-bit flip vs a truncation vs a whole-file
|
||||||
|
// swap all leave distinctive signatures.
|
||||||
|
// `expected_hash` was NOT reused as a name to
|
||||||
|
// avoid mistaking it for "the hash we expect to
|
||||||
|
// see on disk (i.e. what will fix this)".
|
||||||
if args.deep {
|
if args.deep {
|
||||||
match verify_hash(self.backend.as_ref(), &row.hash).await {
|
match recompute_hash(self.backend.as_ref(), &row.hash).await {
|
||||||
Ok(true) => {}
|
Ok(computed_hash) if computed_hash == row.hash => {}
|
||||||
Ok(false) => {
|
Ok(computed_hash) => {
|
||||||
finding_count += 1;
|
finding_count += 1;
|
||||||
let affected = affected_files(self.pool.as_ref(), &row.hash).await;
|
let affected = affected_files(self.pool.as_ref(), &row.hash).await;
|
||||||
record_or_log(
|
record_or_log(
|
||||||
@@ -371,6 +382,7 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
|
|||||||
None,
|
None,
|
||||||
serde_json::json!({
|
serde_json::json!({
|
||||||
"hash": row.hash,
|
"hash": row.hash,
|
||||||
|
"computed_hash": computed_hash,
|
||||||
"size": row.size,
|
"size": row.size,
|
||||||
"ref_count": row.ref_count,
|
"ref_count": row.ref_count,
|
||||||
"affected_files": affected,
|
"affected_files": affected,
|
||||||
@@ -381,11 +393,11 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
|
|||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::warn!(
|
tracing::warn!(
|
||||||
target: "oxicloud::consistency",
|
target: "oxicloud::consistency",
|
||||||
event = "blobs_consistency.verify_hash_error",
|
event = "blobs_consistency.recompute_hash_error",
|
||||||
run_id = %store.run_id(),
|
run_id = %store.run_id(),
|
||||||
hash = %row.hash,
|
hash = %row.hash,
|
||||||
error = %e,
|
error = %e,
|
||||||
"verify_hash failed; not a corruption signal on its own"
|
"recompute_hash failed; not a corruption signal on its own"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -452,10 +464,19 @@ async fn affected_files(pool: &PgPool, hash: &str) -> Vec<String> {
|
|||||||
/// (bit-rot), `Err(_)` on any backend-side error (network blip,
|
/// (bit-rot), `Err(_)` on any backend-side error (network blip,
|
||||||
/// permission issue) — callers log-and-skip errors since a transient
|
/// permission issue) — callers log-and-skip errors since a transient
|
||||||
/// failure isn't a corruption signal.
|
/// failure isn't a corruption signal.
|
||||||
async fn verify_hash(
|
/// Deep-mode helper — read the blob from the backend and recompute
|
||||||
|
/// its BLAKE3 hash. Returns the recomputed hex string; callers
|
||||||
|
/// compare against the expected hash themselves. Returning the
|
||||||
|
/// actual hash (not just a bool) lets the finding surface WHAT the
|
||||||
|
/// bytes now hash to, which is diagnostic gold: a specific one-bit
|
||||||
|
/// flip has a very different signature from a chunk-boundary
|
||||||
|
/// corruption or a truncated read. `Err(_)` on backend-side error
|
||||||
|
/// (network blip, permission issue) — callers log-and-skip since
|
||||||
|
/// transient failure isn't a corruption signal.
|
||||||
|
async fn recompute_hash(
|
||||||
backend: &dyn BlobStorageBackend,
|
backend: &dyn BlobStorageBackend,
|
||||||
expected_hash: &str,
|
expected_hash: &str,
|
||||||
) -> Result<bool, crate::common::errors::DomainError> {
|
) -> Result<String, crate::common::errors::DomainError> {
|
||||||
use crate::common::errors::DomainError;
|
use crate::common::errors::DomainError;
|
||||||
use futures::StreamExt;
|
use futures::StreamExt;
|
||||||
|
|
||||||
@@ -469,6 +490,5 @@ async fn verify_hash(
|
|||||||
hasher.update(&bytes);
|
hasher.update(&bytes);
|
||||||
}
|
}
|
||||||
|
|
||||||
let actual = hasher.finalize().to_hex().to_string();
|
Ok(hasher.finalize().to_hex().to_string())
|
||||||
Ok(actual == expected_hash)
|
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user