crates/models: resolve capture interval from the connector tag

crates/agent/src/controllers/abandon.rs

@@ -435,7 +435,7 @@ mod test {
             bindings: vec![],
             shards: Default::default(),
             auto_discover: None,
-            interval: std::time::Duration::from_secs(300),
+            interval: None,
             redact_salt: None,
             expect_pub_id: None,
             delete: false,

crates/agent/src/discovers.rs

@@ -327,7 +327,7 @@ async fn prepare_discover(
                     add_new_bindings: true,
                     evolve_incompatible_collections: true,
                 }),
-                interval: models::CaptureDef::default_interval(),
+                interval: None,
                 redact_salt: None,
                 shards: models::ShardTemplate::default(),
                 expect_pub_id: None,

crates/agent/src/integration_tests/capture_intervals.rs

@@ -0,0 +1,123 @@
+use crate::integration_tests::harness::{TestHarness, draft_catalog};
+
+/// Exercises the real `connectors`/`connector_tags` join behind
+/// `connector_tags.default_capture_interval`, which the validation-layer tests mock out entirely,
+/// and confirms that a changed default reaches an already-published capture through its periodic
+/// touch publication.
+#[tokio::test]
+async fn test_capture_interval_follows_connector_tag_default() {
+    let mut harness =
+        TestHarness::init("test_capture_interval_follows_connector_tag_default").await;
+    let user_id = harness.setup_tenant("gadgets").await;
+
+    set_default_capture_interval(&harness, Some("90 seconds")).await;
+
+    let setup = draft_catalog(serde_json::json!({
+        "collections": {
+            "gadgets/widgets": {
+                "schema": {
+                    "type": "object",
+                    "properties": { "id": { "type": "string" } },
+                    "required": ["id"],
+                },
+                "key": ["/id"],
+            }
+        },
+        "captures": {
+            // Sets its own interval, which must win over the connector default.
+            "gadgets/pinned": {
+                "interval": "42s",
+                "endpoint": { "connector": { "image": "source/test:test", "config": {} } },
+                "bindings": [
+                    { "resource": { "id": "widgets" }, "target": "gadgets/widgets" }
+                ],
+            },
+            // Sets no interval, and so tracks the connector default.
+            "gadgets/tracking": {
+                "endpoint": { "connector": { "image": "source/test:test", "config": {} } },
+                "bindings": [
+                    { "resource": { "id": "widgets" }, "target": "gadgets/widgets" }
+                ],
+            },
+        },
+    }));
+    let result = harness
+        .user_publication(user_id, "initial publication", setup)
+        .await;
+    assert!(
+        result.status.is_success(),
+        "setup publication failed: {:?}",
+        result.errors
+    );
+    harness.run_pending_controllers(None).await;
+
+    assert_eq!(
+        42,
+        built_interval_seconds(&mut harness, "gadgets/pinned").await
+    );
+    assert_eq!(
+        90,
+        built_interval_seconds(&mut harness, "gadgets/tracking").await
+    );
+
+    // The connector default isn't baked into the model, so a capture which set
+    // no interval must pick up a later change to its connector tag. There's no
+    // fan-out which pushes the change out, so it lands with the next periodic
+    // touch publication.
+    set_default_capture_interval(&harness, Some("30 seconds")).await;
+    touch_and_run(&mut harness, 30, 42).await;
+
+    // Clearing the connector default falls back to the global default.
+    set_default_capture_interval(&harness, None).await;
+    touch_and_run(&mut harness, 300, 42).await;
+}
+
+async fn set_default_capture_interval(harness: &TestHarness, interval: Option<&str>) {
+    sqlx::query!(
+        r#"
+        update connector_tags set default_capture_interval = $1::text::interval
+        where image_tag = ':test'
+          and connector_id = (select id from connectors where image_name = 'source/test')
+        "#,
+        interval as Option<&str>,
+    )
+    .execute(&harness.pool)
+    .await
+    .expect("failed to set default_capture_interval");
+}
+
+async fn built_interval_seconds(harness: &mut TestHarness, catalog_name: &str) -> u32 {
+    let state = harness.get_controller_state(catalog_name).await;
+    let Some(proto_flow::AnyBuiltSpec::Capture(spec)) = state.built_spec.as_ref() else {
+        panic!("expected a capture spec, got: {:?}", state.built_spec);
+    };
+    spec.interval_seconds
+}
+
+/// Makes each capture's periodic touch publication come due, runs it, and asserts the intervals of
+/// the resulting builds.
+async fn touch_and_run(harness: &mut TestHarness, expect_tracking: u32, expect_pinned: u32) {
+    let captures = vec!["gadgets/pinned".to_string(), "gadgets/tracking".to_string()];
+
+    sqlx::query!(
+        r#"update live_specs set updated_at = now() - '21days'::interval
+        where catalog_name = any($1::text[])"#,
+        &captures as &Vec<String>,
+    )
+    .execute(&harness.pool)
+    .await
+    .expect("failed to age live specs");
+
+    for catalog_name in &captures {
+        harness.run_pending_controller(catalog_name).await;
+    }
+
+    assert_eq!(
+        expect_tracking,
+        built_interval_seconds(harness, "gadgets/tracking").await
+    );
+    assert_eq!(
+        expect_pinned,
+        built_interval_seconds(harness, "gadgets/pinned").await
+    );
+}

