diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9e6ca3e..add28fb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -96,3 +96,14 @@ jobs: cd conduit-ui npm install npm test + + gitleaks: + name: Secret Scan + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + - uses: gitleaks/gitleaks-action@v2 + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/.gitleaks.toml b/.gitleaks.toml new file mode 100644 index 0000000..0b958fa --- /dev/null +++ b/.gitleaks.toml @@ -0,0 +1,9 @@ +[extend] +useDefault = true + +[[allowlists]] +description = "Documentation placeholders and test fixtures, not real credentials" +regexes = [ + '''your-api-key''', + '''cdt_abc123def456abc123def456abc123''', +] diff --git a/README.md b/README.md index ff36561..8485d4c 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ **A Rust-native data pipeline orchestrator.** -Conduit is not "Airflow but faster." It solves problems that Airflow architecturally *cannot* solve — virtual pipeline environments, time-travel debugging, compile-time DAG validation, and plan/apply deployments — all in a single binary with zero external dependencies. +Conduit is not "Airflow but faster." It solves problems that Airflow architecturally *cannot* solve — virtual pipeline environments, time-travel debugging, compile-time DAG validation, and plan/apply deployments — all in a single binary with no external services (no database, no message broker; task execution shells out to your `python3`/`bash`). ![Conduit Web UI](conduit-demo-final.gif) @@ -81,8 +81,9 @@ conduit/ conduit-executor/ Process-based task runtime with timeout, retry, and sensor polling conduit-planner/ Fingerprint diffing, impact analysis, and plan/apply deployment workflow conduit-lineage/ Column-level SQL lineage via sqlparser-rs AST + TableCatalog - conduit-providers/ Data source adapters: Postgres, MySQL, SQLite, CockroachDB, Redshift, - TimescaleDB, HTTP/REST, and 26 additional provider stubs + conduit-providers/ Data source adapters: 12 implemented (Postgres, MySQL, SQLite, + CockroachDB, Redshift, TimescaleDB, BigQuery, Snowflake, S3, GCS, + DuckDB, HTTP/REST) plus 20 provider stubs conduit-api/ REST + WebSocket API with live run dispatch conduit-distributed/ Distributed executor for multi-node task dispatch conduit-ui/ React web UI (DAG visualization, run monitoring, log streaming) @@ -99,18 +100,19 @@ conduit/ Tree-sitter parses Python `@dag`/`@task` definitions **without executing Python**. Extracts schedules, tags, retry policies, pools, timeouts, and data-flow dependencies from call chains. Kahn's algorithm detects cycles, duplicates, and unknown references at compile time. Includes Criterion benchmarks for 10–1,000 DAG workloads. ### Event-Driven Scheduler (`conduit-scheduler`) -Fully async tokio-channel scheduler (no database polling). Manages DAG run state machines (Pending → Queued → Running → Success/Failed/Skipped/Retrying). Evaluates trigger rules (AllSuccess, AllDone, OneSuccess, OneFailed, NoDeps), enforces named resource pools, and parses 5-field cron expressions. +Fully async tokio-channel scheduler (no database polling). Manages task state machines (Pending → Queued → Success/Failed/Skipped/Retrying). Evaluates trigger rules (AllSuccess, AllDone, OneSuccess, OneFailed, NoDeps), enforces named resource pools declared in `conduit.yaml` (slot-limited, shared across runs), retries failed tasks with fixed or exponential backoff, and fires 5-field cron schedules (`conduit serve` ticks the scheduler every minute). ### Task Executor (`conduit-executor`) -Process-isolated task execution with stdin/stdout protocol. Supports Python, Bash, SQL, Sensor, and generic Executable task types. Enforces timeouts via `tokio::time::timeout`, implements fixed and exponential backoff retry policies, and parses structured protocol messages (XCOM, LOG, PROGRESS, METRIC) from task output. Sensor tasks poll at configurable `poke_interval` until success or timeout. Tasks are dispatched concurrently via `tokio::spawn`. +Process-isolated task execution with a stdout protocol (`CONDUIT::` lines) and env/file-based XCom injection. Supports Python, Bash, SQL, Sensor, and generic Executable task types. Enforces timeouts via `tokio::time::timeout` and parses structured protocol messages (XCOM, LOG, PROGRESS, METRIC) from task output. Retries are owned by the scheduler: fixed delay by default, exponential when a task sets `retry_backoff` (e.g. `@task(retries=3, retry_delay="30s", retry_backoff=2.0)`), and the scheduler re-dispatches the task itself when the delay elapses. Sensor tasks poll at configurable `poke_interval` until success or timeout. Tasks are dispatched concurrently via `tokio::spawn`. ### Data Providers (`conduit-providers`) -7 fully implemented providers with real database/HTTP connections: -- **SQL**: PostgreSQL, MySQL, SQLite, CockroachDB, TimescaleDB, Redshift (via sqlx connection pools) +12 implemented providers with real database/API connections: +- **SQL**: PostgreSQL, MySQL, SQLite, CockroachDB, TimescaleDB, Redshift (via sqlx connection pools), DuckDB (embedded) +- **Cloud**: BigQuery and Snowflake (REST APIs), S3 and GCS (object storage) - **HTTP**: REST API provider (via reqwest) -- 26 additional providers stubbed with the trait interface ready for implementation +- 20 additional providers stubbed with the trait interface ready for implementation (they report `NotImplemented` rather than pretending to work) -All SQL providers use lazy connection pooling, parameterized queries, and percent-encoded credentials. +All sqlx providers use lazy connection pooling and percent-encoded credentials. Task SQL supports named `:param` placeholders, rewritten to native placeholders and bound as real query parameters (never string-spliced). ### SQL Lineage (`conduit-lineage`) Column-level lineage via `sqlparser-rs` AST walking (not regex). Handles SELECT, JOINs, CTEs, UNIONs, subqueries, window functions, INSERT...SELECT, and CREATE TABLE AS SELECT. Optional `TableCatalog` integration enables bare column resolution, `SELECT *` expansion, CTE column propagation, and view column registration. OpenLineage RunEvent generation emits output `columnLineage` facets. Lineage is currently labeled **beta** — known limitations include semantic dbt/Jinja resolution and dialect-specific constructs such as BigQuery `SELECT * EXCEPT`. @@ -184,7 +186,7 @@ cargo bench -p conduit-compiler 3. **Virtual environments** (not physical copies) — snapshot pointers, not data duplication 4. **Event-driven scheduling** (not polling) — react to state changes via tokio channels 5. **Content-addressable snapshots** — fingerprint-based reuse skips unchanged tasks automatically -6. **Process isolation** — tasks run as child processes with cgroup resource limits +6. **Process isolation** — tasks run as child processes with enforced timeouts (declared CPU/memory limits are carried in the task model but not yet enforced) **What makes this architecturally different from Airflow/Dagster/Prefect**: - They poll a database to find ready tasks; Conduit reacts to events in microseconds. diff --git a/conduit-api/src/handlers/runs.rs b/conduit-api/src/handlers/runs.rs index 63b82d7..2cecda0 100644 --- a/conduit-api/src/handlers/runs.rs +++ b/conduit-api/src/handlers/runs.rs @@ -24,6 +24,9 @@ pub struct ListRunsQuery { pub status: Option, /// Filter to runs that targeted this environment (e.g. "production", "staging"). pub environment: Option, + /// Filter to runs of one DAG (accepts `dag_id` or `dagId`). + #[serde(alias = "dagId")] + pub dag_id: Option, } /// Request body for triggering a DAG run. @@ -96,6 +99,7 @@ pub async fn trigger_run( started_at: now, finished_at: None, task_states: task_states.clone(), + task_logs: HashMap::new(), triggered_by: "api".to_string(), environment: environment.clone(), }; @@ -181,6 +185,7 @@ pub async fn get_run( "startedAt": run.started_at.to_rfc3339(), "endedAt": run.finished_at.map(|t| t.to_rfc3339()), "taskStates": run.task_states, + "taskLogs": run.task_logs, "triggeredBy": run.triggered_by, "environment": run.environment, }))) @@ -192,7 +197,7 @@ pub async fn list_all_runs( Query(params): Query, ) -> Json { let limit = params.limit.unwrap_or(100); - let mut runs = state.get_runs(None); + let mut runs = state.get_runs(params.dag_id.as_deref()); if let Some(ref status) = params.status { runs.retain(|r| r.status == *status); diff --git a/conduit-api/src/lib.rs b/conduit-api/src/lib.rs index 4656ae2..ea93a70 100644 --- a/conduit-api/src/lib.rs +++ b/conduit-api/src/lib.rs @@ -19,7 +19,7 @@ pub mod routes; pub mod state; pub mod websocket; -pub use state::AppState; +pub use state::{AppState, DagRunInfo}; use std::net::SocketAddr; use std::sync::Arc; diff --git a/conduit-api/src/state.rs b/conduit-api/src/state.rs index 89724e3..315772b 100644 --- a/conduit-api/src/state.rs +++ b/conduit-api/src/state.rs @@ -37,6 +37,10 @@ pub struct DagRunInfo { #[serde(rename = "endedAt")] pub finished_at: Option>, pub task_states: HashMap, + /// Captured stdout/stderr per task (truncated), for post-hoc debugging + /// of completed runs. Live streaming still happens over the WebSocket. + #[serde(default)] + pub task_logs: HashMap, pub triggered_by: String, /// Virtual environment this run targets. Persisted JSON written before /// this field existed deserializes as "production". @@ -131,6 +135,22 @@ impl AppState { } } + // Open (creating if missing) the persistent event store. Failure is + // non-fatal: the events API reports empty and lifecycle events are + // simply not recorded. The CLI's replay command reads the same DB. + let events_dir = state_dir.join("events"); + let event_store = match conduit_state::EventStore::open(&events_dir) { + Ok(store) => Some(Arc::new(store)), + Err(e) => { + tracing::warn!( + path = %events_dir.display(), + error = %e, + "Failed to open event store; events will not be recorded", + ); + None + } + }; + // Attach env history store so promote/rollback record versions on disk. let env_manager = EnvironmentManager::new(); let env_manager = match conduit_state::EnvHistoryStore::open(state_dir.join("env_history")) @@ -140,30 +160,14 @@ impl AppState { }; let snapshot_store = Arc::new(SnapshotStore::new()); let env_manager = env_manager.with_snapshot_store(Arc::clone(&snapshot_store)); + let env_manager = match &event_store { + Some(store) => env_manager.with_event_store(Arc::clone(store)), + None => env_manager, + }; let external_lineage = Arc::new(open_external_lineage_store(&state_dir)); let plan_cache = Arc::new(PlanCache::new(dags_path.clone())); - // Open the persistent event store if it exists. Failure is non-fatal: - // the events API just reports an empty result with a clear note. The - // CLI's audit/replay commands open the same DB at the same location. - let events_dir = state_dir.join("events"); - let event_store = if events_dir.exists() { - match conduit_state::EventStore::open(&events_dir) { - Ok(store) => Some(Arc::new(store)), - Err(e) => { - tracing::warn!( - path = %events_dir.display(), - error = %e, - "Failed to open event store; /events API will return empty", - ); - None - } - } - } else { - None - }; - Arc::new(Self { dags_path, state_dir, @@ -309,8 +313,10 @@ impl AppState { .unwrap_or_default() } - /// Seed the state with realistic demo run history. - /// Called at startup when `--demo` flag is passed. + /// Seed the state with fabricated demo run history so the UI has data + /// to display. Only called when `conduit serve --demo` is passed; all + /// seeded runs carry `triggered_by: "demo"` so they are distinguishable + /// from real runs. pub fn seed_demo_data(&self) { let now = Utc::now(); @@ -365,7 +371,8 @@ impl AppState { started_at: started, finished_at: ended, task_states, - triggered_by: "scheduler".to_string(), + task_logs: HashMap::new(), + triggered_by: "demo".to_string(), environment: "production".to_string(), }); } @@ -402,7 +409,8 @@ impl AppState { started_at: started, finished_at: Some(started + Duration::hours(1) + Duration::minutes(23)), task_states, - triggered_by: "scheduler".to_string(), + task_logs: HashMap::new(), + triggered_by: "demo".to_string(), environment: "production".to_string(), }); } @@ -465,11 +473,8 @@ impl AppState { started_at: started, finished_at: ended, task_states, - triggered_by: if day == 3 { - "api".to_string() - } else { - "scheduler".to_string() - }, + task_logs: HashMap::new(), + triggered_by: "demo".to_string(), environment: if day == 4 { "staging".to_string() } else { @@ -504,7 +509,8 @@ impl AppState { started_at: started, finished_at: Some(started + Duration::minutes(28 + day * 3)), task_states, - triggered_by: "scheduler".to_string(), + task_logs: HashMap::new(), + triggered_by: "demo".to_string(), environment: if day < 2 { "staging".to_string() } else { diff --git a/conduit-api/tests/handler_tests.rs b/conduit-api/tests/handler_tests.rs index 1bc2c19..e4e20bc 100644 --- a/conduit-api/tests/handler_tests.rs +++ b/conduit-api/tests/handler_tests.rs @@ -579,6 +579,7 @@ async fn runs_list_with_seeded_data() { started_at: chrono::Utc::now(), finished_at: Some(chrono::Utc::now()), task_states: HashMap::new(), + task_logs: HashMap::new(), triggered_by: "test".to_string(), environment: "production".to_string(), }); @@ -609,6 +610,7 @@ async fn runs_list_respects_limit_param() { started_at: chrono::Utc::now(), finished_at: Some(chrono::Utc::now()), task_states: HashMap::new(), + task_logs: HashMap::new(), triggered_by: "test".to_string(), environment: "production".to_string(), }); @@ -633,6 +635,7 @@ async fn runs_list_filters_by_status() { started_at: chrono::Utc::now(), finished_at: None, task_states: HashMap::new(), + task_logs: HashMap::new(), triggered_by: "test".to_string(), environment: "production".to_string(), }); @@ -658,6 +661,7 @@ async fn runs_list_filters_by_environment() { started_at: chrono::Utc::now(), finished_at: None, task_states: HashMap::new(), + task_logs: HashMap::new(), triggered_by: "test".to_string(), environment: env.to_string(), }); @@ -730,6 +734,7 @@ async fn run_get_by_id() { ("task_a".into(), "success".into()), ("task_b".into(), "success".into()), ]), + task_logs: HashMap::new(), triggered_by: "api".to_string(), environment: "production".to_string(), }); @@ -1045,3 +1050,56 @@ async fn cluster_status_has_expected_fields() { // Should return an object with cluster info. assert!(body.is_object()); } + +#[tokio::test] +async fn runs_list_filters_by_dag_id_param() { + let (router, state) = app(false); + + for (i, dag) in ["alpha", "beta", "alpha"].iter().enumerate() { + state.record_run(DagRunInfo { + run_id: format!("run-{}", i), + dag_id: dag.to_string(), + status: "success".to_string(), + started_at: chrono::Utc::now(), + finished_at: Some(chrono::Utc::now()), + task_states: HashMap::new(), + task_logs: HashMap::new(), + triggered_by: "test".to_string(), + environment: "production".to_string(), + }); + } + + // Both snake_case and camelCase spellings must filter. + let (_, body) = get(&router, "/api/v1/runs?dag_id=alpha").await; + assert_eq!(body["total"], 2, "dag_id filter must apply: {}", body); + + let (_, body) = get(&router, "/api/v1/runs?dagId=beta").await; + assert_eq!(body["total"], 1, "dagId filter must apply: {}", body); +} + +#[tokio::test] +async fn run_detail_exposes_task_logs() { + let (router, state) = app(false); + + let mut task_logs = HashMap::new(); + task_logs.insert("greet".to_string(), "Hello from Conduit!\n".to_string()); + state.record_run(DagRunInfo { + run_id: "run-logs".to_string(), + dag_id: "hello".to_string(), + status: "success".to_string(), + started_at: chrono::Utc::now(), + finished_at: Some(chrono::Utc::now()), + task_states: HashMap::new(), + task_logs, + triggered_by: "test".to_string(), + environment: "production".to_string(), + }); + + let (status, body) = get(&router, "/api/v1/runs/run-logs").await; + assert_eq!(status, StatusCode::OK); + assert_eq!( + body["taskLogs"]["greet"], "Hello from Conduit!\n", + "run detail must expose captured task logs: {}", + body + ); +} diff --git a/conduit-api/tests/pipeline_e2e_test.rs b/conduit-api/tests/pipeline_e2e_test.rs index 60fe2ee..8e642d7 100644 --- a/conduit-api/tests/pipeline_e2e_test.rs +++ b/conduit-api/tests/pipeline_e2e_test.rs @@ -48,6 +48,8 @@ fn make_etl_dag() -> Dag { dependencies: vec![], retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: Some("60s".into()), priority: 0, @@ -73,6 +75,8 @@ fn make_etl_dag() -> Dag { }], retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: Some("120s".into()), priority: 0, @@ -98,6 +102,8 @@ fn make_etl_dag() -> Dag { }], retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: Some("60s".into()), priority: 0, @@ -150,6 +156,8 @@ fn make_diamond_dag() -> Dag { dependencies: vec![], retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, @@ -176,6 +184,8 @@ fn make_diamond_dag() -> Dag { }], retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, @@ -208,6 +218,8 @@ fn make_diamond_dag() -> Dag { ], retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-bench/src/lib.rs b/conduit-bench/src/lib.rs index 5f262d9..cd8d59c 100644 --- a/conduit-bench/src/lib.rs +++ b/conduit-bench/src/lib.rs @@ -40,6 +40,8 @@ pub fn generate_dag(dag_id: &str, n_tasks: usize) -> Dag { trigger_rule: TriggerRule::AllSuccess, retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, @@ -76,6 +78,8 @@ pub fn generate_dag(dag_id: &str, n_tasks: usize) -> Dag { trigger_rule: TriggerRule::AllSuccess, retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, @@ -98,6 +102,8 @@ pub fn generate_dag(dag_id: &str, n_tasks: usize) -> Dag { trigger_rule: TriggerRule::AllSuccess, retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, @@ -127,6 +133,8 @@ pub fn generate_dag(dag_id: &str, n_tasks: usize) -> Dag { trigger_rule: TriggerRule::AllSuccess, retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, @@ -156,6 +164,8 @@ pub fn generate_dag(dag_id: &str, n_tasks: usize) -> Dag { trigger_rule: TriggerRule::AllSuccess, retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-cli/src/main.rs b/conduit-cli/src/main.rs index 23db4ab..509eaed 100644 --- a/conduit-cli/src/main.rs +++ b/conduit-cli/src/main.rs @@ -61,6 +61,9 @@ fn resolve_state_dir(dags_path: &Path) -> PathBuf { struct PersistentState { env_manager: conduit_state::EnvironmentManager, snapshot_store: std::sync::Arc, + /// Best-effort event log for `conduit replay`. None when the events DB + /// is unavailable (e.g. locked by a running `conduit serve`). + event_store: Option>, state_dir: PathBuf, } @@ -117,9 +120,30 @@ impl PersistentState { // that gate on snapshot age can resolve snapshot IDs. let env_manager = env_manager.with_snapshot_store(std::sync::Arc::clone(&snapshot_store)); + // Open the event log so env lifecycle and apply operations are + // recorded for `conduit replay`. Best-effort: a locked/unavailable + // DB (e.g. `conduit serve` holds it) only disables recording. + let events_dir = state_dir.join("events"); + let event_store = match conduit_state::EventStore::open(&events_dir) { + Ok(store) => Some(std::sync::Arc::new(store)), + Err(e) => { + tracing::warn!( + path = %events_dir.display(), + error = %e, + "Event store unavailable; state changes will not be recorded for replay" + ); + None + } + }; + let env_manager = match &event_store { + Some(store) => env_manager.with_event_store(std::sync::Arc::clone(store)), + None => env_manager, + }; + Ok(Self { env_manager, snapshot_store, + event_store, state_dir: state_dir.to_path_buf(), }) } @@ -285,6 +309,12 @@ enum Commands { /// --cors-origin http://localhost:3000 #[arg(long = "cors-origin")] cors_origins: Vec, + + /// Seed fabricated demo run history (for trying out the UI with + /// data to look at). Never enabled by default: without this flag + /// the server starts with only your real runs. + #[arg(long)] + demo: bool, }, /// Show system status @@ -740,6 +770,7 @@ fn main() -> Result<()> { state_dir, auth_enabled, cors_origins, + demo, } => rt.block_on(cmd_serve( &host, port, @@ -747,6 +778,7 @@ fn main() -> Result<()> { &state_dir, auth_enabled, cors_origins, + demo, )), Commands::Status { env, dags_path } => cmd_status(env.as_deref(), &dags_path), Commands::Env { action, dags_path } => match action { @@ -1081,6 +1113,44 @@ fn cmd_compile(path: &PathBuf, output: Option<&Path>, check: bool) -> Result<()> } /// Walk every SQL task across every DAG; for each one, look up the named +/// Locate the project's conduit.yaml by walking up from the dags path. +fn find_conduit_yaml(dags_path: &Path) -> Option { + let mut candidate = dags_path.to_path_buf(); + if candidate.is_relative() { + candidate = std::env::current_dir().unwrap_or_default().join(candidate); + } + let project_root = if candidate.ends_with("dags") { + candidate.parent().map(Path::to_path_buf) + } else { + Some(candidate) + }; + project_root.and_then(|p| { + let y = p.join("conduit.yaml"); + y.exists().then_some(y) + }) +} + +/// Load named resource pools from the project's conduit.yaml (`pools:` +/// section). Missing file/section means no declared pools — tasks that +/// reference an undeclared pool run unlimited (the scheduler warns). +fn load_pools(dags_path: &Path) -> Vec { + let Some(yaml_path) = find_conduit_yaml(dags_path) else { + return Vec::new(); + }; + let Ok(config) = conduit_common::config::ConduitConfig::load(&yaml_path) else { + return Vec::new(); + }; + config + .pools + .into_iter() + .map(|(name, p)| conduit_common::dag::Pool { + name, + slots: p.slots, + description: p.description, + }) + .collect() +} + /// connection in the project's conduit.yaml; if its `conn_type` is a stub, /// print a warning. We surface the offending DAG + task so users can find /// and replace it before deploying. @@ -1218,16 +1288,34 @@ async fn cmd_run( let mut dag_map = HashMap::new(); dag_map.insert(dag_id.to_string(), dag.clone()); - let pools = PoolManager::new(vec![]); + let pools = PoolManager::new(load_pools(dags_path)); let scheduler = Scheduler::new(event_rx, cmd_tx, pools, dag_map)?; + // Persist run history so `conduit replay` can reconstruct it. Best + // effort: if the store is locked (e.g. `conduit serve` is running) the + // run proceeds without persistence. + let events_dir = resolve_state_dir(dags_path).join("events"); + let scheduler = match conduit_state::EventStore::open(&events_dir) { + Ok(store) => scheduler.with_event_store(std::sync::Arc::new(store)), + Err(e) => { + tracing::warn!( + path = %events_dir.display(), + error = %e, + "Event store unavailable; run will not be recorded for replay" + ); + scheduler + } + }; + // Request the DAG run let run_id = format!("run_{}", Utc::now().format("%Y%m%d_%H%M%S")); + let mut run_config = HashMap::new(); + run_config.insert("triggered_by".to_string(), "cli".to_string()); event_tx.send(SchedulerEvent::DagRunRequested { dag_id: dag_id.to_string(), run_id: run_id.clone(), logical_date, - config: HashMap::new(), + config: run_config, })?; // Spawn scheduler @@ -1236,6 +1324,8 @@ async fn cmd_run( // Executor loop: receives SchedulerCommands, runs real processes let executor_event_tx = event_tx.clone(); let dag_for_executor = dag.clone(); + let run_failed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let run_failed_flag = std::sync::Arc::clone(&run_failed); let executor_handle = tokio::spawn(async move { let mut completed = 0usize; let total = dag_for_executor.tasks.len(); @@ -1362,6 +1452,9 @@ async fn cmd_run( } => { println!(); println!("DAG '{}' run '{}' completed: {:?}", dag_id, run_id, status); + if !matches!(status, conduit_scheduler::scheduler::RunStatus::Success) { + run_failed_flag.store(true, std::sync::atomic::Ordering::SeqCst); + } let _ = executor_event_tx.send(SchedulerEvent::Shutdown); break; } @@ -1372,7 +1465,12 @@ async fn cmd_run( completed += 1; } SchedulerCommand::RetryTask { task_id, delay, .. } => { - println!(" [RETRY] {} (delay: {:?})", task_id, delay); + // The scheduler re-dispatches the task itself once the + // delay elapses; this is just progress output. The failed + // attempt wasn't final, so give its slot back to the + // progress counter. + completed = completed.saturating_sub(1); + println!(" [RETRY] {} (retrying in {}s)", task_id, delay.num_seconds()); } } } @@ -1383,6 +1481,11 @@ async fn cmd_run( let total_duration = start.elapsed(); println!("Total time: {:.1}ms", total_duration.as_secs_f64() * 1000.0); + // A failed run must fail the command (CI gates on this exit code). + if run_failed.load(std::sync::atomic::Ordering::SeqCst) { + anyhow::bail!("DAG run failed — see task output above"); + } + Ok(()) } @@ -1641,6 +1744,16 @@ async fn cmd_apply( metadata: HashMap::new(), }; let _ = state.snapshot_store.put(snapshot); + if let Some(store) = &state.event_store { + let _ = store.append( + conduit_common::event::EventKind::SnapshotCreated { + snapshot_id: snap_id.clone(), + fingerprint: fp.0.clone(), + dag_id: action.dag_id.clone(), + task_id: action.task_id.clone(), + }, + ); + } } new_snapshots @@ -1719,6 +1832,15 @@ async fn cmd_apply( // Persist state to disk state.save()?; + if let Some(store) = &state.event_store { + let _ = store.append(conduit_common::event::EventKind::PlanApplied { + plan_id: deploy.id.clone(), + environment: environment.to_string(), + tasks_executed: executed as u32, + tasks_skipped: reused as u32, + }); + } + let duration = start.elapsed(); println!(); println!("Apply complete:"); @@ -1751,6 +1873,7 @@ async fn cmd_serve( state_dir: &PathBuf, auth_enabled: bool, cors_origins: Vec, + demo: bool, ) -> Result<()> { use std::net::SocketAddr; @@ -1824,8 +1947,12 @@ async fn cmd_serve( println!(); } - // Seed realistic demo run history so the UI has data to display immediately - state.seed_demo_data(); + // Demo mode only: seed fabricated run history so the UI has data to + // look at. A real deployment must never mix fake runs into real ones. + if demo { + println!(" Demo mode: seeding fabricated run history (--demo)"); + state.seed_demo_data(); + } println!( " Demo data: seeded {} historical runs", state.get_runs(None).len() @@ -1861,11 +1988,39 @@ async fn cmd_serve( // Attach event sender to AppState so trigger_run can dispatch events state.with_scheduler(event_tx.clone()); - // Spawn the scheduler event loop - let pools = PoolManager::new(vec![]); - let scheduler = Scheduler::new(event_rx, cmd_tx, pools, dag_map.clone())?; + // Spawn the scheduler event loop. Share the API's event store so + // run lifecycle events are persisted for `conduit replay` and the + // /events API. + let pools = PoolManager::new(load_pools(dags_path)); + let mut scheduler = Scheduler::new(event_rx, cmd_tx, pools, dag_map.clone())?; + if let Some(store) = &state.event_store { + scheduler = scheduler.with_event_store(std::sync::Arc::clone(store)); + } tokio::spawn(async move { scheduler.run().await }); + // Cron tick source: wake the scheduler at the top of every minute so + // DAG schedules actually fire. Five-field cron has minute resolution, + // and the scheduler dedupes within a minute, so once per minute is + // exactly right. + let cron_tx = event_tx.clone(); + tokio::spawn(async move { + loop { + let now = chrono::Utc::now(); + let millis_into_minute = + (now.timestamp_millis().rem_euclid(60_000)) as u64; + let sleep_ms = 60_000 - millis_into_minute; + tokio::time::sleep(std::time::Duration::from_millis(sleep_ms)).await; + if cron_tx + .send(SchedulerEvent::CronTick { + timestamp: chrono::Utc::now(), + }) + .is_err() + { + break; // scheduler is gone; stop ticking + } + } + }); + // Spawn the executor loop — receives SchedulerCommands, runs tasks, // updates AppState, and broadcasts WebSocket events let exec_state = state.clone(); @@ -1898,6 +2053,11 @@ async fn cmd_serve( } }; + // Scheduler-initiated runs (cron) were never recorded + // via the POST /runs handler — create the run entry on + // first dispatch so they show up in the UI and API. + ensure_run_recorded(&exec_state, &dag_id, &run_id); + // Mark task as running update_run_task_state(&exec_state, &run_id, &task_id, "running"); exec_state.broadcast_event( @@ -1928,6 +2088,13 @@ async fn cmd_serve( match ProcessRunner::run(&task, &context).await { Ok(output) => { let duration = task_start.elapsed(); + update_run_task_logs( + &spawn_state, + &run_id, + &task_id, + &output.stdout, + &output.stderr, + ); if output.exit_code == 0 { update_run_task_state( &spawn_state, @@ -1979,6 +2146,13 @@ async fn cmd_serve( } } Err(e) => { + update_run_task_logs( + &spawn_state, + &run_id, + &task_id, + "", + &e.to_string(), + ); update_run_task_state( &spawn_state, &run_id, @@ -2058,6 +2232,68 @@ async fn cmd_serve( Ok(()) } +/// Record a scheduler-initiated run (e.g. cron) in the API run cache if it +/// isn't already there. Runs triggered via POST /runs are recorded by the +/// handler; this covers every other dispatch source. +fn ensure_run_recorded(state: &conduit_api::AppState, dag_id: &str, run_id: &str) { + if let Ok(runs) = state.runs.read() { + if runs.iter().any(|r| r.run_id == run_id) { + return; + } + } + state.record_run(conduit_api::DagRunInfo { + run_id: run_id.to_string(), + dag_id: dag_id.to_string(), + status: "running".to_string(), + started_at: chrono::Utc::now(), + finished_at: None, + task_states: std::collections::HashMap::new(), + task_logs: std::collections::HashMap::new(), + triggered_by: "scheduler".to_string(), + environment: "production".to_string(), + }); +} + +/// Store a completed task's captured output on its run so the run detail +/// view can show logs after the fact. Capped per task to bound memory; the +/// live WebSocket stream remains the full-fidelity path. +fn update_run_task_logs( + state: &conduit_api::AppState, + run_id: &str, + task_id: &str, + stdout: &str, + stderr: &str, +) { + const MAX_LOG_BYTES: usize = 16 * 1024; + let mut text = String::new(); + if !stdout.trim().is_empty() { + text.push_str(stdout.trim_end()); + text.push('\n'); + } + if !stderr.trim().is_empty() { + text.push_str("--- stderr ---\n"); + text.push_str(stderr.trim_end()); + text.push('\n'); + } + if text.is_empty() { + return; + } + if text.len() > MAX_LOG_BYTES { + let cut = text.len() - MAX_LOG_BYTES; + // keep the tail — failures usually end with the interesting part + let mut idx = cut; + while !text.is_char_boundary(idx) { + idx += 1; + } + text = format!("… (truncated {} bytes)\n{}", cut, &text[idx..]); + } + if let Ok(mut runs) = state.runs.write() { + if let Some(run) = runs.iter_mut().find(|r| r.run_id == run_id) { + run.task_logs.insert(task_id.to_string(), text); + } + } +} + /// Update a specific task's state within a run. fn update_run_task_state( state: &conduit_api::AppState, @@ -2194,10 +2430,9 @@ fn cmd_env_diff(a: &str, b: &str, dags_path: &PathBuf) -> Result<()> { return Ok(()); } - let short = |s: &str| -> String { - let n = s.len().min(12); - s[..n].to_string() - }; + // Snapshot IDs are "snap__"; the unique part is the + // tail, so truncating to a prefix would render every version identical. + let short = |s: &str| -> String { s.to_string() }; let mut removed = diff.removed.clone(); removed.sort_by(|x, y| (&x.dag_id, &x.task_id).cmp(&(&y.dag_id, &y.task_id))); @@ -3106,6 +3341,21 @@ async fn cmd_backfill( println!("Executing partitions..."); println!(); + // Open the event store once for the whole backfill; each partition's + // scheduler shares it so runs are recorded for `conduit replay`. + let backfill_events_dir = resolve_state_dir(dags_path).join("events"); + let backfill_event_store = match conduit_state::EventStore::open(&backfill_events_dir) { + Ok(store) => Some(std::sync::Arc::new(store)), + Err(e) => { + tracing::warn!( + path = %backfill_events_dir.display(), + error = %e, + "Event store unavailable; backfill runs will not be recorded for replay" + ); + None + } + }; + let mut succeeded = 0usize; let mut failed = 0usize; let skipped = 0usize; @@ -3130,8 +3380,11 @@ async fn cmd_backfill( let mut dag_map = HashMap::new(); dag_map.insert(dag_id.to_string(), dag.clone()); - let pools = PoolManager::new(vec![]); - let scheduler = Scheduler::new(event_rx, cmd_tx, pools, dag_map)?; + let pools = PoolManager::new(load_pools(dags_path)); + let mut scheduler = Scheduler::new(event_rx, cmd_tx, pools, dag_map)?; + if let Some(store) = &backfill_event_store { + scheduler = scheduler.with_event_store(std::sync::Arc::clone(store)); + } let run_id = format!("bf_{}_{}", dag_id, partition.partition_key); event_tx.send(SchedulerEvent::DagRunRequested { diff --git a/conduit-common/src/dag.rs b/conduit-common/src/dag.rs index 42e6a27..85d5b52 100644 --- a/conduit-common/src/dag.rs +++ b/conduit-common/src/dag.rs @@ -152,6 +152,17 @@ pub struct Task { /// Delay between retries (e.g., "5m", "30s"). pub retry_delay: Option, + /// Exponential backoff multiplier applied to `retry_delay` per attempt + /// (e.g., 2.0 doubles the delay each retry). None or <= 1.0 = fixed delay. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub retry_backoff: Option, + + /// Stable hash of the task's source text, set by the compiler for + /// Python tasks (whose `task_type` carries only module/function names). + /// Folded into the planner fingerprint so body edits are detected. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source_hash: Option, + /// Named resource pool this task draws from. pub pool: Option, @@ -246,7 +257,8 @@ impl TaskType { } } -/// Resource limits for a task's cgroup. +/// Declared resource limits for a task. Carried through the task model +/// and distributed protocol, but not yet enforced at process spawn time. #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct ResourceLimits { /// CPU limit in millicores (e.g., 1000 = 1 core). diff --git a/conduit-common/src/fingerprint.rs b/conduit-common/src/fingerprint.rs index 7500607..385393d 100644 --- a/conduit-common/src/fingerprint.rs +++ b/conduit-common/src/fingerprint.rs @@ -8,7 +8,8 @@ use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; use std::fmt; -/// A content-addressable fingerprint (SHA-256 hex string). +/// A content-addressable fingerprint (64-bit SipHash of content, config, +/// and upstream fingerprints, rendered as a hex string). #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] pub struct Fingerprint(pub String); diff --git a/conduit-compiler/src/parser.rs b/conduit-compiler/src/parser.rs index 0510ce5..3fc2c93 100644 --- a/conduit-compiler/src/parser.rs +++ b/conduit-compiler/src/parser.rs @@ -42,6 +42,13 @@ pub struct ParsedTask { pub task_type: TaskType, pub retries: u32, pub retry_delay: Option, + /// Exponential backoff multiplier per retry attempt (None = fixed delay). + pub retry_backoff: Option, + /// Stable hash of the task's source text (nested @task def plus any + /// module-level function of the same name). Lets the planner detect + /// Python body edits. None for task types whose content is already + /// carried verbatim (SQL query text, bash command). + pub source_hash: Option, pub pool: Option, pub timeout: Option, pub priority: i32, @@ -59,6 +66,16 @@ pub struct ParsedTask { pub outputs: Vec, } +/// Stable, deterministic hash of task source text, used for change +/// detection in the planner (same construction as `Fingerprint::compute`). +fn hash_source(text: &str) -> String { + use std::collections::hash_map::DefaultHasher; + use std::hash::{Hash, Hasher}; + let mut hasher = DefaultHasher::new(); + text.hash(&mut hasher); + format!("{:016x}", hasher.finish()) +} + impl DagParser { /// Create a new DAG parser with the Python tree-sitter grammar. pub fn new() -> ConduitResult { @@ -114,9 +131,55 @@ impl DagParser { debug!(file = file_name, "No DAG definitions found"); } + // Fold module-level function bodies into task source hashes. The + // SDK's executable-body pattern defines the code that actually runs + // at module level under the same name as the nested @task stub — + // editing it must invalidate the task's fingerprint too. + let module_fns = self.collect_module_functions(&root, source_bytes); + for dag in &mut dags { + for task in &mut dag.tasks { + if let Some(module_text) = module_fns.get(&task.id) { + let combined = format!( + "{}\n{}", + task.source_hash.as_deref().unwrap_or(""), + hash_source(module_text) + ); + task.source_hash = Some(hash_source(&combined)); + } + } + } + Ok(dags) } + /// Map every module-level function name to its full source text + /// (including any decorators). + fn collect_module_functions( + &self, + root: &tree_sitter::Node, + source: &[u8], + ) -> HashMap { + let mut fns = HashMap::new(); + let mut cursor = root.walk(); + for child in root.children(&mut cursor) { + let mut inner = child.walk(); + let func = match child.kind() { + "function_definition" => Some(child), + "decorated_definition" => child + .children(&mut inner) + .find(|c| c.kind() == "function_definition"), + _ => None, + }; + if let Some(f) = func { + if let Some(name_node) = f.child_by_field_name("name") { + let name = self.node_text(&name_node, source); + fns.insert(name, self.node_text(&child, source)); + } + } + } + fns + } + /// Try to parse a decorated_definition as a @dag-decorated function. fn try_parse_dag( &self, @@ -376,6 +439,8 @@ impl DagParser { .and_then(|s| s.parse().ok()) .unwrap_or(0), retry_delay: task_args.get("retry_delay").cloned(), + retry_backoff: task_args.get("retry_backoff").and_then(|s| s.parse().ok()), + source_hash: Some(hash_source(&self.node_text(node, source))), pool: task_args.get("pool").cloned(), timeout: task_args.get("timeout").cloned(), priority: task_args @@ -864,7 +929,7 @@ from conduit import dag, task, Param def daily_warehouse_refresh(date: Param[str] = "{{ ds }}"): """Refresh the warehouse daily.""" - @task(retries=3, retry_delay="5m", pool="snowflake") + @task(retries=3, retry_delay="5m", retry_backoff=2.0, pool="snowflake") def extract_orders(date: str): """Pull orders from source.""" pass @@ -907,6 +972,7 @@ def daily_warehouse_refresh(date: Param[str] = "{{ ds }}"): let extract = dag.tasks.iter().find(|t| t.id == "extract_orders").unwrap(); assert_eq!(extract.retries, 3); assert_eq!(extract.retry_delay, Some("5m".to_string())); + assert_eq!(extract.retry_backoff, Some(2.0)); assert_eq!(extract.pool, Some("snowflake".to_string())); let transform = dag @@ -1018,3 +1084,92 @@ def bad(): ); } } + +#[cfg(test)] +mod source_hash_tests { + use super::*; + + const DAG_V1: &str = r#" +from conduit_sdk import dag, task + +@dag(schedule="@daily") +def my_dag(): + @task() + def work(): + print("v1") +"#; + + // Identical to V1 except for the task body. + const DAG_V2: &str = r#" +from conduit_sdk import dag, task + +@dag(schedule="@daily") +def my_dag(): + @task() + def work(): + print("v2 - changed body") +"#; + + // V1 plus a module-level function that shares the task's name (the + // executable-body pattern used by the SDK / demo pipelines). + const DAG_V1_WITH_MODULE_BODY: &str = r#" +from conduit_sdk import dag, task + +@dag(schedule="@daily") +def my_dag(): + @task() + def work(): + print("v1") + +def work(): + print("module-level body v1") +"#; + + const DAG_V1_WITH_CHANGED_MODULE_BODY: &str = r#" +from conduit_sdk import dag, task + +@dag(schedule="@daily") +def my_dag(): + @task() + def work(): + print("v1") + +def work(): + print("module-level body v2 - changed") +"#; + + fn parse_hash(source: &str) -> Option { + let mut parser = DagParser::new().unwrap(); + let dags = parser.parse_source(source, "test.py").unwrap(); + dags[0] + .tasks + .iter() + .find(|t| t.id == "work") + .unwrap() + .source_hash + .clone() + } + + #[test] + fn task_body_change_changes_source_hash() { + let h1 = parse_hash(DAG_V1); + let h2 = parse_hash(DAG_V2); + assert!(h1.is_some(), "python tasks must carry a source hash"); + assert_ne!(h1, h2, "editing a task body must change its source hash"); + } + + #[test] + fn module_level_body_change_changes_source_hash() { + let h1 = parse_hash(DAG_V1_WITH_MODULE_BODY); + let h2 = parse_hash(DAG_V1_WITH_CHANGED_MODULE_BODY); + assert_ne!( + h1, h2, + "editing a module-level task body must change its source hash" + ); + } + + #[test] + fn unchanged_source_produces_stable_hash() { + assert_eq!(parse_hash(DAG_V1), parse_hash(DAG_V1)); + } +} diff --git a/conduit-compiler/src/resolver.rs b/conduit-compiler/src/resolver.rs index 5104388..263a4d8 100644 --- a/conduit-compiler/src/resolver.rs +++ b/conduit-compiler/src/resolver.rs @@ -102,6 +102,8 @@ impl DependencyResolver { .collect(), retries: pt.retries, retry_delay: pt.retry_delay, + retry_backoff: pt.retry_backoff, + source_hash: pt.source_hash, pool: pt.pool, timeout: pt.timeout, priority: pt.priority, @@ -221,6 +223,8 @@ mod tests { }, retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-compiler/src/sql_io_inference.rs b/conduit-compiler/src/sql_io_inference.rs index 8a15023..6cb29e2 100644 --- a/conduit-compiler/src/sql_io_inference.rs +++ b/conduit-compiler/src/sql_io_inference.rs @@ -119,6 +119,8 @@ mod tests { dependencies: Vec::new(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-compiler/src/yaml_parser.rs b/conduit-compiler/src/yaml_parser.rs index 67a1fdd..3a6f6c9 100644 --- a/conduit-compiler/src/yaml_parser.rs +++ b/conduit-compiler/src/yaml_parser.rs @@ -157,6 +157,10 @@ pub struct YamlTask { #[serde(default)] pub retry_delay: Option, + /// Exponential backoff multiplier per retry attempt (None = fixed delay). + #[serde(default)] + pub retry_backoff: Option, + /// Resource pool name. #[serde(default)] pub pool: Option, @@ -411,6 +415,8 @@ impl YamlDagParser { task_type, retries: yaml_task.retries, retry_delay: yaml_task.retry_delay.clone(), + retry_backoff: yaml_task.retry_backoff, + source_hash: None, pool: yaml_task.pool.clone(), timeout: yaml_task.timeout.clone(), priority: yaml_task.priority, @@ -773,6 +779,9 @@ tasks: connection: warehouse query: "SELECT * FROM source.orders" retries: 3 + retry_delay: 30s + retry_backoff: 2.0 + source_hash: None, timeout: 30m transform: @@ -798,6 +807,8 @@ tasks: let extract = dag.tasks.iter().find(|t| t.id == "extract").unwrap(); assert!(matches!(extract.task_type, TaskType::Sql { .. })); assert_eq!(extract.retries, 3); + assert_eq!(extract.retry_delay, Some("30s".to_string())); + assert_eq!(extract.retry_backoff, Some(2.0)); let transform = dag.tasks.iter().find(|t| t.id == "transform").unwrap(); assert!(matches!(transform.task_type, TaskType::Python { .. })); diff --git a/conduit-compiler/tests/proptest_resolver.rs b/conduit-compiler/tests/proptest_resolver.rs index 2508519..11c5d4d 100644 --- a/conduit-compiler/tests/proptest_resolver.rs +++ b/conduit-compiler/tests/proptest_resolver.rs @@ -21,6 +21,8 @@ fn make_task(id: &str, deps: &[String]) -> ParsedTask { }, retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-executor/Cargo.toml b/conduit-executor/Cargo.toml index 840799c..1c8b2f1 100644 --- a/conduit-executor/Cargo.toml +++ b/conduit-executor/Cargo.toml @@ -3,7 +3,7 @@ name = "conduit-executor" version.workspace = true edition.workspace = true license.workspace = true -description = "Task runtime: process isolation, resource limits, stdin/stdout protocol" +description = "Task runtime: process isolation, timeouts, stdout protocol with env/file XCom" [dependencies] conduit-common = { path = "../conduit-common" } diff --git a/conduit-executor/src/lib.rs b/conduit-executor/src/lib.rs index 1446313..82560f4 100644 --- a/conduit-executor/src/lib.rs +++ b/conduit-executor/src/lib.rs @@ -15,7 +15,7 @@ pub use conduit_providers::ProviderRegistry; pub use executor::{ExecutorCommand, ExecutorEvent, TaskExecutor, TaskOutcome}; pub use process_runner::{ProcessOutput, ProcessRunner, TaskContext}; pub use protocol::{parse_stdout_line, ProtocolMessage}; -pub use retry::{parse_duration, RetryPolicy}; +pub use retry::parse_duration; #[cfg(test)] mod tests { @@ -27,6 +27,5 @@ mod tests { let _ = std::any::type_name::(); let _ = std::any::type_name::(); let _ = std::any::type_name::(); - let _ = std::any::type_name::(); } } diff --git a/conduit-executor/src/process_runner.rs b/conduit-executor/src/process_runner.rs index c82cb50..2210294 100644 --- a/conduit-executor/src/process_runner.rs +++ b/conduit-executor/src/process_runner.rs @@ -753,6 +753,8 @@ mod tests { dependencies: vec![], retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-executor/src/retry.rs b/conduit-executor/src/retry.rs index 80e2d7c..1ad8e2d 100644 --- a/conduit-executor/src/retry.rs +++ b/conduit-executor/src/retry.rs @@ -1,88 +1,13 @@ -//! Retry policy and backoff strategy management. +//! Duration parsing/formatting for task execution. //! -//! This module provides utilities for parsing retry delays and computing -//! exponential backoff for task retries. +//! Retry timing (fixed delay and exponential backoff) is owned by the +//! scheduler (`conduit-scheduler`), which re-dispatches tasks when their +//! retry delay elapses. This module only provides duration utilities. use conduit_common::{ConduitError, ConduitResult}; use std::time::Duration; use tracing::trace; -/// Retry policy configuration -#[derive(Debug, Clone)] -pub struct RetryPolicy { - pub max_retries: u32, - pub delay: Duration, - pub backoff_factor: f64, -} - -impl RetryPolicy { - /// Create a new retry policy - pub fn new(max_retries: u32, delay: Duration, backoff_factor: f64) -> Self { - Self { - max_retries, - delay, - backoff_factor, - } - } - - /// Create a retry policy with no retries - pub fn no_retries() -> Self { - Self { - max_retries: 0, - delay: Duration::from_secs(0), - backoff_factor: 1.0, - } - } - - /// Create a retry policy with fixed delay - pub fn fixed(max_retries: u32, delay: Duration) -> Self { - Self { - max_retries, - delay, - backoff_factor: 1.0, - } - } - - /// Create a retry policy with exponential backoff - pub fn exponential(max_retries: u32, initial_delay: Duration, backoff_factor: f64) -> Self { - Self { - max_retries, - delay: initial_delay, - backoff_factor, - } - } - - /// Calculate the delay for a given attempt number - /// - /// With exponential backoff: delay * backoff_factor^attempt - pub fn delay_for_attempt(&self, attempt: u32) -> Duration { - if attempt == 0 || self.backoff_factor == 1.0 { - return self.delay; - } - - let multiplier = self.backoff_factor.powi(attempt as i32); - let millis = (self.delay.as_millis() as f64 * multiplier) as u64; - - Duration::from_millis(millis) - } - - /// Calculate the next retry time - /// - /// Returns the duration to wait before retrying after the given attempt - pub fn next_retry_at(&self, attempt: u32) -> Option { - if attempt >= self.max_retries { - return None; - } - - Some(self.delay_for_attempt(attempt)) - } - - /// Check if retries are exhausted - pub fn is_exhausted(&self, attempt: u32) -> bool { - attempt >= self.max_retries - } -} - /// Parse a duration string to Duration /// /// Supports common formats: @@ -241,67 +166,6 @@ mod tests { assert!(parse_duration("abc").is_err()); } - #[test] - fn test_retry_policy_creation() { - let policy = RetryPolicy::new(3, Duration::from_secs(30), 2.0); - assert_eq!(policy.max_retries, 3); - assert_eq!(policy.delay, Duration::from_secs(30)); - assert_eq!(policy.backoff_factor, 2.0); - } - - #[test] - fn test_retry_policy_no_retries() { - let policy = RetryPolicy::no_retries(); - assert_eq!(policy.max_retries, 0); - assert!(policy.next_retry_at(0).is_none()); - assert!(policy.is_exhausted(0)); - } - - #[test] - fn test_retry_policy_fixed_delay() { - let policy = RetryPolicy::fixed(3, Duration::from_secs(60)); - assert_eq!(policy.max_retries, 3); - assert_eq!(policy.backoff_factor, 1.0); - - // All attempts should have same delay - assert_eq!(policy.delay_for_attempt(0), Duration::from_secs(60)); - assert_eq!(policy.delay_for_attempt(1), Duration::from_secs(60)); - assert_eq!(policy.delay_for_attempt(2), Duration::from_secs(60)); - } - - #[test] - fn test_retry_policy_exponential_backoff() { - let policy = RetryPolicy::exponential(5, Duration::from_secs(10), 2.0); - - assert_eq!(policy.delay_for_attempt(0), Duration::from_secs(10)); - assert_eq!(policy.delay_for_attempt(1), Duration::from_secs(20)); - assert_eq!(policy.delay_for_attempt(2), Duration::from_secs(40)); - assert_eq!(policy.delay_for_attempt(3), Duration::from_secs(80)); - assert_eq!(policy.delay_for_attempt(4), Duration::from_secs(160)); - } - - #[test] - fn test_retry_policy_next_retry_at() { - let policy = RetryPolicy::fixed(3, Duration::from_secs(30)); - - assert!(policy.next_retry_at(0).is_some()); - assert!(policy.next_retry_at(1).is_some()); - assert!(policy.next_retry_at(2).is_some()); - assert!(policy.next_retry_at(3).is_none()); // Exhausted - assert!(policy.next_retry_at(4).is_none()); - } - - #[test] - fn test_retry_policy_is_exhausted() { - let policy = RetryPolicy::fixed(3, Duration::from_secs(30)); - - assert!(!policy.is_exhausted(0)); - assert!(!policy.is_exhausted(1)); - assert!(!policy.is_exhausted(2)); - assert!(policy.is_exhausted(3)); - assert!(policy.is_exhausted(4)); - } - #[test] fn test_format_duration() { assert_eq!(format_duration(Duration::from_secs(45)), "45s"); diff --git a/conduit-executor/tests/executor_integration_test.rs b/conduit-executor/tests/executor_integration_test.rs index 3d9c107..f1ca154 100644 --- a/conduit-executor/tests/executor_integration_test.rs +++ b/conduit-executor/tests/executor_integration_test.rs @@ -13,6 +13,8 @@ fn make_bash_task(id: &str, command: &str) -> Task { dependencies: vec![], retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, timeout: None, trigger_rule: TriggerRule::AllSuccess, pool: None, diff --git a/conduit-executor/tests/failure_injection_test.rs b/conduit-executor/tests/failure_injection_test.rs index c423d60..4c07fdf 100644 --- a/conduit-executor/tests/failure_injection_test.rs +++ b/conduit-executor/tests/failure_injection_test.rs @@ -24,6 +24,8 @@ fn make_task(id: &str, command: &str, timeout: Option<&str>) -> Task { dependencies: vec![], retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, timeout: timeout.map(String::from), trigger_rule: TriggerRule::AllSuccess, pool: None, diff --git a/conduit-lineage/src/cross_task.rs b/conduit-lineage/src/cross_task.rs index b6b42a9..d231b48 100644 --- a/conduit-lineage/src/cross_task.rs +++ b/conduit-lineage/src/cross_task.rs @@ -283,6 +283,8 @@ mod tests { .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, @@ -318,6 +320,8 @@ mod tests { .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-lineage/src/plan_impact.rs b/conduit-lineage/src/plan_impact.rs index 6234246..a357a11 100644 --- a/conduit-lineage/src/plan_impact.rs +++ b/conduit-lineage/src/plan_impact.rs @@ -378,6 +378,8 @@ mod tests { .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-lineage/tests/cross_task_e2e.rs b/conduit-lineage/tests/cross_task_e2e.rs index 21c7b92..fb12820 100644 --- a/conduit-lineage/tests/cross_task_e2e.rs +++ b/conduit-lineage/tests/cross_task_e2e.rs @@ -33,6 +33,8 @@ fn make_task( .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-lineage/tests/plan_impact_e2e.rs b/conduit-lineage/tests/plan_impact_e2e.rs index b98b42f..5b78964 100644 --- a/conduit-lineage/tests/plan_impact_e2e.rs +++ b/conduit-lineage/tests/plan_impact_e2e.rs @@ -40,6 +40,8 @@ fn py_task(id: &str, outputs: Vec, inputs: Vec, deps: Vec<&str .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, @@ -75,6 +77,8 @@ fn sql_task( .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-planner/src/change_detector.rs b/conduit-planner/src/change_detector.rs index 4799585..237f973 100644 --- a/conduit-planner/src/change_detector.rs +++ b/conduit-planner/src/change_detector.rs @@ -467,6 +467,8 @@ mod tests { .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-planner/src/deployment_plan.rs b/conduit-planner/src/deployment_plan.rs index 187f52e..d6652d4 100644 --- a/conduit-planner/src/deployment_plan.rs +++ b/conduit-planner/src/deployment_plan.rs @@ -707,6 +707,8 @@ mod tests { .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-planner/src/fingerprinter.rs b/conduit-planner/src/fingerprinter.rs index b705921..243230c 100644 --- a/conduit-planner/src/fingerprinter.rs +++ b/conduit-planner/src/fingerprinter.rs @@ -77,7 +77,14 @@ impl PlanFingerprinter { use conduit_common::dag::TaskType; match &task.task_type { TaskType::Python { module, function } => { - format!("python:{}:{}", module, function) + // module/function only name the entry point — the compiler's + // source_hash carries the actual body so edits are detected. + format!( + "python:{}:{}:{}", + module, + function, + task.source_hash.as_deref().unwrap_or("") + ) } TaskType::Bash { command } => { format!("bash:{}", command) @@ -149,6 +156,8 @@ mod tests { .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, @@ -279,4 +288,28 @@ mod tests { assert_ne!(fps1["a"], fps2["a"]); } + + /// A Python task whose only difference is its source_hash (i.e., its + /// body was edited) must produce a different fingerprint. + #[test] + fn python_source_hash_change_changes_fingerprint() { + use conduit_common::dag::TaskType; + let make = |hash: &str| { + let mut task = make_task("a", vec![]); + task.task_type = TaskType::Python { + module: "dags.hello".to_string(), + function: "a".to_string(), + }; + task.source_hash = Some(hash.to_string()); + make_dag("test", vec![task], vec!["a"]) + }; + + let fps1 = PlanFingerprinter::fingerprint_dag(&make("hash_v1")); + let fps2 = PlanFingerprinter::fingerprint_dag(&make("hash_v2")); + + assert_ne!( + fps1["a"], fps2["a"], + "editing a Python task body (source_hash) must change its fingerprint" + ); + } } diff --git a/conduit-planner/src/impact_analyzer.rs b/conduit-planner/src/impact_analyzer.rs index 0ff25e2..b5a56ba 100644 --- a/conduit-planner/src/impact_analyzer.rs +++ b/conduit-planner/src/impact_analyzer.rs @@ -274,6 +274,8 @@ mod tests { .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-planner/tests/proptest_fingerprint.rs b/conduit-planner/tests/proptest_fingerprint.rs index 83ce11a..49fdc4d 100644 --- a/conduit-planner/tests/proptest_fingerprint.rs +++ b/conduit-planner/tests/proptest_fingerprint.rs @@ -26,6 +26,8 @@ fn make_task(id: &str, deps: Vec<&str>, command: &str) -> (String, Task) { .collect(), retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-providers/src/providers/cockroachdb.rs b/conduit-providers/src/providers/cockroachdb.rs index e567dc7..c330775 100644 --- a/conduit-providers/src/providers/cockroachdb.rs +++ b/conduit-providers/src/providers/cockroachdb.rs @@ -217,9 +217,18 @@ impl SqlProvider for CockroachDbProvider { async fn execute( &self, query: &str, - _params: &HashMap, + params: &HashMap, ) -> Result { let query = super::sanitize::sanitize_query(query, &self.name)?; + let (query, bind_values) = super::params::bind_named_params( + &query, + params, + super::params::PlaceholderStyle::Dollar, + ) + .map_err(|reason| ProviderError::QueryFailed { + connection: self.name.clone(), + reason, + })?; let start = std::time::Instant::now(); let pool = self.ensure_pool().await?; @@ -227,7 +236,11 @@ impl SqlProvider for CockroachDbProvider { let is_select = query_upper.starts_with("SELECT") || query_upper.starts_with("WITH"); if is_select { - let rows = sqlx::query(&query).fetch_all(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_postgres(q, v); + } + let rows = q.fetch_all(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), @@ -270,7 +283,11 @@ impl SqlProvider for CockroachDbProvider { metrics, }) } else { - let result = sqlx::query(&query).execute(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_postgres(q, v); + } + let result = q.execute(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), diff --git a/conduit-providers/src/providers/mod.rs b/conduit-providers/src/providers/mod.rs index cc244f9..d58bd00 100644 --- a/conduit-providers/src/providers/mod.rs +++ b/conduit-providers/src/providers/mod.rs @@ -5,6 +5,7 @@ //! [`ConnectionConfig`]. // ── Security Sanitization ──────────────────────────────────────────────── +pub mod params; pub mod sanitize; // ── SQL Providers ───────────────────────────────────────────────────────── diff --git a/conduit-providers/src/providers/mysql.rs b/conduit-providers/src/providers/mysql.rs index cad5db3..6b00cad 100644 --- a/conduit-providers/src/providers/mysql.rs +++ b/conduit-providers/src/providers/mysql.rs @@ -193,9 +193,18 @@ impl SqlProvider for MySqlProvider { async fn execute( &self, query: &str, - _params: &HashMap, + params: &HashMap, ) -> Result { let query = super::sanitize::sanitize_query(query, &self.name)?; + let (query, bind_values) = super::params::bind_named_params( + &query, + params, + super::params::PlaceholderStyle::Question, + ) + .map_err(|reason| ProviderError::QueryFailed { + connection: self.name.clone(), + reason, + })?; let start = std::time::Instant::now(); let pool = self.ensure_pool().await?; @@ -203,7 +212,11 @@ impl SqlProvider for MySqlProvider { let is_select = query_upper.starts_with("SELECT") || query_upper.starts_with("WITH"); if is_select { - let rows = sqlx::query(&query).fetch_all(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_mysql(q, v); + } + let rows = q.fetch_all(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), @@ -246,7 +259,11 @@ impl SqlProvider for MySqlProvider { metrics, }) } else { - let result = sqlx::query(&query).execute(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_mysql(q, v); + } + let result = q.execute(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), diff --git a/conduit-providers/src/providers/params.rs b/conduit-providers/src/providers/params.rs new file mode 100644 index 0000000..99d6ba0 --- /dev/null +++ b/conduit-providers/src/providers/params.rs @@ -0,0 +1,264 @@ +//! Named-parameter rewriting for SQL providers. +//! +//! User task SQL uses `:name` placeholders (the style the SDK documents). +//! Each provider rewrites those to its native placeholder syntax and binds +//! the values through sqlx — parameters are data, never string-spliced SQL. + +use std::collections::HashMap; + +/// Native placeholder syntax of the target database. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PlaceholderStyle { + /// `$1, $2, …` — PostgreSQL family (Postgres, CockroachDB, Timescale, Redshift). + Dollar, + /// `?` — MySQL and SQLite. + Question, +} + +/// Rewrite `:name` placeholders to the native style and return the ordered +/// values to bind. Skips quoted strings, quoted identifiers, comments, and +/// Postgres `::type` casts. Referencing a param that isn't in `params` is an +/// error; a query with no `:name` placeholders passes through untouched. +pub fn bind_named_params( + query: &str, + params: &HashMap, + style: PlaceholderStyle, +) -> Result<(String, Vec), String> { + let bytes = query.as_bytes(); + let mut out = String::with_capacity(query.len()); + let mut values = Vec::new(); + let mut i = 0; + let n = bytes.len(); + + while i < n { + let c = bytes[i] as char; + + // Single-quoted string literal (with '' escaping). + if c == '\'' { + let start = i; + i += 1; + while i < n { + if bytes[i] == b'\'' { + if i + 1 < n && bytes[i + 1] == b'\'' { + i += 2; + continue; + } + i += 1; + break; + } + i += 1; + } + out.push_str(&query[start..i]); + continue; + } + + // Double-quoted identifier. + if c == '"' { + let start = i; + i += 1; + while i < n && bytes[i] != b'"' { + i += 1; + } + i = (i + 1).min(n); + out.push_str(&query[start..i]); + continue; + } + + // Line comment. + if c == '-' && i + 1 < n && bytes[i + 1] == b'-' { + let start = i; + while i < n && bytes[i] != b'\n' { + i += 1; + } + out.push_str(&query[start..i]); + continue; + } + + // Block comment. + if c == '/' && i + 1 < n && bytes[i + 1] == b'*' { + let start = i; + i += 2; + while i + 1 < n && !(bytes[i] == b'*' && bytes[i + 1] == b'/') { + i += 1; + } + i = (i + 2).min(n); + out.push_str(&query[start..i]); + continue; + } + + // `::` cast — copy both colons verbatim, never a placeholder. + if c == ':' && i + 1 < n && bytes[i + 1] == b':' { + out.push_str("::"); + i += 2; + continue; + } + + // `:name` placeholder (must start with a letter or underscore). + if c == ':' + && i + 1 < n + && ((bytes[i + 1] as char).is_ascii_alphabetic() || bytes[i + 1] == b'_') + { + let mut j = i + 1; + while j < n + && ((bytes[j] as char).is_ascii_alphanumeric() || bytes[j] == b'_') + { + j += 1; + } + let name = &query[i + 1..j]; + let value = params.get(name).ok_or_else(|| { + format!( + "query references :{} but no such parameter was supplied \ + (available: {:?})", + name, + params.keys().collect::>() + ) + })?; + values.push(value.clone()); + match style { + PlaceholderStyle::Dollar => { + out.push_str(&format!("${}", values.len())); + } + PlaceholderStyle::Question => out.push('?'), + } + i = j; + continue; + } + + out.push(c); + i += c.len_utf8(); + } + + Ok((out, values)) +} + +/// Bind a string value with type inference: integers, floats, and booleans +/// bind as their native types (so `WHERE id = :id` works against numeric +/// columns in strictly-typed databases); everything else binds as text. +macro_rules! bind_inferred { + ($fn_name:ident, $db:ty) => { + pub fn $fn_name<'q>( + q: sqlx::query::Query<'q, $db, <$db as sqlx::Database>::Arguments<'q>>, + value: &'q str, + ) -> sqlx::query::Query<'q, $db, <$db as sqlx::Database>::Arguments<'q>> { + if let Ok(i) = value.parse::() { + q.bind(i) + } else if let Ok(f) = value.parse::() { + q.bind(f) + } else if value.eq_ignore_ascii_case("true") { + q.bind(true) + } else if value.eq_ignore_ascii_case("false") { + q.bind(false) + } else { + q.bind(value) + } + } + }; +} + +bind_inferred!(bind_inferred_sqlite, sqlx::Sqlite); +bind_inferred!(bind_inferred_postgres, sqlx::Postgres); +bind_inferred!(bind_inferred_mysql, sqlx::MySql); + +#[cfg(test)] +mod tests { + use super::*; + + fn params(pairs: &[(&str, &str)]) -> HashMap { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect() + } + + #[test] + fn rewrites_to_dollar_placeholders_in_order() { + let (q, v) = bind_named_params( + "SELECT * FROM t WHERE a = :a AND b = :b", + ¶ms(&[("a", "1"), ("b", "x")]), + PlaceholderStyle::Dollar, + ) + .unwrap(); + assert_eq!(q, "SELECT * FROM t WHERE a = $1 AND b = $2"); + assert_eq!(v, vec!["1", "x"]); + } + + #[test] + fn rewrites_to_question_placeholders() { + let (q, v) = bind_named_params( + "SELECT * FROM t WHERE a = :a", + ¶ms(&[("a", "1")]), + PlaceholderStyle::Question, + ) + .unwrap(); + assert_eq!(q, "SELECT * FROM t WHERE a = ?"); + assert_eq!(v, vec!["1"]); + } + + #[test] + fn repeated_name_binds_each_occurrence() { + let (q, v) = bind_named_params( + "SELECT :x AS a, :x AS b", + ¶ms(&[("x", "7")]), + PlaceholderStyle::Dollar, + ) + .unwrap(); + assert_eq!(q, "SELECT $1 AS a, $2 AS b"); + assert_eq!(v, vec!["7", "7"]); + } + + #[test] + fn postgres_cast_is_not_a_placeholder() { + let (q, v) = bind_named_params( + "SELECT '5'::int, x::text FROM t WHERE a = :a", + ¶ms(&[("a", "1")]), + PlaceholderStyle::Dollar, + ) + .unwrap(); + assert_eq!(q, "SELECT '5'::int, x::text FROM t WHERE a = $1"); + assert_eq!(v, vec!["1"]); + } + + #[test] + fn colons_inside_strings_comments_and_identifiers_are_untouched() { + let src = "SELECT ':not_me', \":also_not\" -- :nope\n/* :nor_this */ FROM t WHERE a = :a"; + let (q, v) = + bind_named_params(src, ¶ms(&[("a", "1")]), PlaceholderStyle::Dollar).unwrap(); + assert!(q.contains("':not_me'")); + assert!(q.contains("\":also_not\"")); + assert!(q.contains("-- :nope")); + assert!(q.contains("/* :nor_this */")); + assert!(q.ends_with("WHERE a = $1")); + assert_eq!(v, vec!["1"]); + } + + #[test] + fn missing_param_is_an_error() { + let err = bind_named_params( + "SELECT * FROM t WHERE a = :ghost", + ¶ms(&[]), + PlaceholderStyle::Dollar, + ); + assert!(err.is_err()); + assert!(err.unwrap_err().contains(":ghost")); + } + + #[test] + fn query_without_placeholders_passes_through() { + let src = "SELECT * FROM t WHERE ts < now()"; + let (q, v) = bind_named_params(src, ¶ms(&[]), PlaceholderStyle::Dollar).unwrap(); + assert_eq!(q, src); + assert!(v.is_empty()); + } + + #[test] + fn array_slice_syntax_is_untouched() { + let (q, v) = bind_named_params( + "SELECT arr[1:3] FROM t", + ¶ms(&[]), + PlaceholderStyle::Dollar, + ) + .unwrap(); + assert_eq!(q, "SELECT arr[1:3] FROM t"); + assert!(v.is_empty()); + } +} diff --git a/conduit-providers/src/providers/postgres.rs b/conduit-providers/src/providers/postgres.rs index 93848ab..a8af5c1 100644 --- a/conduit-providers/src/providers/postgres.rs +++ b/conduit-providers/src/providers/postgres.rs @@ -217,9 +217,18 @@ impl SqlProvider for PostgresProvider { async fn execute( &self, query: &str, - _params: &HashMap, + params: &HashMap, ) -> Result { let query = super::sanitize::sanitize_query(query, &self.name)?; + let (query, bind_values) = super::params::bind_named_params( + &query, + params, + super::params::PlaceholderStyle::Dollar, + ) + .map_err(|reason| ProviderError::QueryFailed { + connection: self.name.clone(), + reason, + })?; let start = std::time::Instant::now(); let pool = self.ensure_pool().await?; @@ -227,7 +236,11 @@ impl SqlProvider for PostgresProvider { let is_select = query_upper.starts_with("SELECT") || query_upper.starts_with("WITH"); if is_select { - let rows = sqlx::query(&query).fetch_all(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_postgres(q, v); + } + let rows = q.fetch_all(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), @@ -270,7 +283,11 @@ impl SqlProvider for PostgresProvider { metrics, }) } else { - let result = sqlx::query(&query).execute(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_postgres(q, v); + } + let result = q.execute(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), diff --git a/conduit-providers/src/providers/redshift.rs b/conduit-providers/src/providers/redshift.rs index ad36675..2b1f04b 100644 --- a/conduit-providers/src/providers/redshift.rs +++ b/conduit-providers/src/providers/redshift.rs @@ -214,9 +214,18 @@ impl SqlProvider for RedshiftProvider { async fn execute( &self, query: &str, - _params: &HashMap, + params: &HashMap, ) -> Result { let query = super::sanitize::sanitize_query(query, &self.name)?; + let (query, bind_values) = super::params::bind_named_params( + &query, + params, + super::params::PlaceholderStyle::Dollar, + ) + .map_err(|reason| ProviderError::QueryFailed { + connection: self.name.clone(), + reason, + })?; let start = std::time::Instant::now(); let pool = self.ensure_pool().await?; @@ -224,7 +233,11 @@ impl SqlProvider for RedshiftProvider { let is_select = query_upper.starts_with("SELECT") || query_upper.starts_with("WITH"); if is_select { - let rows = sqlx::query(&query).fetch_all(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_postgres(q, v); + } + let rows = q.fetch_all(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), @@ -267,7 +280,11 @@ impl SqlProvider for RedshiftProvider { metrics, }) } else { - let result = sqlx::query(&query).execute(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_postgres(q, v); + } + let result = q.execute(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), diff --git a/conduit-providers/src/providers/sqlite.rs b/conduit-providers/src/providers/sqlite.rs index d339b76..ec9777e 100644 --- a/conduit-providers/src/providers/sqlite.rs +++ b/conduit-providers/src/providers/sqlite.rs @@ -187,9 +187,18 @@ impl SqlProvider for SqliteProvider { async fn execute( &self, query: &str, - _params: &HashMap, + params: &HashMap, ) -> Result { let query = super::sanitize::sanitize_query(query, &self.name)?; + let (query, bind_values) = super::params::bind_named_params( + &query, + params, + super::params::PlaceholderStyle::Question, + ) + .map_err(|reason| ProviderError::QueryFailed { + connection: self.name.clone(), + reason, + })?; let start = std::time::Instant::now(); let pool = self.ensure_pool().await?; @@ -197,7 +206,11 @@ impl SqlProvider for SqliteProvider { let is_select = query_upper.starts_with("SELECT") || query_upper.starts_with("WITH"); if is_select { - let rows = sqlx::query(&query).fetch_all(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_sqlite(q, v); + } + let rows = q.fetch_all(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), @@ -240,7 +253,11 @@ impl SqlProvider for SqliteProvider { metrics, }) } else { - let result = sqlx::query(&query).execute(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_sqlite(q, v); + } + let result = q.execute(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), @@ -328,3 +345,89 @@ impl SqlProvider for SqliteProvider { .collect()) } } + +#[cfg(test)] +mod param_binding_tests { + use super::*; + use crate::traits::SqlProvider; + + fn memory_provider() -> SqliteProvider { + let config = ConnectionConfig { + conn_type: "sqlite".to_string(), + host: None, + port: None, + database: Some(":memory:".to_string()), + credentials: None, + extra: HashMap::new(), + }; + SqliteProvider::from_config("test_sqlite", &config).unwrap() + } + + /// `execute` must bind named `:params` as real bind parameters — + /// the params map is not decorative. + #[tokio::test] + async fn execute_binds_named_params() { + let p = memory_provider(); + p.execute_statement("CREATE TABLE t (id INTEGER, name TEXT)") + .await + .unwrap(); + p.execute_statement("INSERT INTO t VALUES (1, 'alice'), (2, 'bob')") + .await + .unwrap(); + + let mut params = HashMap::new(); + params.insert("id".to_string(), "2".to_string()); + let result = p + .execute("SELECT name FROM t WHERE id = :id", ¶ms) + .await + .unwrap(); + + assert_eq!(result.sample_rows.len(), 1, "bound :id must filter to one row"); + assert_eq!(result.sample_rows[0][0], serde_json::json!("bob")); + } + + /// A string param must not open an injection hole: the bound value is + /// data, never SQL. + #[tokio::test] + async fn bound_param_is_not_interpreted_as_sql() { + let p = memory_provider(); + p.execute_statement("CREATE TABLE t (id INTEGER, name TEXT)") + .await + .unwrap(); + p.execute_statement("INSERT INTO t VALUES (1, 'alice')") + .await + .unwrap(); + + let mut params = HashMap::new(); + params.insert( + "name".to_string(), + "alice' OR '1'='1".to_string(), + ); + let result = p + .execute("SELECT id FROM t WHERE name = :name", ¶ms) + .await + .unwrap(); + + assert_eq!( + result.sample_rows.len(), + 0, + "injection payload must be treated as a literal value" + ); + } + + /// Referencing a param that wasn't supplied is a clear error, not a + /// silently-unbound query. + #[tokio::test] + async fn missing_param_is_an_error() { + let p = memory_provider(); + p.execute_statement("CREATE TABLE t (id INTEGER)") + .await + .unwrap(); + + let params = HashMap::new(); + let err = p + .execute("SELECT * FROM t WHERE id = :missing", ¶ms) + .await; + assert!(err.is_err(), "missing param must error, got {:?}", err); + } +} diff --git a/conduit-providers/src/providers/timescaledb.rs b/conduit-providers/src/providers/timescaledb.rs index dc22304..674eba5 100644 --- a/conduit-providers/src/providers/timescaledb.rs +++ b/conduit-providers/src/providers/timescaledb.rs @@ -217,9 +217,18 @@ impl SqlProvider for TimescaleDbProvider { async fn execute( &self, query: &str, - _params: &HashMap, + params: &HashMap, ) -> Result { let query = super::sanitize::sanitize_query(query, &self.name)?; + let (query, bind_values) = super::params::bind_named_params( + &query, + params, + super::params::PlaceholderStyle::Dollar, + ) + .map_err(|reason| ProviderError::QueryFailed { + connection: self.name.clone(), + reason, + })?; let start = std::time::Instant::now(); let pool = self.ensure_pool().await?; @@ -227,7 +236,11 @@ impl SqlProvider for TimescaleDbProvider { let is_select = query_upper.starts_with("SELECT") || query_upper.starts_with("WITH"); if is_select { - let rows = sqlx::query(&query).fetch_all(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_postgres(q, v); + } + let rows = q.fetch_all(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), @@ -270,7 +283,11 @@ impl SqlProvider for TimescaleDbProvider { metrics, }) } else { - let result = sqlx::query(&query).execute(pool).await.map_err(|e| { + let mut q = sqlx::query(&query); + for v in &bind_values { + q = super::params::bind_inferred_postgres(q, v); + } + let result = q.execute(pool).await.map_err(|e| { ProviderError::QueryFailed { connection: self.name.clone(), reason: super::sanitize::sanitize_error(&e.to_string()), diff --git a/conduit-scheduler/src/pool_manager.rs b/conduit-scheduler/src/pool_manager.rs index f76798e..2b440b7 100644 --- a/conduit-scheduler/src/pool_manager.rs +++ b/conduit-scheduler/src/pool_manager.rs @@ -66,13 +66,30 @@ impl PoolManager { /// Try to acquire a slot in the named pool for the given task. /// /// Returns `true` if a slot was acquired, `false` if no slots available. + /// + /// A pool that was never defined is auto-registered as unlimited (with a + /// warning): tasks referencing an undeclared pool must run, not deadlock. pub fn acquire(&mut self, pool_name: &str, task_id: &str) -> bool { + if !self.pools.contains_key(pool_name) { + warn!( + pool = %pool_name, + task = %task_id, + "Pool not defined in config; treating as unlimited" + ); + self.pools.insert( + pool_name.to_string(), + PoolState { + name: pool_name.to_string(), + total_slots: u32::MAX, + available_slots: u32::MAX, + description: None, + occupants: HashSet::new(), + }, + ); + } let pool = match self.pools.get_mut(pool_name) { Some(p) => p, - None => { - warn!(pool = %pool_name, task = %task_id, "Pool not found"); - return false; - } + None => unreachable!("pool was just inserted"), }; if pool.available_slots > 0 { @@ -271,15 +288,18 @@ mod tests { } #[test] - fn test_nonexistent_pool() { + fn test_nonexistent_pool_is_treated_as_unlimited() { + // An undeclared pool must not block (or deadlock) tasks that + // reference it — it is auto-registered as unlimited with a warning. let pools = vec![]; let mut mgr = PoolManager::new(pools); - assert!(!mgr.acquire("nonexistent", "task_1")); - assert_eq!(mgr.available("nonexistent"), 0); - assert_eq!(mgr.total("nonexistent"), 0); + assert!(mgr.acquire("nonexistent", "task_1")); + assert!(mgr.acquire("nonexistent", "task_2")); + assert_eq!(mgr.total("nonexistent"), u32::MAX); mgr.release("nonexistent", "task_1"); + mgr.release("nonexistent", "task_2"); } #[test] diff --git a/conduit-scheduler/src/scheduler.rs b/conduit-scheduler/src/scheduler.rs index 75825d3..c5d13b1 100644 --- a/conduit-scheduler/src/scheduler.rs +++ b/conduit-scheduler/src/scheduler.rs @@ -17,6 +17,7 @@ use tracing::{debug, error, info, warn}; use conduit_common::dag::{Dag, DagId, TaskId, TriggerRule}; use conduit_common::error::ConduitResult; +use conduit_common::event::{EventKind, RunStatus as EventRunStatus}; use conduit_common::metrics; use conduit_state::EventStore; @@ -57,6 +58,13 @@ pub enum SchedulerEvent { sensor_id: String, payload: HashMap, }, + /// A task's retry delay has elapsed; re-dispatch it. Produced internally + /// by the scheduler's own retry timers, never by external callers. + TaskRetryReady { + dag_id: DagId, + run_id: String, + task_id: TaskId, + }, /// Graceful shutdown signal. Shutdown, } @@ -153,13 +161,21 @@ pub enum TaskState { pub struct Scheduler { event_rx: mpsc::UnboundedReceiver, command_tx: mpsc::UnboundedSender, + /// Internal wake channel for the scheduler's own timers (retry delays). + /// Kept separate from `event_rx` so retry wakeups work even when the + /// external event sender has been cloned/dropped by the embedding app. + self_tx: mpsc::UnboundedSender, + self_rx: mpsc::UnboundedReceiver, dag_runs: HashMap, - #[allow(dead_code)] pools: PoolManager, plans: HashMap, cron_schedules: HashMap, /// Optional persistent event store for catchup queries. event_store: Option>, + /// Last minute (unix_ts / 60) each DAG was cron-fired. Five-field cron + /// has minute resolution; this guard makes ticks idempotent within a + /// minute so the tick source's frequency can't create duplicate runs. + last_cron_fire: HashMap, /// Alert hooks fired on every non-success DAG-run completion. Each hook /// is spawned on the tokio runtime so a slow notifier (PagerDuty, Slack) /// can't stall the scheduler event loop. Empty by default. @@ -193,14 +209,19 @@ impl Scheduler { } } + let (self_tx, self_rx) = mpsc::unbounded_channel(); + Ok(Self { event_rx, command_tx, + self_tx, + self_rx, dag_runs: HashMap::new(), pools, plans, cron_schedules, event_store: None, + last_cron_fire: HashMap::new(), alert_hooks: Vec::new(), }) } @@ -261,6 +282,55 @@ impl Scheduler { self } + /// Release the pool slot a task held, if it draws from a pool. + /// Returns true when a slot was actually freed. + fn release_pool_slot(&mut self, dag_id: &DagId, run_key: &str, task_id: &TaskId) -> bool { + let Some(pool) = self + .plans + .get(dag_id) + .and_then(|d| d.tasks.get(task_id)) + .and_then(|t| t.pool.clone()) + else { + return false; + }; + self.pools + .release(&pool, &format!("{}/{}", run_key, task_id)); + true + } + + /// Re-evaluate every active run and dispatch newly-eligible tasks. + /// Called after a pool slot frees up: the waiter may live in a + /// different run than the task that released the slot. + fn dispatch_waiting_runs(&mut self) { + let runs: Vec<(DagId, String)> = self + .dag_runs + .values() + .map(|rs| (rs.dag_id.clone(), format!("{}/{}", rs.dag_id, rs.run_id))) + .collect(); + for (dag_id, run_key) in runs { + let ready = { + let (Some(dag), Some(rs)) = + (self.plans.get(&dag_id), self.dag_runs.get(&run_key)) + else { + continue; + }; + self.evaluate_ready_tasks(dag, rs) + }; + self.transition_and_dispatch(&dag_id, &run_key, ready); + } + } + + /// Best-effort append to the attached event store. Persistence powers + /// `conduit replay` and post-hoc debugging; it must never stall or fail + /// scheduling, so errors are logged and swallowed. + fn persist_event(&self, kind: EventKind) { + if let Some(store) = &self.event_store { + if let Err(e) = store.append(kind) { + warn!(error = %e, "Failed to persist scheduler event"); + } + } + } + /// Send a command to the executor, logging and recording metrics on failure. fn send_command(&self, cmd: SchedulerCommand) { if let Err(e) = self.command_tx.send(cmd) { @@ -353,7 +423,21 @@ impl Scheduler { // Perform catchup on startup before processing new events self.perform_catchup(); - while let Some(event) = self.event_rx.recv().await { + loop { + // Merge external events with the scheduler's own timer wakeups. + // `self_rx` can never yield None while `self` holds `self_tx`; + // a closed external channel ends the loop like it always did. + let event = tokio::select! { + ev = self.event_rx.recv() => match ev { + Some(e) => e, + None => break, + }, + ev = self.self_rx.recv() => match ev { + Some(e) => e, + None => break, + }, + }; + // Record scheduler event metric if let Some(m) = metrics::try_global() { let event_type = match &event { @@ -362,6 +446,7 @@ impl Scheduler { SchedulerEvent::TaskFailed { .. } => "task_failed", SchedulerEvent::CronTick { .. } => "cron_tick", SchedulerEvent::SensorTriggered { .. } => "sensor_triggered", + SchedulerEvent::TaskRetryReady { .. } => "task_retry_ready", SchedulerEvent::Shutdown => "shutdown", }; m.scheduler_events_total @@ -413,6 +498,13 @@ impl Scheduler { SchedulerEvent::SensorTriggered { sensor_id, payload } => { self.handle_sensor_triggered(&sensor_id, payload); } + SchedulerEvent::TaskRetryReady { + dag_id, + run_id, + task_id, + } => { + self.handle_task_retry_ready(&dag_id, &run_id, &task_id); + } SchedulerEvent::Shutdown => { info!("Scheduler shutdown requested"); break; @@ -466,6 +558,28 @@ impl Scheduler { "DAG run created" ); + // Callers can pass environment/triggered_by via run config. + let (environment, triggered_by) = { + let config = &self.dag_runs[&run_key].config; + ( + config + .get("environment") + .cloned() + .unwrap_or_else(|| "production".to_string()), + config + .get("triggered_by") + .cloned() + .unwrap_or_else(|| "scheduler".to_string()), + ) + }; + self.persist_event(EventKind::DagRunCreated { + dag_id: dag_id.clone(), + run_id: run_id.to_string(), + logical_date, + environment, + triggered_by, + }); + // Record metrics if let Some(m) = metrics::try_global() { m.dag_runs_total @@ -498,6 +612,7 @@ impl Scheduler { duration_ms: u64, ) { let run_key = format!("{}/{}", dag_id, run_id); + let snapshot_id_for_event = snapshot_id.clone(); // Mutate first, then drop the mutable borrow. Drop late or duplicate // completion events for already-terminal tasks instead of overwriting @@ -542,6 +657,16 @@ impl Scheduler { "Task completed" ); + self.persist_event(EventKind::TaskCompleted { + dag_id: dag_id.clone(), + run_id: run_id.to_string(), + task_id: task_id.clone(), + duration_ms, + snapshot_id: snapshot_id_for_event, + }); + + let released_pool_slot = self.release_pool_slot(dag_id, &run_key, task_id); + // Record metrics if let Some(m) = metrics::try_global() { m.record_task_event(dag_id, task_id, "completed"); @@ -566,6 +691,11 @@ impl Scheduler { self.transition_and_dispatch(dag_id, &run_key, ready); + // A freed pool slot may unblock tasks in other runs. + if released_pool_slot { + self.dispatch_waiting_runs(); + } + // Re-borrow to check completion after dispatch state mutations. if let (Some(dag), Some(rs)) = (self.plans.get(dag_id), self.dag_runs.get(&run_key)) { self.check_dag_run_complete(dag, rs); @@ -588,7 +718,11 @@ impl Scheduler { Some(dag) => match dag.tasks.get(task_id) { Some(task) => { if attempt < task.retries { - (true, parse_duration(&task.retry_delay)) + let base = parse_duration(&task.retry_delay); + ( + true, + retry_delay_for_attempt(base, attempt, task.retry_backoff), + ) } else { (false, Duration::zero()) } @@ -634,6 +768,12 @@ impl Scheduler { } } + // The failed attempt no longer occupies its pool slot. A retry will + // re-acquire when it is re-dispatched. + if self.release_pool_slot(dag_id, &run_key, task_id) { + self.dispatch_waiting_runs(); + } + if should_retry { debug!( dag_id = %dag_id, @@ -643,6 +783,14 @@ impl Scheduler { "Task will be retried" ); + self.persist_event(EventKind::TaskRetrying { + dag_id: dag_id.clone(), + run_id: run_id.to_string(), + task_id: task_id.clone(), + attempt: attempt + 1, + next_retry_at: Utc::now() + retry_delay, + }); + if let Some(m) = metrics::try_global() { m.record_task_event(dag_id, task_id, "retried"); } @@ -653,6 +801,23 @@ impl Scheduler { task_id: task_id.clone(), delay: retry_delay, }); + + // Arm the retry timer: once the delay elapses, wake the event + // loop via the internal channel and re-dispatch the task. The + // RetryTask command above is a notification for executors/UIs; + // this timer is what actually makes the retry happen. + let self_tx = self.self_tx.clone(); + let (wake_dag, wake_run, wake_task) = + (dag_id.clone(), run_id.to_string(), task_id.clone()); + let sleep_for = retry_delay.to_std().unwrap_or_default(); + tokio::spawn(async move { + tokio::time::sleep(sleep_for).await; + let _ = self_tx.send(SchedulerEvent::TaskRetryReady { + dag_id: wake_dag, + run_id: wake_run, + task_id: wake_task, + }); + }); } else { warn!( dag_id = %dag_id, @@ -662,6 +827,15 @@ impl Scheduler { "Task failed with no retries remaining" ); + self.persist_event(EventKind::TaskFailed { + dag_id: dag_id.clone(), + run_id: run_id.to_string(), + task_id: task_id.clone(), + error: error.clone(), + traceback: None, + attempt, + }); + if let Some(m) = metrics::try_global() { m.record_task_event(dag_id, task_id, "failed"); m.active_tasks.dec(); @@ -674,17 +848,105 @@ impl Scheduler { } } + /// Re-dispatch a task whose retry delay has elapsed. + /// + /// Idempotent: only acts if the task is still `Retrying` — a duplicate + /// or stale wakeup (run finished, task re-resolved some other way) is + /// silently ignored. + fn handle_task_retry_ready(&mut self, dag_id: &DagId, run_id: &str, task_id: &TaskId) { + let run_key = format!("{}/{}", dag_id, run_id); + + let attempt = match self + .dag_runs + .get(&run_key) + .and_then(|rs| rs.task_states.get(task_id)) + { + Some(TaskState::Retrying { attempt, .. }) => *attempt, + _ => { + debug!( + dag_id = %dag_id, + run_id = %run_id, + task_id = %task_id, + "Ignoring stale retry wakeup — task no longer Retrying" + ); + return; + } + }; + + // Pooled retries re-acquire their slot; if the pool is full, check + // again shortly rather than stealing or deadlocking. + let pool = self + .plans + .get(dag_id) + .and_then(|d| d.tasks.get(task_id)) + .and_then(|t| t.pool.clone()); + if let Some(p) = &pool { + if !self.pools.acquire(p, &format!("{}/{}", run_key, task_id)) { + debug!( + dag_id = %dag_id, + task_id = %task_id, + pool = %p, + "Pool full at retry time — re-checking in 1s" + ); + let self_tx = self.self_tx.clone(); + let (wake_dag, wake_run, wake_task) = + (dag_id.clone(), run_id.to_string(), task_id.clone()); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + let _ = self_tx.send(SchedulerEvent::TaskRetryReady { + dag_id: wake_dag, + run_id: wake_run, + task_id: wake_task, + }); + }); + return; + } + } + + if let Some(rs) = self.dag_runs.get_mut(&run_key) { + rs.task_states.insert(task_id.clone(), TaskState::Queued); + } + + info!( + dag_id = %dag_id, + run_id = %run_id, + task_id = %task_id, + attempt = %attempt, + "Retry delay elapsed — re-dispatching task" + ); + + if let Some(m) = metrics::try_global() { + m.record_task_event(dag_id, task_id, "dispatched"); + m.active_tasks.inc(); + } + + self.send_command(SchedulerCommand::DispatchTask { + dag_id: dag_id.clone(), + run_id: run_id.to_string(), + task_id: task_id.clone(), + attempt, + }); + } + /// Handle periodic cron tick. fn handle_cron_tick(&mut self, timestamp: DateTime) { - // Collect due DAGs first to avoid borrowing self immutably while mutating + let minute = timestamp.timestamp() / 60; + + // Collect due DAGs first to avoid borrowing self immutably while + // mutating. Skip DAGs already fired this minute — `is_due` matches + // the whole minute, so ticks arriving more frequently than 1/min + // must not create duplicate runs. let due_dags: Vec = self .cron_schedules .iter() - .filter(|(_, cron)| cron.is_due(timestamp)) + .filter(|(dag_id, cron)| { + cron.is_due(timestamp) && self.last_cron_fire.get(*dag_id) != Some(&minute) + }) .map(|(dag_id, _)| dag_id.clone()) .collect(); for dag_id in due_dags { + self.last_cron_fire.insert(dag_id.clone(), minute); let run_id = format!("{}_{}", dag_id, timestamp.timestamp()); info!( @@ -778,6 +1040,18 @@ impl Scheduler { "DAG run completed" ); + self.persist_event(EventKind::DagRunCompleted { + dag_id: dag.id.clone(), + run_id: run_state.run_id.clone(), + status: match status { + RunStatus::Success => EventRunStatus::Success, + RunStatus::Failed => EventRunStatus::Failed, + RunStatus::Cancelled => EventRunStatus::Cancelled, + }, + duration_ms: (Utc::now() - run_state.started_at).num_milliseconds().max(0) + as u64, + }); + // Record metrics if let Some(m) = metrics::try_global() { let status_str = match status { @@ -914,6 +1188,20 @@ impl Scheduler { // Send commands after releasing the mutable borrow. for cmd in commands_to_send { + if let SchedulerCommand::SkipTask { + dag_id, + run_id, + task_id, + reason, + } = &cmd + { + self.persist_event(EventKind::TaskSkipped { + dag_id: dag_id.clone(), + run_id: run_id.clone(), + task_id: task_id.clone(), + reason: reason.clone(), + }); + } self.send_command(cmd); } @@ -941,22 +1229,68 @@ impl Scheduler { return; } + // Pool gate: a task in a full pool stays Pending and is re-elected + // on a later event (a pooled task completing releases its slot and + // re-evaluates all runs). + let mut dispatchable = Vec::new(); + for task_id in ready { + let pool = self + .plans + .get(dag_id) + .and_then(|d| d.tasks.get(&task_id)) + .and_then(|t| t.pool.clone()); + let acquired = match &pool { + Some(p) => self + .pools + .acquire(p, &format!("{}/{}", run_key, task_id)), + None => true, + }; + if acquired { + dispatchable.push(task_id); + } else { + debug!( + dag_id = %dag_id, + task_id = %task_id, + pool = %pool.as_deref().unwrap_or(""), + "Pool full — task waits for a slot" + ); + } + } + if dispatchable.is_empty() { + return; + } + let run_id = { let Some(rs) = self.dag_runs.get_mut(run_key) else { return; }; - for task_id in &ready { + for task_id in &dispatchable { rs.task_states.insert(task_id.clone(), TaskState::Queued); } rs.run_id.clone() }; - for task_id in ready { + for task_id in dispatchable { if let Some(m) = metrics::try_global() { m.record_task_event(dag_id, &task_id, "dispatched"); m.active_tasks.inc(); } + let (priority, pool) = self + .plans + .get(dag_id) + .and_then(|d| d.tasks.get(&task_id)) + .map(|t| (t.priority, t.pool.clone())) + .unwrap_or((0, None)); + self.persist_event(EventKind::TaskQueued { + dag_id: dag_id.clone(), + run_id: run_id.clone(), + task_id: task_id.clone(), + priority, + pool, + snapshot_fingerprint: None, + }); + self.send_command(SchedulerCommand::DispatchTask { dag_id: dag_id.clone(), run_id: run_id.clone(), @@ -974,6 +1308,26 @@ impl Scheduler { } } +/// Compute the retry delay for a given attempt. +/// +/// Without a backoff multiplier (or with a multiplier <= 1.0) the delay is +/// fixed at `base` for every attempt. With a multiplier > 1.0 the delay +/// grows exponentially: `base * backoff^attempt`, capped at one day so a +/// large attempt count can never overflow or park a task for years. +fn retry_delay_for_attempt(base: Duration, attempt: u32, backoff: Option) -> Duration { + const MAX_DELAY_SECS: f64 = 86_400.0; // 1 day + + let factor = match backoff { + Some(b) if b > 1.0 => b, + _ => return base, + }; + + let base_secs = base.num_milliseconds() as f64 / 1000.0; + let scaled = base_secs * factor.powi(attempt.min(1_000) as i32); + let capped = scaled.min(MAX_DELAY_SECS); + Duration::milliseconds((capped * 1000.0) as i64) +} + /// Parse a duration string (e.g., "5m", "30s", "1h"). fn parse_duration(duration_str: &Option) -> Duration { match duration_str { @@ -1021,4 +1375,45 @@ mod tests { fn test_parse_duration_default() { assert_eq!(parse_duration(&None), Duration::minutes(5)); } + + #[test] + fn retry_delay_without_backoff_is_fixed_across_attempts() { + let base = Duration::seconds(10); + assert_eq!(retry_delay_for_attempt(base, 0, None), base); + assert_eq!(retry_delay_for_attempt(base, 3, None), base); + } + + #[test] + fn retry_delay_with_backoff_grows_exponentially() { + let base = Duration::seconds(10); + assert_eq!( + retry_delay_for_attempt(base, 0, Some(2.0)), + Duration::seconds(10) + ); + assert_eq!( + retry_delay_for_attempt(base, 1, Some(2.0)), + Duration::seconds(20) + ); + assert_eq!( + retry_delay_for_attempt(base, 2, Some(2.0)), + Duration::seconds(40) + ); + } + + #[test] + fn retry_delay_with_backoff_is_capped() { + // A huge attempt count must not overflow — capped at 1 day. + let base = Duration::seconds(60); + assert_eq!( + retry_delay_for_attempt(base, 1000, Some(10.0)), + Duration::days(1) + ); + } + + #[test] + fn retry_delay_backoff_of_one_or_less_behaves_as_fixed() { + let base = Duration::seconds(10); + assert_eq!(retry_delay_for_attempt(base, 5, Some(1.0)), base); + assert_eq!(retry_delay_for_attempt(base, 5, Some(0.5)), base); + } } diff --git a/conduit-scheduler/src/trigger.rs b/conduit-scheduler/src/trigger.rs index 7617111..1f7ae85 100644 --- a/conduit-scheduler/src/trigger.rs +++ b/conduit-scheduler/src/trigger.rs @@ -175,6 +175,8 @@ mod tests { dependencies, retries: 0, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-scheduler/tests/proptest_scheduler.rs b/conduit-scheduler/tests/proptest_scheduler.rs index 65584f0..1b645f4 100644 --- a/conduit-scheduler/tests/proptest_scheduler.rs +++ b/conduit-scheduler/tests/proptest_scheduler.rs @@ -32,6 +32,8 @@ fn task(id: &str, deps: &[&str], retries: u32) -> Task { .collect(), retries, retry_delay: None, + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, diff --git a/conduit-scheduler/tests/scheduler_integration_test.rs b/conduit-scheduler/tests/scheduler_integration_test.rs index 6d78f11..fb265a1 100644 --- a/conduit-scheduler/tests/scheduler_integration_test.rs +++ b/conduit-scheduler/tests/scheduler_integration_test.rs @@ -39,6 +39,8 @@ fn make_task( .collect(), retries, retry_delay: retry_delay.map(|s| s.to_string()), + retry_backoff: None, + source_hash: None, pool: None, timeout: None, priority: 0, @@ -659,3 +661,535 @@ async fn alert_hook_does_not_fire_on_success() { "hook must not fire for a successful run" ); } + +// --------------------------------------------------------------------------- +// Retry re-dispatch (the scheduler must act on its own retry decision) +// --------------------------------------------------------------------------- + +/// Await the next command, panicking after `secs` seconds. +async fn next_command( + rx: &mut mpsc::UnboundedReceiver, + secs: u64, + context: &str, +) -> SchedulerCommand { + tokio::time::timeout(std::time::Duration::from_secs(secs), rx.recv()) + .await + .unwrap_or_else(|_| panic!("timed out waiting for command: {}", context)) + .unwrap_or_else(|| panic!("command channel closed while waiting for: {}", context)) +} + +/// A failing task with retries must be re-dispatched after its retry delay, +/// and the run must complete once the retried attempt succeeds. +#[tokio::test] +async fn failed_task_with_retries_is_redispatched_and_run_completes() { + let task_a = make_task("A", vec![], TriggerRule::AllSuccess, 1, Some("1s")); + let dag = make_dag("dag1", vec![task_a], vec!["A"]); + + let mut plans = HashMap::new(); + plans.insert("dag1".to_string(), dag); + + let (tx, mut rx, scheduler_fut) = create_test_scheduler(plans); + let driver = tokio::spawn(scheduler_fut); + + tx.send(SchedulerEvent::DagRunRequested { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + logical_date: Utc::now(), + config: HashMap::new(), + }) + .unwrap(); + + // Initial dispatch, attempt 0. + let cmd = next_command(&mut rx, 5, "initial DispatchTask for A").await; + assert!( + matches!(&cmd, SchedulerCommand::DispatchTask { task_id, attempt: 0, .. } if task_id == "A"), + "expected initial DispatchTask for A attempt 0, got: {:?}", + cmd + ); + + // The task fails on attempt 0 — one retry remains. + tx.send(SchedulerEvent::TaskFailed { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + task_id: "A".to_string(), + error: "boom".to_string(), + attempt: 0, + }) + .unwrap(); + + // The scheduler announces the retry... + let cmd = next_command(&mut rx, 5, "RetryTask notification for A").await; + assert!( + matches!(&cmd, SchedulerCommand::RetryTask { task_id, .. } if task_id == "A"), + "expected RetryTask for A, got: {:?}", + cmd + ); + + // ...and after the delay elapses it must RE-DISPATCH the task itself. + let cmd = next_command(&mut rx, 5, "re-dispatch of A after retry delay").await; + assert!( + matches!(&cmd, SchedulerCommand::DispatchTask { task_id, attempt: 1, .. } if task_id == "A"), + "expected re-dispatched DispatchTask for A attempt 1, got: {:?}", + cmd + ); + + // The retried attempt succeeds — the run must complete Success. + tx.send(SchedulerEvent::TaskCompleted { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + task_id: "A".to_string(), + snapshot_id: None, + duration_ms: 5, + }) + .unwrap(); + + let cmd = next_command(&mut rx, 5, "CompleteDagRun after retried success").await; + assert!( + matches!( + &cmd, + SchedulerCommand::CompleteDagRun { + status: conduit_scheduler::RunStatus::Success, + .. + } + ), + "expected CompleteDagRun Success, got: {:?}", + cmd + ); + + tx.send(SchedulerEvent::Shutdown).unwrap(); + let _ = driver.await; +} + +/// When retries are exhausted the run must complete Failed (no hang). +#[tokio::test] +async fn exhausted_retries_complete_the_run_as_failed() { + let task_a = make_task("A", vec![], TriggerRule::AllSuccess, 1, Some("1s")); + let dag = make_dag("dag1", vec![task_a], vec!["A"]); + + let mut plans = HashMap::new(); + plans.insert("dag1".to_string(), dag); + + let (tx, mut rx, scheduler_fut) = create_test_scheduler(plans); + let driver = tokio::spawn(scheduler_fut); + + tx.send(SchedulerEvent::DagRunRequested { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + logical_date: Utc::now(), + config: HashMap::new(), + }) + .unwrap(); + + let _ = next_command(&mut rx, 5, "initial DispatchTask for A").await; + tx.send(SchedulerEvent::TaskFailed { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + task_id: "A".to_string(), + error: "boom".to_string(), + attempt: 0, + }) + .unwrap(); + + let _ = next_command(&mut rx, 5, "RetryTask notification").await; + let cmd = next_command(&mut rx, 5, "re-dispatch attempt 1").await; + assert!( + matches!(&cmd, SchedulerCommand::DispatchTask { attempt: 1, .. }), + "expected re-dispatch attempt 1, got: {:?}", + cmd + ); + + // Fail the retried attempt too — retries are now exhausted. + tx.send(SchedulerEvent::TaskFailed { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + task_id: "A".to_string(), + error: "boom again".to_string(), + attempt: 1, + }) + .unwrap(); + + let cmd = next_command(&mut rx, 5, "CompleteDagRun Failed").await; + assert!( + matches!( + &cmd, + SchedulerCommand::CompleteDagRun { + status: conduit_scheduler::RunStatus::Failed, + .. + } + ), + "expected CompleteDagRun Failed after exhausted retries, got: {:?}", + cmd + ); + + tx.send(SchedulerEvent::Shutdown).unwrap(); + let _ = driver.await; +} + +/// With a retry_backoff multiplier, each successive retry's announced delay +/// must grow exponentially (1s, then 2s for backoff=2.0). +#[tokio::test] +async fn retry_backoff_multiplier_grows_the_delay() { + let mut task_a = make_task("A", vec![], TriggerRule::AllSuccess, 2, Some("1s")); + task_a.retry_backoff = Some(2.0); + let dag = make_dag("dag1", vec![task_a], vec!["A"]); + + let mut plans = HashMap::new(); + plans.insert("dag1".to_string(), dag); + + let (tx, mut rx, scheduler_fut) = create_test_scheduler(plans); + let driver = tokio::spawn(scheduler_fut); + + tx.send(SchedulerEvent::DagRunRequested { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + logical_date: Utc::now(), + config: HashMap::new(), + }) + .unwrap(); + + let _ = next_command(&mut rx, 5, "initial DispatchTask").await; + + let fail = |attempt: u32| SchedulerEvent::TaskFailed { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + task_id: "A".to_string(), + error: "boom".to_string(), + attempt, + }; + + // Attempt 0 fails → first retry delay = base = 1s. + tx.send(fail(0)).unwrap(); + let cmd = next_command(&mut rx, 5, "first RetryTask").await; + match &cmd { + SchedulerCommand::RetryTask { delay, .. } => { + assert_eq!(delay.num_seconds(), 1, "first retry delay must be 1s (base)") + } + other => panic!("expected RetryTask, got {:?}", other), + } + let _ = next_command(&mut rx, 5, "re-dispatch attempt 1").await; + + // Attempt 1 fails → second retry delay = base * 2.0 = 2s. + tx.send(fail(1)).unwrap(); + let cmd = next_command(&mut rx, 5, "second RetryTask").await; + match &cmd { + SchedulerCommand::RetryTask { delay, .. } => { + assert_eq!( + delay.num_seconds(), + 2, + "second retry delay must double with backoff=2.0" + ) + } + other => panic!("expected RetryTask, got {:?}", other), + } + + tx.send(SchedulerEvent::Shutdown).unwrap(); + let _ = driver.await; +} + +// --------------------------------------------------------------------------- +// Event persistence (event sourcing must actually record run history) +// --------------------------------------------------------------------------- + +/// With an event store attached, a completed run must leave a persistent +/// TaskCompleted + DagRunCompleted trail that `conduit replay` can fold. +#[tokio::test] +async fn scheduler_persists_run_lifecycle_events() { + use std::sync::Arc; + + let tmp = tempfile::tempdir().unwrap(); + let store = Arc::new(conduit_state::EventStore::open(tmp.path()).unwrap()); + + let task_a = make_task("A", vec![], TriggerRule::AllSuccess, 0, None); + let dag = make_dag("dag1", vec![task_a], vec!["A"]); + let mut plans = HashMap::new(); + plans.insert("dag1".to_string(), dag); + + let (event_tx, event_rx) = mpsc::unbounded_channel(); + let (command_tx, mut command_rx) = mpsc::unbounded_channel(); + let scheduler = Scheduler::new(event_rx, command_tx, PoolManager::new(vec![]), plans) + .unwrap() + .with_event_store(Arc::clone(&store)); + let driver = tokio::spawn(async move { + let _ = scheduler.run().await; + }); + + event_tx + .send(SchedulerEvent::DagRunRequested { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + logical_date: Utc::now(), + config: HashMap::new(), + }) + .unwrap(); + + let _ = next_command(&mut command_rx, 5, "DispatchTask A").await; + event_tx + .send(SchedulerEvent::TaskCompleted { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + task_id: "A".to_string(), + snapshot_id: None, + duration_ms: 7, + }) + .unwrap(); + + let _ = next_command(&mut command_rx, 5, "CompleteDagRun").await; + event_tx.send(SchedulerEvent::Shutdown).unwrap(); + let _ = driver.await; + + let events = store.range(0, store.current_sequence()).unwrap(); + use conduit_common::event::EventKind; + + assert!( + events.iter().any(|e| matches!( + &e.kind, + EventKind::DagRunCreated { dag_id, run_id, .. } + if dag_id == "dag1" && run_id == "run1" + )), + "expected a persisted DagRunCreated event, got: {:?}", + events.iter().map(|e| &e.kind).collect::>() + ); + assert!( + events.iter().any(|e| matches!( + &e.kind, + EventKind::TaskCompleted { task_id, .. } if task_id == "A" + )), + "expected a persisted TaskCompleted event" + ); + assert!( + events.iter().any(|e| matches!( + &e.kind, + EventKind::DagRunCompleted { + status: conduit_common::event::RunStatus::Success, + .. + } + )), + "expected a persisted DagRunCompleted Success event" + ); +} + +// --------------------------------------------------------------------------- +// Cron scheduling +// --------------------------------------------------------------------------- + +/// Two cron ticks inside the same minute must create exactly one run for a +/// `* * * * *` DAG; a tick in the next minute creates the second run. +#[tokio::test] +async fn cron_tick_fires_due_dag_once_per_minute() { + let task_a = make_task("A", vec![], TriggerRule::AllSuccess, 0, None); + let mut dag = make_dag("cron_dag", vec![task_a], vec!["A"]); + dag.schedule = Some("* * * * *".to_string()); + + let mut plans = HashMap::new(); + plans.insert("cron_dag".to_string(), dag); + + let (tx, mut rx, scheduler_fut) = create_test_scheduler(plans); + let driver = tokio::spawn(scheduler_fut); + + let t0 = Utc.with_ymd_and_hms(2026, 7, 13, 10, 30, 5).unwrap(); + let t0_later = Utc.with_ymd_and_hms(2026, 7, 13, 10, 30, 45).unwrap(); // same minute + let t1 = Utc.with_ymd_and_hms(2026, 7, 13, 10, 31, 5).unwrap(); // next minute + + tx.send(SchedulerEvent::CronTick { timestamp: t0 }).unwrap(); + tx.send(SchedulerEvent::CronTick { timestamp: t0_later }) + .unwrap(); + tx.send(SchedulerEvent::CronTick { timestamp: t1 }).unwrap(); + tx.send(SchedulerEvent::Shutdown).unwrap(); + let _ = driver.await; + + let cmds = drain_commands(&mut rx); + let dispatches = cmds + .iter() + .filter(|c| matches!(c, SchedulerCommand::DispatchTask { dag_id, .. } if dag_id == "cron_dag")) + .count(); + assert_eq!( + dispatches, 2, + "expected exactly 2 dispatches (one per due minute, deduped within a minute), got {} in {:?}", + dispatches, cmds + ); +} + +// --------------------------------------------------------------------------- +// Resource pool enforcement +// --------------------------------------------------------------------------- + +fn create_pool_scheduler( + plans: HashMap, + pools: Vec, +) -> ( + mpsc::UnboundedSender, + mpsc::UnboundedReceiver, + tokio::task::JoinHandle<()>, +) { + let (event_tx, event_rx) = mpsc::unbounded_channel(); + let (command_tx, command_rx) = mpsc::unbounded_channel(); + let scheduler = + Scheduler::new(event_rx, command_tx, PoolManager::new(pools), plans).unwrap(); + let handle = tokio::spawn(async move { + let _ = scheduler.run().await; + }); + (event_tx, command_rx, handle) +} + +/// Two independent tasks sharing a 1-slot pool must execute serially: +/// only one dispatches up front; the second dispatches after the first +/// completes and releases the slot. +#[tokio::test] +async fn one_slot_pool_serializes_independent_tasks() { + let mut task_a = make_task("A", vec![], TriggerRule::AllSuccess, 0, None); + let mut task_b = make_task("B", vec![], TriggerRule::AllSuccess, 0, None); + task_a.pool = Some("solo".to_string()); + task_b.pool = Some("solo".to_string()); + let dag = make_dag("dag1", vec![task_a, task_b], vec!["A", "B"]); + + let mut plans = HashMap::new(); + plans.insert("dag1".to_string(), dag); + + let (tx, mut rx, driver) = create_pool_scheduler( + plans, + vec![Pool { + name: "solo".to_string(), + slots: 1, + description: None, + }], + ); + + tx.send(SchedulerEvent::DagRunRequested { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + logical_date: Utc::now(), + config: HashMap::new(), + }) + .unwrap(); + + // Exactly one task may dispatch while the pool has one slot. + let first = next_command(&mut rx, 5, "first pooled dispatch").await; + let first_task = match &first { + SchedulerCommand::DispatchTask { task_id, .. } => task_id.clone(), + other => panic!("expected DispatchTask, got {:?}", other), + }; + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!( + rx.try_recv().is_err(), + "second task must NOT dispatch while the 1-slot pool is occupied" + ); + + // Completing the first frees the slot; the second must now dispatch. + tx.send(SchedulerEvent::TaskCompleted { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + task_id: first_task.clone(), + snapshot_id: None, + duration_ms: 1, + }) + .unwrap(); + + let second = next_command(&mut rx, 5, "second pooled dispatch after release").await; + match &second { + SchedulerCommand::DispatchTask { task_id, .. } => { + assert_ne!(task_id, &first_task, "the other task must dispatch next") + } + other => panic!("expected DispatchTask, got {:?}", other), + } + + tx.send(SchedulerEvent::Shutdown).unwrap(); + let _ = driver.await; +} + +/// Pool slots are shared across runs: a waiter in run2 must wake up when +/// run1's task releases the slot. +#[tokio::test] +async fn pool_release_wakes_waiters_in_other_runs() { + let mut task_a = make_task("A", vec![], TriggerRule::AllSuccess, 0, None); + task_a.pool = Some("solo".to_string()); + let dag = make_dag("dag1", vec![task_a], vec!["A"]); + + let mut plans = HashMap::new(); + plans.insert("dag1".to_string(), dag); + + let (tx, mut rx, driver) = create_pool_scheduler( + plans, + vec![Pool { + name: "solo".to_string(), + slots: 1, + description: None, + }], + ); + + for run in ["run1", "run2"] { + tx.send(SchedulerEvent::DagRunRequested { + dag_id: "dag1".to_string(), + run_id: run.to_string(), + logical_date: Utc::now(), + config: HashMap::new(), + }) + .unwrap(); + } + + let first = next_command(&mut rx, 5, "first cross-run dispatch").await; + let first_run = match &first { + SchedulerCommand::DispatchTask { run_id, .. } => run_id.clone(), + other => panic!("expected DispatchTask, got {:?}", other), + }; + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!( + rx.try_recv().is_err(), + "run2's task must wait for the pool slot" + ); + + tx.send(SchedulerEvent::TaskCompleted { + dag_id: "dag1".to_string(), + run_id: first_run.clone(), + task_id: "A".to_string(), + snapshot_id: None, + duration_ms: 1, + }) + .unwrap(); + + // run1 completes (CompleteDagRun) and run2's task dispatches. + let mut saw_second_dispatch = false; + for _ in 0..3 { + let cmd = next_command(&mut rx, 5, "commands after release").await; + if let SchedulerCommand::DispatchTask { run_id, .. } = &cmd { + assert_ne!(run_id, &first_run); + saw_second_dispatch = true; + break; + } + } + assert!(saw_second_dispatch, "run2's task must dispatch after release"); + + tx.send(SchedulerEvent::Shutdown).unwrap(); + let _ = driver.await; +} + +/// A task referencing an undefined pool must still run (unlimited, warn) +/// rather than deadlock. +#[tokio::test] +async fn undefined_pool_does_not_block_dispatch() { + let mut task_a = make_task("A", vec![], TriggerRule::AllSuccess, 0, None); + task_a.pool = Some("never_defined".to_string()); + let dag = make_dag("dag1", vec![task_a], vec!["A"]); + + let mut plans = HashMap::new(); + plans.insert("dag1".to_string(), dag); + + let (tx, mut rx, driver) = create_pool_scheduler(plans, vec![]); + + tx.send(SchedulerEvent::DagRunRequested { + dag_id: "dag1".to_string(), + run_id: "run1".to_string(), + logical_date: Utc::now(), + config: HashMap::new(), + }) + .unwrap(); + + let cmd = next_command(&mut rx, 5, "dispatch with undefined pool").await; + assert!( + matches!(&cmd, SchedulerCommand::DispatchTask { task_id, .. } if task_id == "A"), + "task with undefined pool must dispatch, got {:?}", + cmd + ); + + tx.send(SchedulerEvent::Shutdown).unwrap(); + let _ = driver.await; +} diff --git a/conduit-state/src/environment_manager.rs b/conduit-state/src/environment_manager.rs index 3fe0f7d..878edaf 100644 --- a/conduit-state/src/environment_manager.rs +++ b/conduit-state/src/environment_manager.rs @@ -16,13 +16,16 @@ use conduit_common::snapshot::{ use tracing::info; use crate::env_history_store::EnvHistoryStore; +use crate::event_store::EventStore; use crate::snapshot_store::SnapshotStore; +use conduit_common::event::EventKind; /// Manages virtual pipeline environments. pub struct EnvironmentManager { environments: RwLock>, history: Option, snapshots: Option>, + events: Option>, } impl EnvironmentManager { @@ -35,6 +38,7 @@ impl EnvironmentManager { environments: RwLock::new(envs), history: None, snapshots: None, + events: None, } } @@ -55,11 +59,29 @@ impl EnvironmentManager { self } + /// Attach an event store. Environment lifecycle operations (create, + /// promote, rollback) will leave a persistent event trail that + /// `conduit replay` can fold. Best-effort: append failures are logged, + /// never propagated. + pub fn with_event_store(mut self, store: std::sync::Arc) -> Self { + self.events = Some(store); + self + } + /// Reference to the attached history store, if any. pub fn history_store(&self) -> Option<&EnvHistoryStore> { self.history.as_ref() } + /// Best-effort append to the attached event store. + fn persist_event(&self, kind: EventKind) { + if let Some(store) = &self.events { + if let Err(e) = store.append(kind) { + tracing::warn!(error = %e, "Failed to persist environment event"); + } + } + } + /// Update the promotion policy for an environment. pub fn set_promotion_policy( &self, @@ -112,6 +134,7 @@ impl EnvironmentManager { environments: RwLock::new(map), history: None, snapshots: None, + events: None, }) } @@ -161,6 +184,11 @@ impl EnvironmentManager { ); envs.insert(name.to_string(), env.clone()); + drop(envs); + self.persist_event(EventKind::EnvironmentCreated { + env_name: name.to_string(), + based_on: based_on.map(|s| s.to_string()), + }); Ok(env) } @@ -259,6 +287,13 @@ impl EnvironmentManager { "Environment promoted" ); + drop(envs); + self.persist_event(EventKind::EnvironmentPromoted { + source_env: source.to_string(), + target_env: target.to_string(), + snapshot_changes: diff, + }); + Ok(diff) } @@ -409,6 +444,12 @@ impl EnvironmentManager { "Environment rolled back" ); + drop(envs); + self.persist_event(EventKind::EnvironmentRolledBack { + env_name: env_name.to_string(), + rolled_back_to: target_version as u64, + }); + Ok((next, changes)) } @@ -601,6 +642,44 @@ mod tests { assert!(all.iter().any(|e| e.id == "custom-only")); } + /// Environment lifecycle operations must leave a persistent event + /// trail when an event store is attached (powers `conduit replay`). + #[test] + fn env_lifecycle_emits_events_when_store_attached() { + use conduit_common::event::EventKind; + + let hist_dir = tempfile::tempdir().unwrap(); + let events_dir = tempfile::tempdir().unwrap(); + let history = EnvHistoryStore::open(hist_dir.path()).unwrap(); + let events = std::sync::Arc::new(crate::EventStore::open(events_dir.path()).unwrap()); + + let mgr = EnvironmentManager::new() + .with_history_store(history) + .with_event_store(std::sync::Arc::clone(&events)); + + mgr.create("staging", Some("production")).unwrap(); + mgr.promote("staging", "production").unwrap(); + + let recorded = events.range(0, events.current_sequence()).unwrap(); + assert!( + recorded.iter().any(|e| matches!( + &e.kind, + EventKind::EnvironmentCreated { env_name, based_on } + if env_name == "staging" && based_on.as_deref() == Some("production") + )), + "expected EnvironmentCreated event, got {:?}", + recorded.iter().map(|e| &e.kind).collect::>() + ); + assert!( + recorded.iter().any(|e| matches!( + &e.kind, + EventKind::EnvironmentPromoted { source_env, target_env, .. } + if source_env == "staging" && target_env == "production" + )), + "expected EnvironmentPromoted event" + ); + } + #[test] fn cannot_delete_production() { let mgr = EnvironmentManager::new(); diff --git a/conduit-ui/src/api.js b/conduit-ui/src/api.js index 275f84a..51eed1c 100644 --- a/conduit-ui/src/api.js +++ b/conduit-ui/src/api.js @@ -151,6 +151,7 @@ function normalizeRun(run) { startedAt: run.startedAt || run.started_at, endedAt: run.endedAt || run.finished_at || run.ended_at, taskStates: run.taskStates || run.task_states || {}, + taskLogs: run.taskLogs || run.task_logs || {}, tasks: run.tasks || [], triggeredBy: run.triggeredBy || run.triggered_by, environment: run.environment, diff --git a/conduit-ui/src/pages/RunDetail.jsx b/conduit-ui/src/pages/RunDetail.jsx index 0aa9eaa..99bd28c 100644 --- a/conduit-ui/src/pages/RunDetail.jsx +++ b/conduit-ui/src/pages/RunDetail.jsx @@ -217,6 +217,16 @@ export default function RunDetail() {

