diff --git a/.sqlx/query-08e7942349b6f97210391b4fd3fdac705ddf26f92846737add3288ebaf71b0c1.json b/.sqlx/query-08e7942349b6f97210391b4fd3fdac705ddf26f92846737add3288ebaf71b0c1.json new file mode 100644 index 00000000000..09d801159f1 --- /dev/null +++ b/.sqlx/query-08e7942349b6f97210391b4fd3fdac705ddf26f92846737add3288ebaf71b0c1.json @@ -0,0 +1,30 @@ +{ + "db_name": "PostgreSQL", + "query": "\n select\n i.image as \"image!: String\",\n ct.default_capture_interval as \"default_capture_interval: crate::Interval\"\n from unnest($1::text[], $2::text[], $3::text[]) as i(image, image_name, image_tag)\n join connectors c on c.image_name = i.image_name\n join connector_tags ct on c.id = ct.connector_id and ct.image_tag = i.image_tag\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "image!: String", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "default_capture_interval: crate::Interval", + "type_info": "Interval" + } + ], + "parameters": { + "Left": [ + "TextArray", + "TextArray", + "TextArray" + ] + }, + "nullable": [ + null, + true + ] + }, + "hash": "08e7942349b6f97210391b4fd3fdac705ddf26f92846737add3288ebaf71b0c1" +} diff --git a/crates/agent/src/controllers/abandon.rs b/crates/agent/src/controllers/abandon.rs index 3ecf5e27d1a..8938d8df2aa 100644 --- a/crates/agent/src/controllers/abandon.rs +++ b/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, diff --git a/crates/agent/src/discovers.rs b/crates/agent/src/discovers.rs index a0bfdbc3583..87e559e3b91 100644 --- a/crates/agent/src/discovers.rs +++ b/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, diff --git a/crates/agent/src/integration_tests/capture_intervals.rs b/crates/agent/src/integration_tests/capture_intervals.rs new file mode 100644 index 00000000000..9171a2f7435 --- /dev/null +++ b/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, + ) + .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 + ); +} diff --git a/crates/agent/src/integration_tests/mod.rs b/crates/agent/src/integration_tests/mod.rs index 9b1367bf3ab..b2a4e1f499d 100644 --- a/crates/agent/src/integration_tests/mod.rs +++ b/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; diff --git a/crates/control-plane-api/src/publications/specs.rs b/crates/control-plane-api/src/publications/specs.rs index 74387fd44a0..fd7ce2afeaa 100644 --- a/crates/control-plane-api/src/publications/specs.rs +++ b/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, Vec) = 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, diff --git a/crates/models/src/captures.rs b/crates/models/src/captures.rs index 3d0275a824d..e3f85496e64 100644 --- a/crates/models/src/captures.rs +++ b/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, /// # 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, diff --git a/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap b/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap index de606792003..f24b12c72ce 100644 --- a/crates/sources/tests/snapshots/schema_generation__catalog_schema_snapshot.snap +++ b/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" diff --git a/crates/tables/src/lib.rs b/crates/tables/src/lib.rs index 5e8623cafe8..7fe3738ed83 100644 --- a/crates/tables/src/lib.rs +++ b/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, + } + 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, diff --git a/crates/tables/src/live.rs b/crates/tables/src/live.rs index f0b9734ed0e..ff5e5b2ca36 100644 --- a/crates/tables/src/live.rs +++ b/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, diff --git a/crates/validation/src/capture.rs b/crates/validation/src/capture.rs index f1f77db2080..c79bedca135 100644 --- a/crates/validation/src/capture.rs +++ b/crates/validation/src/capture.rs @@ -23,6 +23,7 @@ pub async fn walk_all_captures( live_captures: &tables::LiveCaptures, built_collections: &tables::BuiltCollections, connectors: &C, + connector_tags: &tables::ConnectorTags, data_planes: &tables::DataPlanes, explicit_plane: Option<&tables::DataPlane>, dependencies: &tables::Dependencies<'_>, @@ -52,6 +53,7 @@ pub async fn walk_all_captures( eob, built_collections, connectors, + connector_tags, data_planes, explicit_plane, dependencies, @@ -84,6 +86,7 @@ async fn walk_capture( eob: EOB<&tables::LiveCapture, &tables::DraftCapture>, built_collections: &tables::BuiltCollections, connectors: &C, + connector_tags: &tables::ConnectorTags, data_planes: &tables::DataPlanes, explicit_plane: Option<&tables::DataPlane>, dependencies: &tables::Dependencies<'_>, @@ -493,12 +496,27 @@ async fn walk_capture( false, // Don't disable wait_for_ack. &network_ports, ); + // Resolve the interval for this build only: a user-set `interval` wins, then + // the connector tag's default, then the global default. + let tag_default = match &endpoint { + models::CaptureEndpoint::Connector(config) => connector_tags + .get_key(&config.image) + .and_then(|tag| tag.default_capture_interval_seconds) + .map(|secs| std::time::Duration::from_secs(secs as u64)), + // Local captures run a command rather than an image, so they have no tag. + models::CaptureEndpoint::Local(_) => None, + }; + let interval_seconds = interval + .or(tag_default) + .unwrap_or_else(models::CaptureDef::default_interval) + .as_secs() as u32; + let mut spec = flow::CaptureSpec { name: capture.to_string(), connector_type, config_json, bindings: bindings_spec, - interval_seconds: interval.as_secs() as u32, + interval_seconds, recovery_log_template: Some(recovery_log_template), shard_template: Some(shard_template), network_ports, diff --git a/crates/validation/src/lib.rs b/crates/validation/src/lib.rs index 7c154e44081..ed631aed7cf 100644 --- a/crates/validation/src/lib.rs +++ b/crates/validation/src/lib.rs @@ -181,6 +181,7 @@ pub async fn validate( &live.captures, &built_collections, connectors, + &live.connector_tags, &live.data_planes, explicit_plane, &dependencies, diff --git a/crates/validation/tests/common.rs b/crates/validation/tests/common.rs index 17a54e6d380..a977bdd0188 100644 --- a/crates/validation/tests/common.rs +++ b/crates/validation/tests/common.rs @@ -170,7 +170,7 @@ pub fn run(fixture_yaml: &str, patch_yaml: &str) -> Outcome { bindings: mock.bindings.clone(), endpoint: models::CaptureEndpoint::Connector(live_connector_fixture.clone()), expect_pub_id: None, - interval: std::time::Duration::from_secs(32), + interval: Some(std::time::Duration::from_secs(32)), shards: models::ShardTemplate::default(), delete: false, reset: false, @@ -431,6 +431,11 @@ pub fn run(fixture_yaml: &str, patch_yaml: &str) -> Outcome { live.inferred_schemas .insert_row(collection, schema, "an-md5".to_string()); } + // Load into LiveCatalog::connector_tags. + for (image, default_capture_interval_seconds) in &mock_calls.connector_tags { + live.connector_tags + .insert_row(image, default_capture_interval_seconds); + } // Load into LiveCatalog::storage_mappings. for (prefix, storage) in &mock_calls.storage_mappings { live.storage_mappings.insert_row( @@ -641,6 +646,8 @@ struct MockDriverCalls { // Live catalog mocks: #[serde(default)] + connector_tags: BTreeMap>, + #[serde(default)] data_planes: BTreeMap, #[serde(default)] live_captures: BTreeMap, diff --git a/crates/validation/tests/scenario_tests.rs b/crates/validation/tests/scenario_tests.rs index 0d0d9aa8dfb..40fe3f04df1 100644 --- a/crates/validation/tests/scenario_tests.rs +++ b/crates/validation/tests/scenario_tests.rs @@ -2248,3 +2248,110 @@ test://example/db-views: let errors = common::run_errors(MODEL_YAML, &patch); insta::assert_debug_snapshot!(errors); } + +#[test] +fn capture_interval_resolves_through_connector_tag_defaults() { + // Each rung of the interval precedence ladder: an explicit model `interval` + // wins over the connector tag's default, the tag's default wins over the + // global default. + let fixture = r##" +test://example/catalog.yaml: + collections: + acmeCo/things: + schema: + type: object + properties: + id: { type: string } + required: [id] + key: [/id] + captures: + acmeCo/model-interval: + interval: 90s + shards: { disable: true } + endpoint: + connector: + image: acmeCo/source-paced:v1 + config: {} + bindings: + - resource: { table: things } + target: acmeCo/things + acmeCo/tag-interval: + shards: { disable: true } + endpoint: + connector: + image: acmeCo/source-paced:v1 + config: {} + bindings: + - resource: { table: things } + target: acmeCo/things + acmeCo/tag-without-default: + shards: { disable: true } + endpoint: + connector: + image: acmeCo/source-unpaced:v1 + config: {} + bindings: + - resource: { table: things } + target: acmeCo/things + acmeCo/unknown-image: + shards: { disable: true } + endpoint: + connector: + image: acmeCo/source-unknown:v1 + config: {} + bindings: + - resource: { table: things } + target: acmeCo/things + +driver: + dataPlanes: + "1d:1d:1d:1d:1d:1d:1d:1d": {} + connectorTags: + "acmeCo/source-paced:v1": 30 + "acmeCo/source-unpaced:v1": null +"##; + + let outcome = common::run(fixture, "{}"); + assert!(outcome.errors.is_empty(), "{:?}", outcome.errors); + + // The resolved interval must not be written back into the model: captures + // which set nothing keep tracking their connector tag's current default. + let resolved: Vec<(String, u32, Option)> = outcome + .built_captures + .iter() + .map(|built| { + ( + built.capture.to_string(), + built.spec.as_ref().unwrap().interval_seconds, + built.model.as_ref().unwrap().interval, + ) + }) + .collect(); + + insta::assert_debug_snapshot!(resolved, @r#" + [ + ( + "acmeCo/model-interval", + 90, + Some( + 90s, + ), + ), + ( + "acmeCo/tag-interval", + 30, + None, + ), + ( + "acmeCo/tag-without-default", + 300, + None, + ), + ( + "acmeCo/unknown-image", + 300, + None, + ), + ] + "#); +} diff --git a/site/docs/concepts/captures.md b/site/docs/concepts/captures.md index 1701aafdea2..0693a275d72 100644 --- a/site/docs/concepts/captures.md +++ b/site/docs/concepts/captures.md @@ -160,6 +160,7 @@ captures: # capture finishes after two minutes, then the next invocation will be started # after three additional minutes. # - # Optional. Default: Five minutes. + # Optional. When omitted, the default comes from the capture's connector, + # and is five minutes for connectors which don't specify one of their own. interval: 5m ```