Add CronNextRun struct and table (#36026)

nipunn1313 · Convex, Inc. · commit 74616bbd0293 · 2025-04-01T21:15:00.000Z
Add CronNextRun struct and table to split out some data from CronJob struct.
Double write to them. Goal is to decouple the cron job executor from the push
process to avoid OCCs.

To do this migration correctly requires a ton of pushes

1. Add table + Double write
2. Run a migration to copy into second table.
3. Read from new table (but keep writing to old table)
4. Stop writing to old table

GitOrigin-RevId: 10d4a7769e10a7df5ba9d11b1e6d2da8731e58e4
diff --git a/crates/model/src/cron_jobs/mod.rs b/crates/model/src/cron_jobs/mod.rs
@@ -30,6 +30,7 @@ use sync_types::CanonicalizedModulePath;
 use value::{
     heap_size::WithHeapSize,
     ConvexValue,
+    DeveloperDocumentId,
     FieldPath,
     ResolvedDocumentId,
     TableName,
@@ -46,6 +47,7 @@ use crate::{
             CronJobLogLines,
             CronJobState,
             CronJobStatus,
+            CronNextRun,
             CronSpec,
         },
     },
@@ -85,6 +87,21 @@ pub static CRON_JOB_LOGS_NAME_FIELD: LazyLock<FieldPath> =
 static CRON_JOB_LOGS_TS_FIELD: LazyLock<FieldPath> =
     LazyLock::new(|| "ts".parse().expect("invalid ts field"));
 
+pub static CRON_NEXT_RUN_TABLE: LazyLock<TableName> = LazyLock::new(|| {
+    "_cron_next_run"
+        .parse()
+        .expect("_cron_next_run is not a valid system table name")
+});
+
+pub static CRON_NEXT_RUN_INDEX_BY_NEXT_TS: LazyLock<IndexName> =
+    LazyLock::new(|| system_index(&CRON_NEXT_RUN_TABLE, "by_next_ts"));
+pub static CRON_NEXT_RUN_INDEX_BY_CRON_JOB_ID: LazyLock<IndexName> =
+    LazyLock::new(|| system_index(&CRON_NEXT_RUN_TABLE, "by_cron_job_id"));
+static CRON_NEXT_RUN_NEXT_TS_FIELD: LazyLock<FieldPath> =
+    LazyLock::new(|| "nextTs".parse().expect("invalid nextTs field"));
+static CRON_NEXT_RUN_CRON_JOB_ID_FIELD: LazyLock<FieldPath> =
+    LazyLock::new(|| "cronJobId".parse().expect("invalid cronJobId field"));
+
 pub struct CronJobsTable;
 impl SystemTable for CronJobsTable {
     fn table_name(&self) -> &'static TableName {
@@ -134,6 +151,34 @@ impl SystemTable for CronJobLogsTable {
     }
 }
 
+pub struct CronNextRunTable;
+impl SystemTable for CronNextRunTable {
+    fn table_name(&self) -> &'static TableName {
+        &CRON_NEXT_RUN_TABLE
+    }
+
+    fn indexes(&self) -> Vec<SystemIndex> {
+        vec![
+            SystemIndex {
+                name: CRON_NEXT_RUN_INDEX_BY_NEXT_TS.clone(),
+                fields: vec![CRON_NEXT_RUN_NEXT_TS_FIELD.clone()]
+                    .try_into()
+                    .unwrap(),
+            },
+            SystemIndex {
+                name: CRON_NEXT_RUN_INDEX_BY_CRON_JOB_ID.clone(),
+                fields: vec![CRON_NEXT_RUN_CRON_JOB_ID_FIELD.clone()]
+                    .try_into()
+                    .unwrap(),
+            },
+        ]
+    }
+
+    fn validate_document(&self, document: ResolvedDocument) -> anyhow::Result<()> {
+        ParseDocument::<CronNextRun>::parse(document).map(|_| ())
+    }
+}
+
 const MAX_LOGS_PER_CRON: usize = 5;
 
 pub struct CronModel<'a, RT: Runtime> {