{status}

+ + {/* Captured task logs */} + {run.taskLogs?.[taskId] && ( +
+

Logs

+ 
+                              {run.taskLogs[taskId]}
+
+
+ )} diff --git a/docs/src/introduction.md b/docs/src/introduction.md index 4e797ea..89d2f97 100644 --- a/docs/src/introduction.md +++ b/docs/src/introduction.md @@ -118,7 +118,7 @@ Tree-sitter extracts the DAG structure, task dependencies, schedules, and retry Each task runs as an **isolated child process** with: - Timeout enforcement (no hung tasks) - Exit-code based success/failure detection -- Resource limits via cgroups (CPU, memory, file descriptors) +- Enforced per-task timeouts (declared CPU/memory limits are not yet enforced) - Structured protocol for logging, metrics, and XCom Tasks communicate via stdout: diff --git a/sdk/python/conduit_sdk/context.py b/sdk/python/conduit_sdk/context.py index 0714c78..eb46a4a 100644 --- a/sdk/python/conduit_sdk/context.py +++ b/sdk/python/conduit_sdk/context.py @@ -2,8 +2,8 @@ Task execution context — runtime information available to tasks. When Conduit's executor runs a task, it provides context via environment -variables and stdin JSON. This module reads that context and exposes it -as a clean Python API. +variables (including upstream XCom as JSON in CONDUIT_XCOM_JSON). This +module reads that context and exposes it as a clean Python API. Environment variables set by the executor: CONDUIT_DAG_ID - The current DAG ID @@ -96,7 +96,7 @@ def my_task(): except ValueError: pass - # Read upstream XCom values from stdin (injected by executor) + # Read upstream XCom values injected by the executor via environment upstream_xcom = _read_upstream_xcom() return TaskContext( @@ -112,10 +112,9 @@ def my_task(): def _read_upstream_xcom() -> dict[str, Any]: """ - Read upstream XCom values from stdin. - - The executor writes a JSON object to stdin before the task starts: - {"extract_orders.return_value": [...], "extract_orders.row_count": 1000} + Read upstream XCom values from the CONDUIT_XCOM_JSON environment + variable, which the executor sets to a JSON object before the task + starts: {"extract_orders.return_value": [...], "extract_orders.row_count": 1000} """ xcom_json = os.environ.get("CONDUIT_XCOM_JSON") if xcom_json: diff --git a/sdk/python/conduit_sdk/decorators.py b/sdk/python/conduit_sdk/decorators.py index c0f235b..2388ca6 100644 --- a/sdk/python/conduit_sdk/decorators.py +++ b/sdk/python/conduit_sdk/decorators.py @@ -48,6 +48,7 @@ class TaskDefinition: function: Callable retries: int = 0 retry_delay: Optional[str] = None + retry_backoff: Optional[float] = None pool: Optional[str] = None timeout: Optional[str] = None priority: int = 0 @@ -160,6 +161,7 @@ def decorator(func: Callable) -> DagDefinition: def task( retries: int = 0, retry_delay: Optional[str] = None, + retry_backoff: Optional[float] = None, pool: Optional[str] = None, timeout: Optional[str] = None, priority: int = 0, @@ -174,6 +176,9 @@ def task( Args: retries: Number of retry attempts on failure. retry_delay: Delay between retries (e.g., "5m", "30s"). + retry_backoff: Exponential backoff multiplier applied to retry_delay + per attempt (e.g., 2.0 doubles the delay each retry; omit for a + fixed delay). pool: Named resource pool for concurrency control. timeout: Maximum execution time (e.g., "1h", "30m"). priority: Execution priority (higher = runs first within pool). @@ -193,6 +198,7 @@ def decorator(func: Callable) -> TaskDefinition: function=func, retries=retries, retry_delay=retry_delay, + retry_backoff=retry_backoff, pool=pool, timeout=timeout, priority=priority, diff --git a/sdk/python/tests/test_decorators.py b/sdk/python/tests/test_decorators.py index 87e991a..b0622e3 100644 --- a/sdk/python/tests/test_decorators.py +++ b/sdk/python/tests/test_decorators.py @@ -46,7 +46,7 @@ def test_task_metadata(): @dag() def test_dag(): - @task(retries=3, retry_delay="5m", pool="my_pool", timeout="30m", priority=10) + @task(retries=3, retry_delay="5m", retry_backoff=2.0, pool="my_pool", timeout="30m", priority=10) def my_task(): """A task with lots of config.""" pass @@ -54,6 +54,7 @@ def my_task(): t = test_dag.tasks["my_task"] assert t.retries == 3 assert t.retry_delay == "5m" + assert t.retry_backoff == 2.0 assert t.pool == "my_pool" assert t.timeout == "30m" assert t.priority == 10