runledger_runtime/catalog/
sync.rs

1use runledger_postgres::DbPool;
2use runledger_postgres::jobs::{
3    JobDefinitionCatalogSyncMode, sync_catalog_job_definitions_exact_tx,
4    sync_catalog_job_definitions_tx,
5};
6
7use super::{
8    CatalogError, JobCatalog, JobCatalogExactSyncReport, JobCatalogSyncReport, JobCatalogSyncScope,
9};
10
11impl JobCatalog {
12    /// Upserts every catalog job into `job_definitions`.
13    ///
14    /// The catalog owns the synced definition fields: repeated syncs overwrite
15    /// `version`, `max_attempts`, `default_timeout_seconds`, and
16    /// `default_priority` for registered jobs. Effectively enabled catalog jobs
17    /// preserve an existing disabled row so operator pauses survive worker
18    /// restarts; effectively disabled catalog jobs explicitly write
19    /// `is_enabled = false`.
20    /// Safe to call repeatedly. Does not delete or disable definitions absent from the catalog.
21    /// Use [`Self::sync_definitions_exact`] with an explicit scope when removed
22    /// catalog entries should be disabled.
23    ///
24    /// Syncs that disable catalog jobs briefly lock `job_schedules` and
25    /// `job_definitions` so active schedule checks and definition disables are
26    /// evaluated against a stable write boundary.
27    ///
28    /// # Errors
29    /// Returns [`CatalogError`] when defaults are invalid or persistence fails.
30    pub async fn sync_definitions(
31        &self,
32        pool: &DbPool,
33    ) -> Result<JobCatalogSyncReport, CatalogError> {
34        self.validate_defaults()?;
35
36        let mut tx = pool.begin().await.map_err(|error| {
37            CatalogError::SyncFailure(Box::new(
38                runledger_postgres::Error::from_query_sqlx_with_context(
39                    "begin job catalog definition sync",
40                    error,
41                ),
42            ))
43        })?;
44
45        let definitions = self.catalog_definitions();
46        let mode = JobDefinitionCatalogSyncMode::PreserveExistingEnabledForEnabledDefinitions;
47        let report = sync_catalog_job_definitions_tx(&mut tx, &definitions, mode)
48            .await
49            .map_err(CatalogError::from_definition_catalog_sync_error)?;
50
51        tx.commit().await.map_err(CatalogError::CommitFailure)?;
52
53        Ok(JobCatalogSyncReport {
54            disabled_catalog_job_types: report.disabled_catalog_job_types,
55        })
56    }
57
58    /// Upserts catalog jobs, then disables enabled `job_definitions` rows in
59    /// `scope` whose job type is absent from the catalog.
60    ///
61    /// This is the stricter startup mode for applications that want the catalog
62    /// to be the active job-definition source of truth. It never deletes rows,
63    /// never operates outside the supplied owned job-type set, and rejects active
64    /// schedules that still reference a job type it would disable. Unlike
65    /// [`Self::sync_definitions`], exact sync restores catalog entries'
66    /// effective `is_enabled` value from catalog defaults and job-specific
67    /// overrides.
68    ///
69    /// Exact sync briefly locks `job_schedules` and `job_definitions` before it
70    /// checks active schedules or disables definitions, so it is heavier than the
71    /// additive sync path.
72    ///
73    /// # Errors
74    /// Returns [`CatalogError`] when the scope is invalid, the catalog is empty,
75    /// a catalog job is outside the scope, active schedules still reference
76    /// absent scoped jobs, or persistence fails.
77    pub async fn sync_definitions_exact(
78        &self,
79        pool: &DbPool,
80        scope: &JobCatalogSyncScope,
81    ) -> Result<JobCatalogExactSyncReport, CatalogError> {
82        self.validate_defaults()?;
83        self.validate_exact_sync_scope(scope)?;
84
85        let mut tx = pool.begin().await.map_err(|error| {
86            CatalogError::SyncFailure(Box::new(
87                runledger_postgres::Error::from_query_sqlx_with_context(
88                    "begin exact job catalog definition sync",
89                    error,
90                ),
91            ))
92        })?;
93
94        let scope_job_types = scope.job_types_for_storage();
95        let definitions = self.catalog_definitions();
96        let report = sync_catalog_job_definitions_exact_tx(&mut tx, &definitions, &scope_job_types)
97            .await
98            .map_err(CatalogError::from_definition_catalog_sync_error)?;
99
100        tx.commit().await.map_err(CatalogError::CommitFailure)?;
101
102        Ok(JobCatalogExactSyncReport {
103            disabled_absent_job_types: report.disabled_absent_job_types,
104            disabled_catalog_job_types: report.disabled_catalog_job_types,
105        })
106    }
107
108    fn catalog_definitions(&self) -> Vec<runledger_postgres::jobs::JobDefinitionUpsert<'static>> {
109        self.jobs
110            .values()
111            .map(|entry| self.materialize_definition(entry))
112            .collect()
113    }
114}
runledger_runtime/catalog/sync.rs

runledger_runtime/catalog/
sync.rs