crates/agent/src/integration_tests/mod.rs

@@ -5,6 +5,7 @@
 mod abandoned_tasks;
 mod alerts;
 mod auto_discovers;
+mod capture_intervals;
 mod collection_resets;
 mod config_updates;
 mod created_at;

crates/control-plane-api/src/publications/specs.rs

@@ -969,6 +969,7 @@ pub async fn resolve_live_specs(
     .collect();
 
     resolve_inferred_schemas(draft, &mut live, db).await?;
+    resolve_connector_tags(draft, &mut live, db).await?;
 
     Ok(live)
 }
@@ -1009,6 +1010,63 @@ async fn resolve_inferred_schemas(
     Ok(())
 }
 
+/// Resolves connector tags of drafted captures and adds them to the live catalog.
+async fn resolve_connector_tags(
+    draft: &tables::DraftCatalog,
+    live: &mut tables::LiveCatalog,
+    db: &sqlx::PgPool,
+) -> anyhow::Result<()> {
+    let images: Vec<&str> = draft
+        .captures
+        .iter()
+        .filter_map(|row| match &row.model.as_ref()?.endpoint {
+            models::CaptureEndpoint::Connector(cfg) => Some(cfg.image.as_str()),
+            // Local captures run a command rather than an image, so they have no tag to resolve.
+            models::CaptureEndpoint::Local(_) => None,
+        })
+        .sorted()
+        .dedup()
+        .collect();
+
+    if images.is_empty() {
+        return Ok(());
+    }
+
+    let (image_names, image_tags): (Vec<String>, Vec<String>) = images
+        .iter()
+        .map(|image| models::split_image_tag(image))
+        .unzip();
+
+    // Unlike `fetch_connector_spec`, this doesn't require a completed spec
+    // discovery: an interval is meaningful even for a tag we've yet to inspect.
+    let rows = sqlx::query!(
+        r#"
+        select
+            i.image as "image!: String",
+            ct.default_capture_interval as "default_capture_interval: crate::Interval"
+        from unnest($1::text[], $2::text[], $3::text[]) as i(image, image_name, image_tag)
+        join connectors c on c.image_name = i.image_name
+        join connector_tags ct on c.id = ct.connector_id and ct.image_tag = i.image_tag
+        "#,
+        &images as &[&str],
+        &image_names as &[String],
+        &image_tags as &[String],
+    )
+    .fetch_all(db)
+    .await
+    .context("fetching connector tags of drafted captures")?;
+
+    for row in rows {
+        live.connector_tags.insert(tables::ConnectorTag {
+            image: row.image,
+            default_capture_interval_seconds: row
+                .default_capture_interval
+                .and_then(|interval| u32::try_from(interval.num_seconds()).ok()),
+        });
+    }
+    Ok(())
+}
+
 fn spec_meta(
     draft: &tables::DraftCatalog,
     catalog_name: &str,

crates/models/src/captures.rs

@@ -29,13 +29,16 @@ pub struct CaptureDef {
     /// For example, if the interval is five minutes, and an invocation of the
     /// capture finishes after two minutes, then the next invocation will be started
     /// after three additional minutes.
+    ///
+    /// When unset, the interval is resolved at build time: first from a default
+    /// configured for the task's connector, and otherwise from a global default.
     #[serde(
-        default = "CaptureDef::default_interval",
+        default,
         with = "humantime_serde",
-        skip_serializing_if = "CaptureDef::is_default_interval"
+        skip_serializing_if = "Option::is_none"
     )]
     #[schemars(schema_with = "super::duration_schema")]