@@ -199,19 +244,51 @@ impl<'a, RT: Runtime> CronModel<'a, RT> {
         cron_spec: CronSpec,
     ) -> anyhow::Result<()> {
         let now = self.runtime().generate_timestamp()?;
+        let next_ts = compute_next_ts(&cron_spec, None, now)?;
         let cron = CronJob {
             name,
-            next_ts: compute_next_ts(&cron_spec, None, now)?,
+            next_ts,
             cron_spec,
             state: CronJobState::Pending,
             prev_ts: None,
         };
-        SystemMetadataModel::new(self.tx, self.component.into())
+
+        let cron_job_id = SystemMetadataModel::new(self.tx, self.component.into())
             .insert(&CRON_JOBS_TABLE, cron.try_into()?)
+            .await?
+            .developer_id;
+
+        let next_run = CronNextRun {
+            cron_job_id,
+            state: CronJobState::Pending,
+            prev_ts: None,
+            next_ts,
+        };
+
+        SystemMetadataModel::new(self.tx, self.component.into())
+            .insert(&CRON_NEXT_RUN_TABLE, next_run.try_into()?)
             .await?;
+
         Ok(())
     }
 
+    async fn next_run(
+        &mut self,
+        cron_job_id: DeveloperDocumentId,
+    ) -> anyhow::Result<Option<ParsedDocument<CronNextRun>>> {
+        let query = Query::index_range(IndexRange {
+            index_name: CRON_NEXT_RUN_INDEX_BY_CRON_JOB_ID.clone(),
+            range: vec![IndexRangeExpression::Eq(
+                CRON_NEXT_RUN_CRON_JOB_ID_FIELD.clone(),
+                ConvexValue::from(cron_job_id).into(),
+            )],
+            order: Order::Asc,
+        });
+        let mut query_stream = ResolvedQuery::new(self.tx, self.component.into(), query)?;
+        let next_run = query_stream.expect_at_most_one(self.tx).await?;
+        next_run.map(|v| v.parse()).transpose()
+    }
+
     pub async fn update(
         &mut self,
         cron_job: ParsedDocument<CronJob>,
@@ -228,9 +305,16 @@ impl<'a, RT: Runtime> CronModel<'a, RT> {
     }
 
     pub async fn delete(&mut self, cron_job: ParsedDocument<CronJob>) -> anyhow::Result<()> {
+        let id = cron_job.id();
         SystemMetadataModel::new(self.tx, self.component.into())
-            .delete(cron_job.clone().id())
+            .delete(id)
             .await?;
+        let next_run = self.next_run(id.developer_id).await?;
+        if let Some(next_run) = next_run {
+            SystemMetadataModel::new(self.tx, self.component.into())
+                .delete(next_run.id())
+                .await?;
+        }
         self.apply_job_log_retention(cron_job.name.clone(), 0)
             .await?;
         Ok(())
@@ -247,8 +331,25 @@ impl<'a, RT: Runtime> CronModel<'a, RT> {
             .namespace(self.component.into())
             .tablet_matches_name(id.tablet_id, &CRON_JOBS_TABLE));
         SystemMetadataModel::new(self.tx, self.component.into())
-            .replace(id, job.try_into()?)
+            .replace(id, job.clone().try_into()?)
             .await?;
+
+        let next_run = CronNextRun {
+            cron_job_id: id.developer_id,
+            state: job.state,
+            prev_ts: None,
+            next_ts: job.next_ts,
+        };
+        let existing_next_run = self.next_run(id.developer_id).await?;
+        if let Some(existing_next_run) = existing_next_run {
+            SystemMetadataModel::new(self.tx, self.component.into())
+                .replace(existing_next_run.id(), next_run.try_into()?)
+                .await?;
+        } else {
+            SystemMetadataModel::new(self.tx, self.component.into())
+                .insert(&CRON_NEXT_RUN_TABLE, next_run.try_into()?)
+                .await?;
+        }
         Ok(())
     }
 
diff --git a/crates/model/src/cron_jobs/types.rs b/crates/model/src/cron_jobs/types.rs
@@ -27,6 +27,7 @@ use sync_types::{
 use value::{
     codegen_convex_serialization,
     heap_size::HeapSize,
+    id_v6::DeveloperDocumentId,
     json_deserialize,
     obj,
     ConvexArray,
@@ -1084,3 +1085,48 @@ mod tests {
         assert_roundtrips::<_, CronJob>(cron_job_obj);
     }
 }
+
+#[derive(Clone, Debug, PartialEq)]
+#[cfg_attr(any(test, feature = "testing"), derive(proptest_derive::Arbitrary))]
+pub struct CronNextRun {
+    // Internally tracked metadata to execute the current run of the cron
+    pub cron_job_id: DeveloperDocumentId,
+    pub state: CronJobState,
+    pub prev_ts: Option<Timestamp>,
+    pub next_ts: Timestamp,
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SerializedCronNextRun {
+    cron_job_id: String,
+    state: CronJobState,
+    prev_ts: Option<i64>,
+    next_ts: i64,
+}
+
+impl From<CronNextRun> for SerializedCronNextRun {
+    fn from(run: CronNextRun) -> Self {
+        Self {
+            state: run.state,
+            prev_ts: run.prev_ts.map(|ts| ts.into()),
+            next_ts: run.next_ts.into(),
+            cron_job_id: run.cron_job_id.encode(),
+        }
+    }
+}
+
+impl TryFrom<SerializedCronNextRun> for CronNextRun {
+    type Error = anyhow::Error;
+
+    fn try_from(value: SerializedCronNextRun) -> anyhow::Result<Self, Self::Error> {
+        Ok(Self {
+            cron_job_id: DeveloperDocumentId::decode(&value.cron_job_id)?,
+            state: value.state,
+            prev_ts: value.prev_ts.map(|ts| ts.try_into()).transpose()?,
+            next_ts: value.next_ts.try_into()?,
+        })
+    }
+}
+
+codegen_convex_serialization!(CronNextRun, SerializedCronNextRun);
diff --git a/crates/model/src/lib.rs b/crates/model/src/lib.rs
@@ -80,6 +80,9 @@ use cron_jobs::{
     CRON_JOBS_TABLE,
     CRON_JOB_LOGS_INDEX_BY_NAME_TS,
     CRON_JOB_LOGS_TABLE,
+    CRON_NEXT_RUN_INDEX_BY_CRON_JOB_ID,
+    CRON_NEXT_RUN_INDEX_BY_NEXT_TS,
+    CRON_NEXT_RUN_TABLE,
 };
 pub use database::defaults::{
     SystemIndex,
@@ -179,6 +182,7 @@ use crate::{
     cron_jobs::{
         CronJobLogsTable,
         CronJobsTable,
+        CronNextRunTable,
     },
     deployment_audit_log::{
         DeploymentAuditLogsTable,
@@ -252,9 +256,10 @@ enum DefaultTableNumber {
     ComponentsTable = 32,
     FunctionHandlesTable = 33,
     CanonicalUrls = 34,
+    CronNextRun = 35,
     // Keep this number and your user name up to date. The number makes it easy to know
     // what to use next. The username on the same line detects merge conflicts
-    // Next Number - 35 - lee
+    // Next Number - 36 - nipunn
 }
 
 impl From<DefaultTableNumber> for TableNumber {
@@ -295,6 +300,7 @@ impl From<DefaultTableNumber> for &'static dyn SystemTable {
             DefaultTableNumber::ComponentsTable => &ComponentsTable,
             DefaultTableNumber::FunctionHandlesTable => &FunctionHandlesTable,
             DefaultTableNumber::CanonicalUrls => &CanonicalUrlsTable,
+            DefaultTableNumber::CronNextRun => &CronNextRunTable,
         }
     }
 }
@@ -326,6 +332,8 @@ static SYSTEM_INDEXES_WITHOUT_CREATION_TIME: LazyLock<BTreeSet<IndexName>> = Laz
         CRON_JOBS_INDEX_BY_NAME.clone(),
         CRON_JOBS_INDEX_BY_NEXT_TS.clone(),
         CRON_JOB_LOGS_INDEX_BY_NAME_TS.clone(),
+        CRON_NEXT_RUN_INDEX_BY_NEXT_TS.clone(),
+        CRON_NEXT_RUN_INDEX_BY_CRON_JOB_ID.clone(),
         ENVIRONMENT_VARIABLES_INDEX_BY_NAME.clone(),
         EXPORTS_BY_STATE_AND_TS_INDEX.clone(),
         FILE_STORAGE_ID_INDEX.clone(),
@@ -517,6 +525,7 @@ pub fn component_system_tables() -> Vec<&'static dyn SystemTable> {
         &ScheduledJobsTable,
         &CronJobsTable,
         &CronJobLogsTable,
+        &CronNextRunTable,
         &ModulesTable,
         &UdfConfigTable,
         &SourcePackagesTable,
@@ -531,6 +540,7 @@ static APP_TABLES_TO_LOAD_IN_MEMORY: LazyLock<BTreeSet<TableName>> = LazyLock::n
         MODULES_TABLE.clone(),
         ENVIRONMENT_VARIABLES_TABLE.clone(),
         CRON_JOBS_TABLE.clone(),
+        CRON_NEXT_RUN_TABLE.clone(),
         BACKEND_STATE_TABLE.clone(),
         CANONICAL_URLS_TABLE.clone(),
         BACKEND_INFO_TABLE.clone(),
@@ -570,6 +580,7 @@ pub static FIRST_SEEN_TABLE: LazyLock<BTreeMap<TableName, DatabaseVersion>> = La
         BACKEND_STATE_TABLE.clone() => 75,
         CRON_JOBS_TABLE.clone() => 47,
         CRON_JOB_LOGS_TABLE.clone() => 51,
+        CRON_NEXT_RUN_TABLE.clone() => 118,
         SOURCE_PACKAGES_TABLE.clone() => 44,
         ENVIRONMENT_VARIABLES_TABLE.clone() => 44,
         AWS_LAMBDA_VERSIONS_TABLE.clone() => 44,
@@ -599,6 +610,8 @@ pub static FIRST_SEEN_INDEX: LazyLock<BTreeMap<IndexName, DatabaseVersion>> = La
         CRON_JOBS_INDEX_BY_NEXT_TS.clone() => 47,
         CRON_JOBS_INDEX_BY_NAME.clone() => 49,
         CRON_JOB_LOGS_INDEX_BY_NAME_TS.clone() => 51,
+        CRON_NEXT_RUN_INDEX_BY_NEXT_TS.clone() => 118,
+        CRON_NEXT_RUN_INDEX_BY_CRON_JOB_ID.clone() => 118,
         EXPORTS_BY_STATE_AND_TS_INDEX.clone() => 88,
         TABLES_INDEX.clone() => 44,
         SCHEMAS_STATE_INDEX.clone() => 44,
diff --git a/crates/model/src/migrations.rs b/crates/model/src/migrations.rs
@@ -81,7 +81,7 @@ impl fmt::Display for MigrationCompletionCriterion {
 // migrations unless explicitly dropping support.
 // Add a user name next to the version when you make a change to highlight merge
 // conflicts.
-pub const DATABASE_VERSION: DatabaseVersion = 117; // nipunn
+pub const DATABASE_VERSION: DatabaseVersion = 118; // nipunn
 
 pub struct MigrationWorker<RT: Runtime> {
     rt: RT,
@@ -392,6 +392,8 @@ impl<RT: Runtime> MigrationWorker<RT> {
                     .await?;
                 MigrationCompletionCriterion::MigrationComplete(to_version)
             },
+            // Empty migration for 118 - represents creation of CronNextRun table
+            118 => MigrationCompletionCriterion::MigrationComplete(to_version),
             // NOTE: Make sure to increase DATABASE_VERSION when adding new migrations.
             _ => anyhow::bail!("Version did not define a migration! {}", to_version),
         };
