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

refactor(job-registry): simplify the job registering*

Browse Source
This commit is contained in:
Edouard Vanbelle
2026-07-28 21:13:09 +02:00
parent 4d64603233
commit 0e8b1fbbeb
8 changed files with 224 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plan/job-registry.md
+16 -16
View File
@@ -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_<NAME>_ENABLED
OXICLOUD_JOB_<NAME>_INTERVAL_HOURS # or _INTERVAL_SECS for sub-hour cadences
OXICLOUD_JOB_<NAME>_<CUSTOM>... # e.g. _GRACE_HOURS, _BATCH_SIZE
```
Earlier drafts proposed a uniform `OXICLOUD_JOB_<NAME>_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
src/application/services/storage_usage_service.rs
+17 -1
View File
@@ -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<Self>` 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<Self>,
registry: &JobRegistry,
interval_secs: u64,
) -> Arc<Self> {
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 {
src/common/di.rs
+31 -77
View File
@@ -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(&reg)` 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<TrashService>)
}
@@ -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<dyn crate::domain::repositories::drive_repository::DriveRepository>,
),
);
// 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<dyn crate::infrastructure::scheduler::JobHandler>,
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!(
src/infrastructure/scheduler/engine.rs
+28 -6
View File
@@ -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<ErrCause>, 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<ErrCause>, 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<ErrCause>, 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;
src/infrastructure/scheduler/registry.rs
+74 -20
View File
@@ -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<dyn JobHandler>,
interval: Option<Duration>,
timeout: Option<Duration>,
) {
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<dyn JobHandler>,
interval: Option<Duration>,
timeout: Option<Duration>,
) -> 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())
src/infrastructure/services/dedup_service.rs
+23
View File
@@ -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<Self>` 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<Self>,
registry: &crate::infrastructure::scheduler::JobRegistry,
) -> std::sync::Arc<Self> {
registry
.register(
self.clone() as std::sync::Arc<dyn crate::infrastructure::scheduler::JobHandler>,
None, // on-demand
None, // no timeout
)
.await;
self
}
}
#[async_trait::async_trait]
impl crate::infrastructure::scheduler::JobHandler for DedupService {
fn name(&self) -> &str {
src/infrastructure/services/grant_cleanup_service.rs
+13 -3
View File
@@ -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<Self>` 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<Self>, registry: &JobRegistry) -> Arc<Self> {
let interval = self.interval();
registry.register(self.clone(), Some(interval), None).await;
self
}
/// Run one purge pass.
///
/// `grace_override`:
src/infrastructure/services/trash_cleanup_service.rs
+22 -3
View File
@@ -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<Self>` 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<Self>, registry: &JobRegistry) -> Arc<Self> {
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) {