From d0f4e1ef5a85c1bb4861b62532a3057b4ae6f65b Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 27 Jul 2026 22:22:20 +0200 Subject: [PATCH] feat(job-registry): wire engine + wire trash cleaner --- src/common/di.rs | 69 +++++++++++++++-- .../services/trash_cleanup_service.rs | 75 +++++++++++++++++++ 2 files changed, 139 insertions(+), 5 deletions(-) diff --git a/src/common/di.rs b/src/common/di.rs index e6d6dc27..564ccf39 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -39,6 +39,7 @@ use crate::infrastructure::repositories::pg::{ FileBlobReadRepository, FileBlobWriteRepository, FileMetadataRepository, FolderDbRepository, TrashDbRepository, }; +use crate::infrastructure::scheduler::{JobRegistry, SchedulerEngine}; use crate::infrastructure::services::file_content_cache::{ FileContentCache, FileContentCacheConfig, }; @@ -429,6 +430,12 @@ impl AppServiceFactory { } let file_lifecycle = Arc::new(fls); + // Empty periodic-job registry; services register themselves + // downstream during their own creation. `SchedulerEngine::start` + // fires at the end of `build_app_state` once all registrations + // have landed. + let job_registry = Arc::new(JobRegistry::new()); + Ok(CoreServices { path_service, file_content_cache, @@ -441,6 +448,7 @@ impl AppServiceFactory { dedup_service, zip_service: None, // Placeholder - replaced after app services init config: self.config.clone(), + job_registry, }) } @@ -865,14 +873,32 @@ impl AppServiceFactory { // Initialize cleanup service (bulk-deletes expired items in 2 SQL // queries, then GCs zero-reference blobs — including chunks orphaned // by aborted streaming uploads). - let cleanup_service = TrashCleanupService::new( + // + // Registers with the periodic-job scheduler + // (`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( trash_repo.clone(), core.dedup_service.clone(), 24, // Run cleanup every 24 hours - ); - - cleanup_service.start_cleanup_job().await; - tracing::info!("Trash service initialized with daily cleanup schedule"); + )); + let interval = cleanup_service.interval(); + if let Err(e) = core + .job_registry + .register(cleanup_service.clone(), 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 + ); + } Some(service as Arc) } @@ -1805,6 +1831,11 @@ impl AppServiceFactory { 50_000, ), ), + // Populated below once every service has finished registering + // with `core.job_registry`. Starting the engine before all + // registrations land would race the first tick against + // late-registered jobs. + scheduler_engine: None, }; let email_bundle = build_email_sender(&self.config.smtp); app_state.email_sender = email_bundle.sender; @@ -2079,6 +2110,21 @@ impl AppServiceFactory { } } + // Start the periodic-job scheduler AFTER every native service has + // finished registering its jobs on `core.job_registry`. Starting + // it earlier would race the first tick against late registrations. + // See `docs/plan/job-registry.md` Part 1. + let registered = app_state.core.job_registry.len().await; + let engine = SchedulerEngine::start(app_state.core.job_registry.clone()); + app_state.scheduler_engine = Some(Arc::new(engine)); + tracing::info!( + target: "oxicloud::scheduler", + event = "scheduler.ready", + registered = registered, + "periodic scheduler ready ({} job(s) registered)", + registered + ); + Ok(app_state) } } @@ -2099,6 +2145,11 @@ pub struct CoreServices { pub dedup_service: Arc, pub zip_service: Option>, pub config: AppConfig, + /// Periodic-job scheduler registry. Services that satisfy the + /// migration criterion (`docs/plan/job-registry.md`) `register()` + /// themselves here during their creation; `SchedulerEngine::start` + /// spins up the supervisor loop at the end of `build_app_state`. + pub job_registry: Arc, } /// Container for repository services @@ -2305,6 +2356,14 @@ pub struct AppState { /// Authenticated callers bypass this limit. pub magic_link_send_per_ip_rate_limiter: Arc, + /// Handle to the periodic-job scheduler's supervisor task, spawned + /// at the end of `build_app_state` after every native service has + /// registered its jobs on `core.job_registry`. `Option` because + /// tests that assemble a partial `AppState` (no full DI) skip the + /// scheduler; production always populates it. Held here purely so + /// the tokio task isn't dropped — the supervisor loop runs off its + /// internal `JoinHandle`, not off this reference. + pub scheduler_engine: Option>, } // All AppState construction is done via struct literal in build_app_state(). diff --git a/src/infrastructure/services/trash_cleanup_service.rs b/src/infrastructure/services/trash_cleanup_service.rs index b4dde9cf..96d494ab 100644 --- a/src/infrastructure/services/trash_cleanup_service.rs +++ b/src/infrastructure/services/trash_cleanup_service.rs @@ -6,7 +6,9 @@ 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}; use crate::infrastructure::services::dedup_service::DedupService; +use async_trait::async_trait; /// Service for automatic cleanup of expired items in the trash. /// @@ -27,6 +29,8 @@ pub struct TrashCleanupService { } impl TrashCleanupService { + pub const JOB_NAME: &'static str = "trash_cleanup"; + pub fn new( trash_repository: Arc, dedup_service: Arc, @@ -39,6 +43,12 @@ impl TrashCleanupService { } } + /// Registered interval as a `Duration` — helper for DI wiring so + /// the composition root doesn't reinvent the `hours × 3600` cast. + pub fn interval(&self) -> Duration { + Duration::from_secs(self.cleanup_interval_hours * 3600) + } + /// Starts the periodic cleanup job #[instrument(skip(self))] pub async fn start_cleanup_job(&self) { @@ -107,4 +117,69 @@ impl TrashCleanupService { Ok(()) } + + /// One-shot execution used by both the legacy `start_cleanup_job` + /// timer AND the new `JobHandler::run` path. Returns the counts a + /// caller can turn into either a log line (legacy) or a `JobOutcome` + /// (scheduler). + async fn run_once(&self) -> Result { + let (files, folders) = self.trash_repository.delete_expired_bulk().await?; + // GC failure is non-fatal — the expiry itself succeeded. Report + // reclaimed bytes when possible; log + swallow otherwise. + let (gc_items, gc_bytes) = match self.dedup_service.garbage_collect().await { + Ok(pair) => pair, + Err(e) => { + error!("Trash cleanup GC failed: {:?}", e); + (0, 0) + } + }; + Ok(TrashCleanupStats { + files_purged: files, + folders_purged: folders, + gc_items, + gc_bytes, + }) + } +} + +/// Structured counters for one trash-cleanup sweep. Consumed by the +/// scheduler's `JobHandler::run` to shape `JobOutcome::Ok.extra`. +#[derive(Debug, Clone, Copy)] +struct TrashCleanupStats { + files_purged: u64, + folders_purged: u64, + gc_items: u64, + gc_bytes: u64, +} + +#[async_trait] +impl JobHandler for TrashCleanupService { + fn name(&self) -> &str { + Self::JOB_NAME + } + + /// Runs one bulk-delete-expired + GC sweep. `count` on the returned + /// `JobOutcome::Ok` is the total number of rows this tick removed + /// from the trash (files + folders); `extra` carries GC reclaim + /// counts so operators can see "how much did this actually free." + /// + /// Failure of the trash sweep itself → `Err`. GC failure alone is + /// non-fatal and stays logged only. + async fn run(&self) -> JobOutcome { + match self.run_once().await { + Ok(stats) => { + let removed = stats.files_purged + stats.folders_purged; + JobOutcome::ok_with( + removed, + serde_json::json!({ + "files_purged": stats.files_purged, + "folders_purged": stats.folders_purged, + "gc_items": stats.gc_items, + "gc_bytes": stats.gc_bytes, + }), + ) + } + Err(e) => JobOutcome::Err(format!("trash cleanup failed: {e}")), + } + } }