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

feat(storage): wire choice of storage

Browse Source
This commit is contained in:
Edouard Vanbelle
2026-08-01 13:54:00 +02:00
parent 2de71b6d9a
commit d7c19570a5
10 changed files with 574 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/services/backend_consistency_service.rs
+101 -12
View File
@@ -57,6 +57,12 @@ use crate::infrastructure::scheduler::{
pub const BACKEND_CONSISTENCY_JOB_NAME: &str = "backend_consistency";
/// Same `params` JSONB key `blobs_consistency` uses — kept identical
/// so operators grepping run rows see the same convention across
/// both storage-audit tenants.
pub const PROBED_STORAGE_PARAM: &str =
crate::infrastructure::services::blobs_consistency_service::PROBED_STORAGE_PARAM;
/// Batch size for backend enumeration + DB probe. 500 is enough to
/// amortise the DB round-trip while keeping the cancel-poll cadence
/// sub-second (each batch = one backend list + one DB probe + Rust
@@ -78,12 +84,32 @@ const _MAX_EXAMPLES: usize = 5;
pub struct BackendConsistencyCheck {
pool: Arc<PgPool>,
/// Default backend to enumerate when `args.storage` is `None` —
/// the live LIVE backend, injected at DI. `?storage=<name>`
/// swaps in a fresh backend for the named entry (via
/// [`build_entry_backend`]).
backend: Arc<dyn BlobStorageBackend>,
/// Snapshot of `AppConfig.storage_entries` for `?storage=<name>`
/// resolution. Same rule blobs_consistency uses.
storage_entries: Vec<crate::common::config::NamedStorageEntry>,
/// `OXICLOUD_STORAGE_PATH` fallback for Local entries with no
/// `_ROOT_DIR`. Same fallback rule as boot.
storage_path_fallback: std::path::PathBuf,
}
impl BackendConsistencyCheck {
pub fn new(pool: Arc<PgPool>, backend: Arc<dyn BlobStorageBackend>) -> Self {
Self { pool, backend }
pub fn new(
pool: Arc<PgPool>,
backend: Arc<dyn BlobStorageBackend>,
storage_entries: Vec<crate::common::config::NamedStorageEntry>,
storage_path_fallback: std::path::PathBuf,
) -> Self {
Self {
pool,
backend,
storage_entries,
storage_path_fallback,
}
}
pub async fn register_recoverable_job(
@@ -138,9 +164,76 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
async fn run_resumable(
&self,
store: &dyn JobStore,
_args: &JobRunArgs,
args: &JobRunArgs,
resume_cursor: Option<Vec<u8>>,
) -> RunOutcome {
// Resolve the backend to probe. Mirrors the shape
// `blobs_consistency` uses — Fresh + args.storage=Some stamps
// probed_storage into params; Resumed reads it back so a
// mid-audit restart re-uses the same target.
let is_fresh = resume_cursor.is_none();
let probed_storage: Option<String> = if is_fresh {
let name = args.storage.clone();
if let Some(n) = &name
&& let Err(e) = store.set_string_param(PROBED_STORAGE_PARAM, n).await
{
return RunOutcome::Failed {
message: format!("persist {PROBED_STORAGE_PARAM} to params: {e}"),
};
}
name
} else {
match store.get_string_param(PROBED_STORAGE_PARAM).await {
Ok(v) => v,
Err(e) => {
return RunOutcome::Failed {
message: format!("read {PROBED_STORAGE_PARAM} from params: {e}"),
};
}
}
};
let backend: Arc<dyn BlobStorageBackend> = match &probed_storage {
None => self.backend.clone(),
Some(name) => match self.storage_entries.iter().find(|e| &e.name == name) {
Some(entry) => crate::infrastructure::services::entry_backend::build_entry_backend(
entry,
&self.storage_path_fallback,
),
None => {
let available = if self.storage_entries.is_empty() {
"(none)".to_string()
} else {
self.storage_entries
.iter()
.map(|e| e.name.as_str())
.collect::<Vec<_>>()
.join(", ")
};
return RunOutcome::Failed {
message: format!(
"storage entry `{name}` not found in OXICLOUD_STORAGE_ENTRIES. \
Available: [{available}]"
),
};
}
},
};
if let Err(e) = backend.initialize().await {
return RunOutcome::Failed {
message: format!("probed backend init: {e}"),
};
}
if let Some(name) = &probed_storage {
tracing::info!(
target: "audit",
event = "backend_consistency.probe_scoped",
run_id = %store.run_id(),
probed_storage = %name,
"backend_consistency enumerating entry `{name}` (via ?storage=<name>) instead \
of live backend"
);
}
// Cursor = opaque backend continuation token, UTF-8-encoded.
// Each backend defines its own format (local = shard/hash,
// S3 = ListObjectsV2 continuation token, Azure = list
@@ -190,11 +283,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
// splits canonical blobs (checked for orphan) from
// "unknown" entries (sidecar files, foreign namespaces —
// emitted as informational notices).
let page = match self
.backend
.list_blob_hashes(cursor.clone(), BATCH_SIZE)
.await
{
let page = match backend.list_blob_hashes(cursor.clone(), BATCH_SIZE).await {
Ok(v) => v,
Err(e) => {
// Backend refuses / can't enumerate. First-batch
@@ -219,7 +308,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
"anomaly",
None,
serde_json::json!({
"backend": self.backend.backend_type(),
"backend": backend.backend_type(),
"error": format!("{e}"),
"note": "backend refused enumeration; no per-blob orphan probes attempted",
}),
@@ -229,7 +318,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
target: "oxicloud::consistency",
event = "backend_consistency.unenumerable",
run_id = %store.run_id(),
backend = self.backend.backend_type(),
backend = backend.backend_type(),
"backend refused enumeration (typical during migration or on backends without list support)"
);
return RunOutcome::Completed;
@@ -265,7 +354,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
serde_json::json!({
"path": unknown.path,
"mtime": unknown.mtime.map(|t| t.to_rfc3339()),
"backend": self.backend.backend_type(),
"backend": backend.backend_type(),
"note": "non-canonical file in blob namespace (sidecar / wrong extension); not managed by dedup",
}),
)
@@ -331,7 +420,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
serde_json::json!({
"hash": entry.hash,
"mtime": entry.mtime.map(|t| t.to_rfc3339()),
"backend": self.backend.backend_type(),
"backend": backend.backend_type(),
}),
)
.await;
src/infrastructure/services/blobs_consistency_service.rs
+110 -4
View File
@@ -47,6 +47,7 @@
//! pointing at reaped chunks) — already covered by
//! `files_consistency::chunk_missing`.
use std::path::PathBuf;
use std::sync::Arc;
use async_trait::async_trait;
@@ -54,13 +55,21 @@ use chrono::{DateTime, Duration, Utc};
use sqlx::PgPool;
use crate::application::ports::blob_storage_ports::BlobStorageBackend;
use crate::common::config::NamedStorageEntry;
use crate::infrastructure::scheduler::{
JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome,
RunStatus, record_or_log,
};
use crate::infrastructure::services::entry_backend::build_entry_backend;
pub const BLOBS_CONSISTENCY_JOB_NAME: &str = "blobs_consistency";
/// `params` JSONB key under which the entry name being probed is
/// stashed on a Fresh run (matches `TARGET_NAME_PARAM` on
/// `storage_migration`). Resumed runs re-read it so a paused audit
/// survives restart without the admin re-specifying the target.
pub const PROBED_STORAGE_PARAM: &str = "probed_storage";
/// Rows per batch. Blobs are numerous (millions on a busy install)
/// but per-row work is one indexed backend probe + one indexed SQL
/// ref-count query. 200 balances cancel-poll cadence against
@@ -82,12 +91,35 @@ const AFFECTED_FILES_SAMPLE: i64 = 5;
pub struct BlobsConsistencyCheck {
pool: Arc<PgPool>,
/// The default backend to probe when `args.storage` is `None` —
/// the currently-active LIVE backend, injected at DI time. Runs
/// with `?storage=<name>` build a fresh backend for the named
/// entry instead (via [`build_entry_backend`]).
backend: Arc<dyn BlobStorageBackend>,
/// Snapshot of `AppConfig.storage_entries` used to resolve
/// `args.storage` to a `NamedStorageEntry`. Empty for the
/// legacy zero-entries path — `?storage=<name>` runs then
/// fail-fast with a clear "no entries declared" message.
storage_entries: Vec<NamedStorageEntry>,
/// Ambient `AppConfig.storage_path` — used as the `root_dir`
/// fallback for a Local target entry with no `_ROOT_DIR`. Same
/// fallback rule the boot path uses.
storage_path_fallback: PathBuf,
}
impl BlobsConsistencyCheck {
pub fn new(pool: Arc<PgPool>, backend: Arc<dyn BlobStorageBackend>) -> Self {
Self { pool, backend }
pub fn new(
pool: Arc<PgPool>,
backend: Arc<dyn BlobStorageBackend>,
storage_entries: Vec<NamedStorageEntry>,
storage_path_fallback: PathBuf,
) -> Self {
Self {
pool,
backend,
storage_entries,
storage_path_fallback,
}
}
pub async fn register_recoverable_job(
@@ -147,6 +179,80 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
args: &JobRunArgs,
resume_cursor: Option<Vec<u8>>,
) -> RunOutcome {
// Resolve the backend to probe. Two paths, mirroring the
// Fresh/Resumed split the storage_migration handler uses:
//
// * Fresh + args.storage=Some — probe that named entry
// instead of the live backend. Stamp probed_storage in
// params so a mid-audit restart resumes against the same
// entry without re-input.
// * Fresh + args.storage=None — probe the live backend
// (today's default; audit of what the app is actually
// using).
// * Resumed — read probed_storage from params; None means
// the original run was against the live backend.
let is_fresh = resume_cursor.is_none();
let probed_storage: Option<String> = if is_fresh {
let name = args.storage.clone();
if let Some(n) = &name
&& let Err(e) = store.set_string_param(PROBED_STORAGE_PARAM, n).await
{
return RunOutcome::Failed {
message: format!(
"persist {PROBED_STORAGE_PARAM} to params: {e}"
),
};
}
name
} else {
match store.get_string_param(PROBED_STORAGE_PARAM).await {
Ok(v) => v,
Err(e) => {
return RunOutcome::Failed {
message: format!("read {PROBED_STORAGE_PARAM} from params: {e}"),
};
}
}
};
let backend: Arc<dyn BlobStorageBackend> = match &probed_storage {
None => self.backend.clone(),
Some(name) => match self.storage_entries.iter().find(|e| &e.name == name) {
Some(entry) => build_entry_backend(entry, &self.storage_path_fallback),
None => {
let available = if self.storage_entries.is_empty() {
"(none)".to_string()
} else {
self.storage_entries
.iter()
.map(|e| e.name.as_str())
.collect::<Vec<_>>()
.join(", ")
};
return RunOutcome::Failed {
message: format!(
"storage entry `{name}` not found in OXICLOUD_STORAGE_ENTRIES. \
Available: [{available}]"
),
};
}
},
};
if let Err(e) = backend.initialize().await {
return RunOutcome::Failed {
message: format!("probed backend init: {e}"),
};
}
if let Some(name) = &probed_storage {
tracing::info!(
target: "audit",
event = "blobs_consistency.probe_scoped",
run_id = %store.run_id(),
probed_storage = %name,
"blobs_consistency probing entry `{name}` (via ?storage=<name>) instead of \
live backend"
);
}
// Cursor = the last-visited `hash` string, UTF-8-encoded. On
// resume, we walk `WHERE hash > $cursor` in ASC order. First
// batch: NULL cursor → start from the smallest hash.
@@ -318,7 +424,7 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
// error (log + skip): a transient S3 network blip
// shouldn't produce a flood of false data_loss
// findings.
let exists = match self.backend.blob_exists(&row.hash).await {
let exists = match backend.blob_exists(&row.hash).await {
Ok(v) => v,
Err(e) => {
tracing::warn!(
@@ -369,7 +475,7 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
// avoid mistaking it for "the hash we expect to
// see on disk (i.e. what will fix this)".
if args.deep {
match recompute_hash(self.backend.as_ref(), &row.hash).await {
match recompute_hash(backend.as_ref(), &row.hash).await {
Ok(computed_hash) if computed_hash == row.hash => {}
Ok(computed_hash) => {
finding_count += 1;