-    pub interval: Duration,
+    pub interval: Option<Duration>,
     /// # Salt used for redacting sensitive fields in captured documents.
     /// When provided, this base64-encoded salt is used instead of a generated one.
     #[serde(
@@ -122,12 +125,11 @@ pub struct CaptureBinding {
 }
 
 impl CaptureDef {
+    /// Global fallback interval, applied when neither the model nor the task's
+    /// connector supplies one.
     pub fn default_interval() -> Duration {
         Duration::from_secs(300) // 5 minutes.
     }
-    fn is_default_interval(interval: &Duration) -> bool {
-        *interval == Self::default_interval()
-    }
 
     pub fn example() -> Self {
         Self {
@@ -137,7 +139,7 @@ impl CaptureDef {
             }),
             endpoint: CaptureEndpoint::Connector(ConnectorConfig::example()),
             bindings: vec![CaptureBinding::example()],
-            interval: Self::default_interval(),
+            interval: None,
             shards: ShardTemplate::default(),
             expect_pub_id: None,
             delete: false,

crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap

@@ -258,7 +258,7 @@ expression: "&schema"
         },
         "interval": {
           "title": "Interval of time between invocations of the capture.",
-          "description": "Configured intervals are applicable only to connectors which are\nunable to continuously tail their source, and which instead produce\na current quantity of output and then exit. Flow will start the\nconnector again after the given interval of time has passed.\n\nIntervals are relative to the start of an invocation and not its completion.\nFor example, if the interval is five minutes, and an invocation of the\ncapture finishes after two minutes, then the next invocation will be started\nafter three additional minutes.",
+          "description": "Configured intervals are applicable only to connectors which are\nunable to continuously tail their source, and which instead produce\na current quantity of output and then exit. Flow will start the\nconnector again after the given interval of time has passed.\n\nIntervals are relative to the start of an invocation and not its completion.\nFor example, if the interval is five minutes, and an invocation of the\ncapture finishes after two minutes, then the next invocation will be started\nafter three additional minutes.\n\nWhen unset, the interval is resolved at build time: first from a default\nconfigured for the task's connector, and otherwise from a global default.",
           "type": [
             "string",
             "null"

crates/tables/src/lib.rs

@@ -71,6 +71,13 @@ tables!(
         val md5: String,
     }
 
+    table ConnectorTags (row ConnectorTag, sql "connector_tags") {
+        // Full connector image, exactly as written in the model (e.g. "ghcr.io/estuary/source-x:v1").
+        key image: String,
+        // Connector-specific default for the interval between capture invocations.
+        val default_capture_interval_seconds: Option<u32>,
+    }
+
     table DataPlanes (row #[derive(Clone, serde::Serialize, serde::Deserialize)] DataPlane, sql "data_planes") {
         // Control-plane identifier for this data-plane.
         key control_id: models::Id,

crates/tables/src/live.rs

@@ -2,7 +2,7 @@ use anyhow::Context;
 use serde_json::value::RawValue;
 
 use crate::{
-    DataPlanes, Errors, InferredSchemas, LiveCapture, LiveCaptures, LiveCollection,
+    ConnectorTags, DataPlanes, Errors, InferredSchemas, LiveCapture, LiveCaptures, LiveCollection,
     LiveCollections, LiveMaterialization, LiveMaterializations, LiveTest, LiveTests,
     StorageMappings,
 };
@@ -203,6 +203,7 @@ impl LiveCatalog {
         let Self {
             captures,
             collections,
+            connector_tags,
             data_planes,
             errors,
             inferred_schemas,
@@ -214,6 +215,7 @@ impl LiveCatalog {
         vec![
             captures,
             collections,
+            connector_tags,
             data_planes,
             errors,
             inferred_schemas,
@@ -228,6 +230,7 @@ impl LiveCatalog {
         let Self {
             captures,
             collections,
+            connector_tags,
             data_planes,
             errors,
             inferred_schemas,
@@ -239,6 +242,7 @@ impl LiveCatalog {
         vec![
             captures,
             collections,
+            connector_tags,
             data_planes,
             errors,
             inferred_schemas,
@@ -254,6 +258,7 @@ impl LiveCatalog {
 pub struct LiveCatalog {
     pub captures: LiveCaptures,
     pub collections: LiveCollections,
+    pub connector_tags: ConnectorTags,
     pub data_planes: DataPlanes,
     pub errors: Errors,
     pub inferred_schemas: InferredSchemas,