From 0e8b1fbbebcd6060fc54822c9874f75d2396924f Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Tue, 28 Jul 2026 21:13:09 +0200 Subject: [PATCH] refactor(job-registry): simplify the job registering* --- docs/plan/job-registry.md | 32 +++--- .../services/storage_usage_service.rs | 18 ++- src/common/di.rs | 108 +++++------------- src/infrastructure/scheduler/engine.rs | 34 +++++- src/infrastructure/scheduler/registry.rs | 94 +++++++++++---- src/infrastructure/services/dedup_service.rs | 23 ++++ .../services/grant_cleanup_service.rs | 16 ++- .../services/trash_cleanup_service.rs | 25 +++- 8 files changed, 224 insertions(+), 126 deletions(-) diff --git a/docs/plan/job-registry.md b/docs/plan/job-registry.md index 33dc21bd..19adc5b3 100644 --- a/docs/plan/job-registry.md +++ b/docs/plan/job-registry.md @@ -792,24 +792,24 @@ the old fields needs updating. ### Config surface — env vars -Canonical form for every job (Part 1 or Part 2 alike, AND for core -workers even though they don't register with the scheduler): +**No new convention.** Each service keeps its natural per-service +prefix (`OXICLOUD_GRANT_CLEANUP_*`, `OXICLOUD_STORAGE_USAGE_*`, …). +The `GET /api/admin/jobs` endpoint already gives operators a runtime +view of every registered job's interval, so grepping env-var prefixes +is no longer the primary discovery path. -``` -OXICLOUD_JOB__ENABLED -OXICLOUD_JOB__INTERVAL_HOURS # or _INTERVAL_SECS for sub-hour cadences -OXICLOUD_JOB__... # e.g. _GRACE_HOURS, _BATCH_SIZE -``` +Earlier drafts proposed a uniform `OXICLOUD_JOB__INTERVAL_*` +convention, with legacy names as warned aliases. Killed 2026-07-28 +(Ed): normalising only the interval knob while leaving domain-specific +tunables (`GRACE_DAYS`, `BATCH_SIZE`, …) at the natural prefix creates +*intra-service* prefix drift — worse than the *cross-service* drift it +was meant to solve. A service either goes fully to `OXICLOUD_JOB_*` +(disruptive rename of every knob) or fully stays at its native prefix +(no rename). We stay. -Core workers reuse this naming purely for uniform operator ergonomics -(e.g. `OXICLOUD_JOB_TREE_ETAG_FLUSH_INTERVAL_MS`) — the convention is -what operators grep for; whether the loop is scheduler-driven or a -dedicated `tokio::spawn` is an implementation detail they don't see. - -Existing per-service env vars keep working as **aliases** during -migration — `OXICLOUD_GRANT_CLEANUP_INTERVAL_HOURS` reads first, falls -back to `OXICLOUD_JOB_GRANT_CLEANUP_INTERVAL_HOURS`. Deprecated aliases -warn once on startup and stay recognised through one minor version. +The one real gap is **trash_cleanup has no env var today** (hardcoded +24h in DI). Adding `OXICLOUD_TRASH_CLEANUP_INTERVAL_HOURS` when we +need it uses the natural prefix — no new convention needed. ### Logging schema diff --git a/src/application/services/storage_usage_service.rs b/src/application/services/storage_usage_service.rs index 36ab045b..ac607a88 100644 --- a/src/application/services/storage_usage_service.rs +++ b/src/application/services/storage_usage_service.rs @@ -578,9 +578,25 @@ impl StorageUsageService { pub const STORAGE_RECONCILE_JOB_NAME: &str = "storage_reconcile"; -use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRunArgs}; +use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRegistry, JobRunArgs}; use async_trait::async_trait; +impl StorageUsageService { + /// Register self with the periodic-job scheduler and return the + /// same `Arc` for DI-style chaining. Scheduled tenant with + /// interval = `max(30s, interval_secs)`. See + /// `docs/plan/job-registry.md` Part 1. + pub async fn register( + self: Arc, + registry: &JobRegistry, + interval_secs: u64, + ) -> Arc { + let interval = Self::reconciliation_interval(interval_secs); + registry.register(self.clone(), Some(interval), None).await; + self + } +} + #[async_trait] impl JobHandler for StorageUsageService { fn name(&self) -> &str { diff --git a/src/common/di.rs b/src/common/di.rs index 83f5b1cf..fc046c71 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -878,27 +878,18 @@ impl AppServiceFactory { // (`docs/plan/job-registry.md` Part 1) instead of spawning its own // tokio interval loop. `SchedulerEngine::start` fires the actual // supervisor task at the end of `build_app_state`. - let cleanup_service = Arc::new(TrashCleanupService::new( + // Self-registering constructor chain — TrashCleanupService owns + // its interval + timeout shape; DI only supplies deps. + // `.register(®)` fires the uniform `job.registered` log line + // and panics on wiring error (duplicate name = boot must fail + // loud). See `docs/plan/job-registry.md` Part 1. + let _ = Arc::new(TrashCleanupService::new( trash_repo.clone(), core.dedup_service.clone(), 24, // Run cleanup every 24 hours - )); - let interval = cleanup_service.interval(); - if let Err(e) = core - .job_registry - .register(cleanup_service.clone(), Some(interval), None) - .await - { - // Duplicate registration is the only failure mode today and - // shouldn't happen in the normal DI flow. Log + continue so - // trash service still lands even if scheduling didn't. - tracing::error!("Failed to register trash_cleanup job with scheduler: {e}"); - } else { - tracing::info!( - "Trash cleanup registered with scheduler (interval {} h)", - interval.as_secs() / 3600 - ); - } + )) + .register(&core.job_registry) + .await; Some(service as Arc) } @@ -1125,7 +1116,12 @@ impl AppServiceFactory { // and invalidation would be a no-op observed by nobody — // this is the trap that regressed the used_bytes freshness // after perf commit `12dc648c`. - let service = Arc::new( + // 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). + // Self-registering constructor chain — StorageUsageService owns + // its interval-clamping via `Self::reconciliation_interval`. + Arc::new( crate::application::services::storage_usage_service::StorageUsageService::new( maintenance_pool.clone(), user_repository, @@ -1134,28 +1130,9 @@ impl AppServiceFactory { drive_repo 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). - // 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(), Some(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 + ) + .register(&core.job_registry, self.config.storage.usage_reconcile_secs) + .await } /// Starts the tree-ETag flush job (requires database). @@ -1279,22 +1256,13 @@ impl AppServiceFactory { // Register on-demand-only jobs whose owning service lives on // CoreServices. Dedup GC has NO periodic tick — trash cleanup's // sweep already runs GC as its tail step, so a periodic dedup - // schedule would double the work. Registering with `interval = - // None` keeps it admin-triggerable through the uniform scheduler - // surface (`POST /api/admin/jobs/dedup_gc/trigger`). - if let Err(e) = core - .job_registry - .register( - core.dedup_service.clone() as Arc, - None, // on-demand only - None, // no timeout - ) - .await - { - tracing::error!("Failed to register dedup_gc job with scheduler: {e}"); - } else { - tracing::info!("Dedup GC registered with scheduler (on-demand only)"); - } + // schedule would double the work. The `register()` method + // encapsulates the on-demand shape. + let _ = core + .dedup_service + .clone() + .register(&core.job_registry) + .await; // 2. Repository services (requires PgPool for all metadata) let repos = self.create_repository_services(&core, &pool); @@ -1484,32 +1452,18 @@ impl AppServiceFactory { self.start_content_index_job(&maintenance_pool, &core, content_index); grant_cleanup_service = if core.config.features.grant_cleanup.enabled { + // Self-registering constructor chain. Grant-cleanup owns + // its interval + on `?force=true` handling; DI only decides + // whether to instantiate at all (feature-gated). let svc = Arc::new( crate::infrastructure::services::grant_cleanup_service::GrantCleanupService::new( authorization.clone(), core.config.features.grant_cleanup.grace_days, core.config.features.grant_cleanup.interval_hours, ), - ); - // Registered with the periodic-job scheduler - // (`docs/plan/job-registry.md` Part 1); the retired - // `start_cleanup_job` used to spawn its own interval loop. - // Admin `?force=true` trigger still calls `svc.purge(Some(0))` - // directly — grace override doesn't fit the JobHandler shape. - let interval = svc.interval(); - if let Err(e) = core - .job_registry - .register(svc.clone(), Some(interval), None) - .await - { - tracing::error!("Failed to register grant_cleanup job: {e}"); - } else { - tracing::info!( - "Grant cleanup registered with scheduler (every {}h, grace = {}d)", - interval.as_secs() / 3600, - core.config.features.grant_cleanup.grace_days, - ); - } + ) + .register(&core.job_registry) + .await; Some(svc) } else { tracing::info!( diff --git a/src/infrastructure/scheduler/engine.rs b/src/infrastructure/scheduler/engine.rs index 35255d55..ee122104 100644 --- a/src/infrastructure/scheduler/engine.rs +++ b/src/infrastructure/scheduler/engine.rs @@ -264,6 +264,10 @@ fn translate_join( /// Distinct Ok/Err branches so the tracing macros pick up the fields at /// compile time — `tracing` doesn't expand conditional field lists. fn log_outcome(name: &str, outcome: &JobOutcome, cause: Option, elapsed_ms: u128) { + // Also render elapsed inline in the human-readable message so + // `tail -f` operators see the duration without waiting on a + // structured log renderer to project the `elapsed_ms` field. + let elapsed = format_elapsed(elapsed_ms); match outcome { JobOutcome::Ok { count, extra } => { tracing::info!( @@ -274,8 +278,10 @@ fn log_outcome(name: &str, outcome: &JobOutcome, cause: Option, elapse count = *count, elapsed_ms = elapsed_ms, extra = %extra, - "job {} ran", + "job {} ran in {} — count={}", name, + elapsed, + count, ); } JobOutcome::Err { message: msg } => { @@ -287,13 +293,31 @@ fn log_outcome(name: &str, outcome: &JobOutcome, cause: Option, elapse cause = %cause.unwrap_or(ErrCause::Handler), elapsed_ms = elapsed_ms, error = %msg, - "job {} failed", + "job {} failed after {} — {}", name, + elapsed, + msg, ); } } } +/// Human-friendly elapsed rendering — `12ms` / `340ms` / `1.4s` / +/// `12.3s` / `4m30s`. The structured `elapsed_ms` field still carries +/// the raw millisecond number for log aggregators. +fn format_elapsed(ms: u128) -> String { + if ms < 1000 { + format!("{}ms", ms) + } else if ms < 60_000 { + format!("{:.1}s", (ms as f64) / 1000.0) + } else { + let secs = ms / 1000; + let m = secs / 60; + let s = secs % 60; + format!("{}m{}s", m, s) + } +} + #[cfg(test)] mod tests { use super::*; @@ -361,8 +385,7 @@ mod tests { let registry = Arc::new(JobRegistry::new()); registry .register(handler, Some(Duration::from_millis(100)), None) - .await - .unwrap(); + .await; let entry = registry.get("overrun").await.unwrap(); // Kick off dispatch 1 in the background — it holds the permit @@ -402,8 +425,7 @@ mod tests { Some(Duration::from_millis(100)), Some(Duration::from_millis(50)), ) - .await - .unwrap(); + .await; let entry = registry.get("slow").await.unwrap(); dispatch("slow", entry.clone(), &JobRunArgs::default()).await; diff --git a/src/infrastructure/scheduler/registry.rs b/src/infrastructure/scheduler/registry.rs index 4470f047..03d9668c 100644 --- a/src/infrastructure/scheduler/registry.rs +++ b/src/infrastructure/scheduler/registry.rs @@ -69,10 +69,7 @@ impl JobRegistry { } } - /// Register a job. Returns an error if a job with the same name - /// is already registered — names are the primary identifier - /// everywhere (logs, admin URLs, env vars) and collisions would - /// hide bugs. + /// Register a job — production wiring path. /// /// - `interval = Some(dur)` → **scheduled**. The supervisor fires /// the job every `dur`, starting `now + dur`. Registration does @@ -83,11 +80,61 @@ impl JobRegistry { /// fires this job. Admin endpoint (or programmatic callers) can /// still invoke it via [`JobRegistry::trigger`] — the dispatch /// goes through the same panic/timeout/exclusivity gates. + /// + /// **Panics on error.** Registration failure (duplicate name today) + /// is a DI-wiring bug — the server must not start with a mis-wired + /// scheduler. Emits a uniform `job.registered` log line on success + /// so callers don't reinvent the log message at every site. + /// + /// For unit tests that need to assert the error path use + /// [`Self::try_register`] instead. pub async fn register( &self, handler: Arc, interval: Option, timeout: Option, + ) { + let name = handler.name().to_string(); + match self.try_register(handler, interval, timeout).await { + Ok(()) => { + let cadence = match interval { + Some(dur) => { + let secs = dur.as_secs(); + if secs % 3600 == 0 { + format!("every {} h", secs / 3600) + } else if secs % 60 == 0 { + format!("every {} min", secs / 60) + } else { + format!("every {} s", secs) + } + } + None => "on-demand".to_string(), + }; + tracing::info!( + target: "oxicloud::scheduler", + event = "job.registered", + job = %name, + cadence = %cadence, + "job {} registered ({})", + name, + cadence, + ); + } + Err(e) => panic!( + "JobRegistry::register({name}) failed — DI wiring bug: {e}" + ), + } + } + + /// Fallible sibling of [`Self::register`]. Returns `Err` on + /// duplicate-name instead of panicking, and does NOT emit the + /// `job.registered` log line — for unit tests that need to + /// assert failure without triggering the boot panic path. + pub async fn try_register( + &self, + handler: Arc, + interval: Option, + timeout: Option, ) -> Result<(), RegisterError> { let name = handler.name().to_string(); let mut guard = self.entries.write().await; @@ -286,11 +333,9 @@ mod tests { async fn register_and_pick_next() { let reg = JobRegistry::new(); reg.register(handler("job_a"), Some(Duration::from_secs(60)), None) - .await - .unwrap(); + .await; reg.register(handler("job_b"), Some(Duration::from_secs(10)), None) - .await - .unwrap(); + .await; let (next_name, _) = reg.pick_next().await.expect("expected a due job"); // job_b has the shorter interval → earlier next_run_at. @@ -300,16 +345,30 @@ mod tests { #[tokio::test] async fn duplicate_registration_rejected() { let reg = JobRegistry::new(); - reg.register(handler("job_x"), Some(Duration::from_secs(60)), None) + // Use the fallible `try_register` here so we can assert the + // Err path without triggering `register`'s boot-time panic. + reg.try_register(handler("job_x"), Some(Duration::from_secs(60)), None) .await .unwrap(); let err = reg - .register(handler("job_x"), Some(Duration::from_secs(60)), None) + .try_register(handler("job_x"), Some(Duration::from_secs(60)), None) .await .expect_err("duplicate name must be rejected"); assert!(matches!(err, RegisterError::DuplicateName(_))); } + #[tokio::test] + #[should_panic(expected = "DI wiring bug")] + async fn register_panics_on_duplicate() { + let reg = JobRegistry::new(); + reg.register(handler("job_dup"), Some(Duration::from_secs(60)), None) + .await; + // Second register with same name — boot panic. Anything doing + // this outside a #[should_panic] test is a mis-wired DI. + reg.register(handler("job_dup"), Some(Duration::from_secs(60)), None) + .await; + } + #[tokio::test] async fn empty_registry_picks_nothing() { let reg = JobRegistry::new(); @@ -320,11 +379,9 @@ mod tests { async fn snapshot_all_returns_every_entry() { let reg = JobRegistry::new(); reg.register(handler("a"), Some(Duration::from_secs(1)), None) - .await - .unwrap(); + .await; reg.register(handler("b"), Some(Duration::from_secs(1)), None) - .await - .unwrap(); + .await; let all = reg.snapshot_all().await; assert_eq!(all.len(), 2); } @@ -334,12 +391,9 @@ mod tests { let reg = JobRegistry::new(); // Scheduled job with a long interval. reg.register(handler("scheduled"), Some(Duration::from_secs(3600)), None) - .await - .unwrap(); + .await; // On-demand job — supervisor must never pick it. - reg.register(handler("on_demand"), None, None) - .await - .unwrap(); + reg.register(handler("on_demand"), None, None).await; let (next_name, _) = reg.pick_next().await.expect("scheduled job due"); assert_eq!( @@ -351,7 +405,7 @@ mod tests { #[tokio::test] async fn trigger_dispatches_on_demand_job() { let reg = Arc::new(JobRegistry::new()); - reg.register(handler("gc"), None, None).await.unwrap(); + reg.register(handler("gc"), None, None).await; let outcome = reg .trigger("gc", &JobRunArgs::default()) diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 77bbf417..6d3c7f7c 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -3127,6 +3127,29 @@ impl DedupPort for DedupService { /// and admin URLs (`POST /api/admin/jobs/dedup_gc/trigger`). pub const DEDUP_GC_JOB_NAME: &str = "dedup_gc"; +impl DedupService { + /// Register self with the periodic-job scheduler and return the + /// same `Arc` for DI-style chaining. **On-demand only** — + /// registered with `interval = None`. The periodic GC role + /// belongs to trash cleanup (which invokes `garbage_collect()` + /// inline as its tail step); a duplicate scheduled tick here + /// would double the reclamation work. Registration exists solely + /// to expose the admin trigger uniformly. + pub async fn register( + self: std::sync::Arc, + registry: &crate::infrastructure::scheduler::JobRegistry, + ) -> std::sync::Arc { + registry + .register( + self.clone() as std::sync::Arc, + None, // on-demand + None, // no timeout + ) + .await; + self + } +} + #[async_trait::async_trait] impl crate::infrastructure::scheduler::JobHandler for DedupService { fn name(&self) -> &str { diff --git a/src/infrastructure/services/grant_cleanup_service.rs b/src/infrastructure/services/grant_cleanup_service.rs index 7dcde2aa..32167b67 100644 --- a/src/infrastructure/services/grant_cleanup_service.rs +++ b/src/infrastructure/services/grant_cleanup_service.rs @@ -23,7 +23,7 @@ use tracing::{error, info}; use crate::application::ports::authorization_ports::AuthorizationEngine; use crate::common::errors::DomainError; -use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRunArgs}; +use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRegistry, JobRunArgs}; use crate::infrastructure::services::pg_acl_engine::PgAclEngine; use async_trait::async_trait; @@ -57,12 +57,22 @@ impl GrantCleanupService { self.grace_days } - /// Cadence exposed as `Duration` so DI passes a sanitised value - /// (post-`.max(1)`) to `JobRegistry::register`. + /// Cadence exposed as `Duration`. Internal helper used by + /// [`Self::register`]; kept `pub` for tests. pub fn interval(&self) -> Duration { Duration::from_secs(self.interval_hours * 3600) } + /// Register self with the periodic-job scheduler and return the + /// same `Arc` for DI-style chaining. Scheduled tenant with + /// interval = `self.interval()`, no timeout. See + /// `docs/plan/job-registry.md` Part 1. + pub async fn register(self: Arc, registry: &JobRegistry) -> Arc { + let interval = self.interval(); + registry.register(self.clone(), Some(interval), None).await; + self + } + /// Run one purge pass. /// /// `grace_override`: diff --git a/src/infrastructure/services/trash_cleanup_service.rs b/src/infrastructure/services/trash_cleanup_service.rs index 035ba992..8fa291b9 100644 --- a/src/infrastructure/services/trash_cleanup_service.rs +++ b/src/infrastructure/services/trash_cleanup_service.rs @@ -6,7 +6,7 @@ use tracing::{debug, error, info, instrument}; use crate::common::errors::Result; use crate::domain::repositories::trash_repository::TrashRepository; use crate::infrastructure::repositories::pg::trash_db_repository::TrashDbRepository; -use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRunArgs}; +use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRegistry, JobRunArgs}; use crate::infrastructure::services::dedup_service::DedupService; use async_trait::async_trait; @@ -43,12 +43,31 @@ impl TrashCleanupService { } } - /// Registered interval as a `Duration` — helper for DI wiring so - /// the composition root doesn't reinvent the `hours × 3600` cast. + /// Registered interval as a `Duration`. Internal helper used by + /// [`Self::register`]; kept `pub` in case a test wants to assert + /// the clamped value. pub fn interval(&self) -> Duration { Duration::from_secs(self.cleanup_interval_hours * 3600) } + /// Register self with the periodic-job scheduler and return the + /// same `Arc` for DI-style method chaining: + /// + /// ```ignore + /// let svc = Arc::new(TrashCleanupService::new(...)) + /// .register(&core.job_registry) + /// .await; + /// ``` + /// + /// Scheduled tenant — interval reads from + /// `self.cleanup_interval_hours`, no timeout. See + /// `docs/plan/job-registry.md` Part 1 §Contract. + pub async fn register(self: Arc, registry: &JobRegistry) -> Arc { + let interval = self.interval(); + registry.register(self.clone(), Some(interval), None).await; + self + } + /// Starts the periodic cleanup job #[instrument(skip(self))] pub async fn start_cleanup_job(&self) {