diff --git a/src/application/ports/storage_ports.rs b/src/application/ports/storage_ports.rs index 697fad68..48efcb26 100644 --- a/src/application/ports/storage_ports.rs +++ b/src/application/ports/storage_ports.rs @@ -420,8 +420,13 @@ pub trait StorageUsagePort: Send + Sync + 'static { username: &str, ) -> Result; - /// Updates storage usage statistics for all users - async fn update_all_users_storage_usage(&self) -> Result<(), DomainError>; + /// Reconcile every user's cached `storage_used_bytes` against the + /// authoritative `SUM(drives.used_bytes)` over their personal + /// drives. Returns the number of user rows the sweep actually + /// corrected (`rows_affected` from the `UPDATE … IS DISTINCT + /// FROM …`) — used by the periodic-scheduler `JobHandler` to + /// populate `JobOutcome::Ok.count`. + async fn update_all_users_storage_usage(&self) -> Result; /// Checks if a user has enough quota for an additional upload. /// Returns Ok(()) if the upload is allowed, or Err(QuotaExceeded) with a @@ -455,7 +460,10 @@ pub trait StorageUsagePort: Send + Sync + 'static { /// `GROUP BY drive_id` aggregate, with an `IS DISTINCT FROM` /// guard so idle drives don't churn dead tuples. Runs from the /// same reconciliation ticker. - async fn update_all_drives_storage_usage(&self) -> Result<(), DomainError>; + /// + /// Returns the number of drive rows actually corrected — used by + /// the periodic-scheduler `JobHandler` to populate `JobOutcome::Ok.count`. + async fn update_all_drives_storage_usage(&self) -> Result; /// Pre-upload quota check on a single drive. /// diff --git a/src/application/services/storage_usage_service.rs b/src/application/services/storage_usage_service.rs index 24fadc87..5f6369df 100644 --- a/src/application/services/storage_usage_service.rs +++ b/src/application/services/storage_usage_service.rs @@ -3,7 +3,6 @@ use crate::common::errors::DomainError; use crate::infrastructure::repositories::pg::UserPgRepository; use sqlx::PgPool; use std::sync::Arc; -use tokio::task; use tracing::{debug, error, info}; use uuid::Uuid; @@ -569,43 +568,55 @@ impl StorageUsageService { Ok(()) } - /// Spawn a background task that periodically reconciles every user's cached - /// `storage_used_bytes` against the actual sum of their files. + /// Interval helper — same clamping (min 30 s) the retired + /// `start_reconciliation_job` applied, exposed so DI can pass a + /// sanitised `Duration` to `JobRegistry::register`. + pub fn reconciliation_interval(interval_secs: u64) -> std::time::Duration { + std::time::Duration::from_secs(interval_secs.max(30)) + } +} + +pub const STORAGE_RECONCILE_JOB_NAME: &str = "storage_reconcile"; + +use crate::infrastructure::scheduler::{JobHandler, JobOutcome}; +use async_trait::async_trait; + +#[async_trait] +impl JobHandler for StorageUsageService { + fn name(&self) -> &str { + STORAGE_RECONCILE_JOB_NAME + } + + /// Runs both reconciliation sweeps — drives first, then users — + /// and reports the total number of rows corrected. /// - /// `GET /api/auth/me` no longer recomputes usage on the request path; this - /// sweep (plus the per-upload update) keeps the cached value current for - /// all mutations — including deletes and trash — without any O(N) work on a - /// hot endpoint. Runs on the maintenance pool. The first sweep is deferred - /// by one interval so it never adds load at boot. - pub fn start_reconciliation_job(&self, interval_secs: u64) { - // Floor the interval so a misconfiguration can't busy-loop the sweep. - let interval_secs = interval_secs.max(30); - let service = self.clone(); - info!( - "Starting storage-usage reconciliation job (every {}s)", - interval_secs - ); - task::spawn(async move { - let mut ticker = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); - // tokio's first `tick()` fires immediately — consume it so the - // first real sweep happens one interval after startup. - ticker.tick().await; - loop { - ticker.tick().await; - debug!("Running scheduled storage-usage reconciliation"); - // Drive sweep runs FIRST: the user-side sweep below - // reads `drives.used_bytes` (the per-drive sum) to - // compute its own counter, so the drive counter must - // be honest first. Failure of one is logged but - // doesn't skip the other or the next tick. - if let Err(e) = service.update_all_drives_storage_usage().await { - error!("Scheduled drive storage-usage reconciliation failed: {}", e); - } - if let Err(e) = service.update_all_users_storage_usage().await { - error!("Scheduled user storage-usage reconciliation failed: {}", e); - } - } - }); + /// **Sweep order matters.** The user envelope is computed from + /// `SUM(drives.used_bytes)` over the caller's personal drives + /// (`docs/plan/drive.md` §7). Running the user sweep before the + /// drive sweep would read a stale per-drive counter and freeze + /// the user counter on the previous tick's numbers — invisible in + /// steady state, breaks any Hurl that trashes + sweeps within one + /// call. See memory note `bug_storage_sweep_order_drive_first`. + /// + /// Failure of one sub-sweep short-circuits the tick to `Err`; + /// operators see `outcome=err, cause=handler` in the scheduler + /// log and the individual sweep's own `error!` line above it. + async fn run(&self) -> JobOutcome { + let drives = match self.update_all_drives_storage_usage().await { + Ok(n) => n, + Err(e) => return JobOutcome::Err(format!("drive reconciliation failed: {e}")), + }; + let users = match self.update_all_users_storage_usage().await { + Ok(n) => n, + Err(e) => return JobOutcome::Err(format!("user reconciliation failed: {e}")), + }; + JobOutcome::ok_with( + drives + users, + serde_json::json!({ + "drives_corrected": drives, + "users_corrected": users, + }), + ) } } @@ -638,7 +649,7 @@ impl StorageUsagePort for StorageUsageService { /// /// External users are excluded — they carry no storage by construction /// (DB CHECK `users_external_no_storage`). - async fn update_all_users_storage_usage(&self) -> Result<(), DomainError> { + async fn update_all_users_storage_usage(&self) -> Result { debug!("Starting storage-usage reconciliation sweep"); // User envelope = SUM of `drives.used_bytes` across the user's @@ -681,11 +692,9 @@ impl StorageUsagePort for StorageUsageService { DomainError::internal_error("StorageUsage", format!("reconciliation sweep: {e}")) })?; - info!( - "Storage-usage reconciliation corrected {} user(s)", - result.rows_affected() - ); - Ok(()) + let corrected = result.rows_affected(); + info!("Storage-usage reconciliation corrected {} user(s)", corrected); + Ok(corrected) } async fn check_storage_quota( @@ -720,7 +729,7 @@ impl StorageUsagePort for StorageUsageService { /// FROM` guard to skip no-op rewrites so idle drives don't churn /// dead tuples. Runs from the same reconciliation ticker as the /// user sweep; failure is logged but doesn't stop the next tick. - async fn update_all_drives_storage_usage(&self) -> Result<(), DomainError> { + async fn update_all_drives_storage_usage(&self) -> Result { debug!("Starting drive storage-usage reconciliation sweep"); let result = sqlx::query( r#" @@ -744,9 +753,10 @@ impl StorageUsagePort for StorageUsageService { DomainError::internal_error("StorageUsage", format!("drive reconciliation sweep: {e}")) })?; + let corrected = result.rows_affected(); info!( "Drive storage-usage reconciliation corrected {} drive(s)", - result.rows_affected() + corrected ); // Unconditional invalidation — do NOT gate on // `rows_affected() > 0`. When a fire-and-forget delta has @@ -762,7 +772,7 @@ impl StorageUsagePort for StorageUsageService { // already right → zero rows → without unconditional // invalidation, cache stays at the previous step's value. self.invalidate_drive_lookup_caches(); - Ok(()) + Ok(corrected) } async fn check_drive_quota( diff --git a/src/common/di.rs b/src/common/di.rs index 564ccf39..4ed769bd 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1104,12 +1104,16 @@ impl AppServiceFactory { /// /// Uses the `maintenance_pool` for batch operations /// (`update_all_users_storage_usage`) to avoid starving user requests. - pub fn create_storage_usage_service( + /// + /// Note: this is `async` (unlike the pre-migration version) because + /// registration with `core.job_registry` requires an `await`. + pub async fn create_storage_usage_service( &self, _repos: &RepositoryServices, db_pool: &Arc, maintenance_pool: &Arc, drive_repo: Arc, + core: &CoreServices, ) -> Arc { let user_repository = Arc::new( crate::infrastructure::repositories::pg::UserPgRepository::new(db_pool.clone()), @@ -1131,11 +1135,27 @@ impl AppServiceFactory { as Arc, ), ); - // Keep cached storage usage fresh off the request path: GET /api/auth/me - // no longer recomputes the O(N) SUM per call; a periodic sweep does it - // instead (on the maintenance pool). - service.start_reconciliation_job(self.config.storage.usage_reconcile_secs); - tracing::info!("Storage usage service initialized"); + // Keep cached storage usage fresh off the request path: GET + // /api/auth/me no longer recomputes the O(N) SUM per call; a + // periodic sweep does it instead (on the maintenance pool). + // Registered with the periodic-job scheduler + // (`docs/plan/job-registry.md` Part 1); the retired + // `start_reconciliation_job` used to spawn its own interval loop. + let interval = StorageUsageService::reconciliation_interval( + self.config.storage.usage_reconcile_secs, + ); + if let Err(e) = core + .job_registry + .register(service.clone(), interval, None) + .await + { + tracing::error!("Failed to register storage_reconcile job: {e}"); + } else { + tracing::info!( + "Storage-usage reconciliation registered with scheduler (interval {}s)", + interval.as_secs() + ); + } service } @@ -1315,8 +1335,9 @@ impl AppServiceFactory { // 3c. Storage usage / quota service (needed by the instant-upload // path inside the application services, and re-exposed on AppState // for the handler-side quota checks of the byte-upload paths). - let storage_usage = - self.create_storage_usage_service(&repos, &pool, &maintenance_pool, drive_repo.clone()); + let storage_usage = self + .create_storage_usage_service(&repos, &pool, &maintenance_pool, drive_repo.clone(), &core) + .await; // 3d. Content index (embedded Tantivy) — opened before application // services so SearchService can hold the query port; the feeding