Store ComponentId in CronJob (#36317)

.. so that we can correctly determine the component when re-querying it

GitOrigin-RevId: 98638fc28efafb3cddddbfbd52b186098d4dbb08

Author: goffrie

crates/application/src/cron_jobs/mod.rs

@@ -642,7 +642,7 @@ impl<RT: Runtime> CronJobExecutor<RT> {
             .begin_with_usage(Identity::Unknown(None), usage_tracker)
             .await?;
         // Verify that the cron job has not changed.
-        let new_job = CronModel::new(&mut tx, ComponentId::Root)
+        let new_job = CronModel::new(&mut tx, expected_state.component)
             .get(expected_state.id)
             .await?;
         Ok((new_job.as_ref() == Some(expected_state)).then_some(tx))

crates/model/src/cron_jobs/mod.rs

@@ -392,7 +392,7 @@ impl<'a, RT: Runtime> CronModel<'a, RT> {
             .await?
             .context("No next run found")?
             .into_value();
-        Ok(Some(CronJob::new(cron, next_run)))
+        Ok(Some(CronJob::new(cron, self.component, next_run)))
     }
 
     pub async fn list(&mut self) -> anyhow::Result<BTreeMap<CronIdentifier, CronJob>> {
@@ -406,7 +406,10 @@ impl<'a, RT: Runtime> CronModel<'a, RT> {
                 .await?
                 .context("No next run found")?
                 .into_value();
-            cron_jobs.insert(cron.name.clone(), CronJob::new(cron, next_run));
+            cron_jobs.insert(
+                cron.name.clone(),
+                CronJob::new(cron, self.component, next_run),
+            );
         }
         Ok(cron_jobs)
     }
@@ -476,7 +479,7 @@ pub async fn stream_cron_jobs_to_run<'a, RT: Runtime>(tx: &'a mut Transaction<RT
                 .await?
                 .context("No cron job found")?
                 .parse()?;
-            Ok::<_, anyhow::Error>(CronJob::new(job, next_run.into_value()))
+            Ok::<_, anyhow::Error>(CronJob::new(job, namespace.into(), next_run.into_value()))
         };
 
     // Initialize streaming query for each namespace

crates/model/src/cron_jobs/types.rs

@@ -11,6 +11,7 @@ use anyhow::{
     Context,
 };
 use common::{
+    components::ComponentId,
     document::ParsedDocument,
     log_lines::RawLogLines,
     types::Timestamp,
@@ -53,6 +54,9 @@ pub struct CronJob {
     // Id into _cron_jobs table
     pub id: ResolvedDocumentId,
 
+    // The component that defined the job
+    pub component: ComponentId,
+
     // Unique identifier of a cron
     pub name: CronIdentifier,
 
@@ -66,10 +70,15 @@ pub struct CronJob {
 }
 
 impl CronJob {
-    pub(crate) fn new(cron: ParsedDocument<CronJobMetadata>, next_run: CronNextRun) -> Self {
+    pub(crate) fn new(
+        cron: ParsedDocument<CronJobMetadata>,
+        component: ComponentId,
+        next_run: CronNextRun,
+    ) -> Self {
         let (id, cron) = cron.into_id_and_value();
         Self {
             id,
+            component,
             name: cron.name,
             cron_spec: cron.cron_spec,
             state: next_run.state,