diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 55d87bef2..abf3dc9a8 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -88,6 +88,7 @@ pub(super) trait Backend: Database { const HISTORY_SEQUENCE_MAX: i64 = i64::MAX; const HISTORY_ADVANCE_RETURNS_ROW: bool = true; const HISTORY_ADVANCE_SQL: &str = "UPDATE workflow_runs SET last_history_sequence=last_history_sequence+1 WHERE id=$1 AND last_history_sequence>=0 AND last_history_sequence<$2 RETURNING last_history_sequence"; + const HISTORY_BATCH_ADVANCE_SQL: &str = "UPDATE workflow_runs SET last_history_sequence=last_history_sequence+$1 WHERE id=$2 AND last_history_sequence>=0 AND last_history_sequence<=$3 RETURNING last_history_sequence"; const ACTIVITY_METADATA_HISTORY: &str = "SELECT sequence,event_type,CASE WHEN projection_bytes<=8388608 THEN payload ELSE NULL END AS payload,workflow_task_id,recorded_at,projection_count FROM (SELECT sequence,event_type,payload,workflow_task_id,recorded_at,COUNT(*) OVER () AS projection_count,SUM(octet_length(CAST(payload AS TEXT))) OVER (ORDER BY sequence) AS projection_bytes FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') ORDER BY sequence LIMIT 10001) AS page ORDER BY sequence"; // Waits whose stored position differs from their chronological ordinal // over every retained wait of the run, as Workflow #941's projector stores. @@ -253,6 +254,7 @@ fn sqlite_instant(value: String) -> Result> { impl Backend for Sqlite { const HISTORY_ADVANCE_SQL: &str = "UPDATE workflow_runs SET last_history_sequence=last_history_sequence+1 WHERE id=$1 AND typeof(last_history_sequence)='integer' AND last_history_sequence>=0 AND last_history_sequence<$2 RETURNING last_history_sequence"; + const HISTORY_BATCH_ADVANCE_SQL: &str = "UPDATE workflow_runs SET last_history_sequence=last_history_sequence+$1 WHERE id=$2 AND typeof(last_history_sequence)='integer' AND last_history_sequence>=0 AND last_history_sequence<=$3 RETURNING last_history_sequence"; type EncodedInteger = Option; type EncodedFloat = Option; type EncodedFlag = Option; @@ -564,6 +566,7 @@ impl Backend for MySql { const HISTORY_SEQUENCE_MAX: i64 = u32::MAX as i64; const HISTORY_ADVANCE_RETURNS_ROW: bool = false; const HISTORY_ADVANCE_SQL: &str = "UPDATE workflow_runs SET last_history_sequence=LAST_INSERT_ID(last_history_sequence+1) WHERE id=$1 AND last_history_sequence>=0 AND last_history_sequence<$2"; + const HISTORY_BATCH_ADVANCE_SQL: &str = "UPDATE workflow_runs SET last_history_sequence=LAST_INSERT_ID(last_history_sequence+$1) WHERE id=$2 AND last_history_sequence>=0 AND last_history_sequence<=$3"; fn advanced_history_sequence(result: &Self::QueryResult) -> Option { i64::try_from(result.last_insert_id()).ok() } diff --git a/rust/src/runtime/cancellation_scopes.rs b/rust/src/runtime/cancellation_scopes.rs index fb4d7c810..689448f47 100644 --- a/rust/src/runtime/cancellation_scopes.rs +++ b/rust/src/runtime/cancellation_scopes.rs @@ -1160,6 +1160,7 @@ store_methods! { /// after each descendant or fence fetches only the rows written since, /// instead of every row again while it holds the transition lock. pub(super) async fn scope_events(tx: &mut Transaction<'_, DB>, run_id: &str) -> Result>> { + Self::write_local_activities(tx).await?; let mut events = match tx.scope_history.take() { Some(cache) if cache.run_id == run_id => cache.events, _ => Arc::new(Vec::new()), diff --git a/rust/src/runtime/commit_handoff_tests.rs b/rust/src/runtime/commit_handoff_tests.rs index 63292965b..c7790f32e 100644 --- a/rust/src/runtime/commit_handoff_tests.rs +++ b/rust/src/runtime/commit_handoff_tests.rs @@ -14,6 +14,10 @@ fn allocation_outcome(result: super::Result) -> std::result::Result>) -> std::result::Result, String> { + result.map_err(|error| allocation_outcome(Err(error)).unwrap_err()) +} + #[derive(Default)] pub(super) struct CommitGate { pub(super) entered: Notify, @@ -137,6 +141,73 @@ store_methods! { } Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2") .bind(initial_sequence).bind(run_id).execute(&self.pool).await.unwrap(); + self.qualify_history_batch_allocation().await; + } + + async fn qualify_history_batch_allocation(&self) { + use crate::codec::{Value as Payload, ValueCodec}; + use serde_json::json; + let blob = ValueCodec::new().unwrap().encode(&Payload::Array(vec![])).unwrap(); + let run = self.start(json!({"workflow_id":"history-batch-allocation","workflow_type":"probe","task_queue":"batch-allocation", + "input":{"codec":"avro","blob":blob}}),Default::default(),None,None).await.unwrap(); + let run_id = run["run_id"].as_str().unwrap(); + let original_sequence = DB::number(&Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") + .bind(run_id).fetch_one(&self.pool).await.unwrap(),"last_history_sequence").unwrap(); + let maximum = DB::HISTORY_SEQUENCE_MAX; + for value in [0,42,maximum-3,maximum-2,maximum-1,maximum] { + Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2") + .bind(value).bind(run_id).execute(&self.pool).await.unwrap(); + let mut original = self.begin().await.unwrap(); + let expected: super::Result> = async { + let mut values = Vec::new(); + for _ in 0..3 { values.push(Self::allocate_history_sequence_fallback(&mut original,run_id).await?); } + Ok(values) + }.await; + let expected = batch_allocation_outcome(expected); + original.rollback().await.unwrap(); + let mut candidate = self.begin().await.unwrap(); + let (answer,statements) = super::sql_counts::measure(Self::allocate_history_sequences(&mut candidate,run_id,3)).await; + assert_eq!(batch_allocation_outcome(answer),expected,"batch at {value}"); + assert_eq!(statements.count(DB::HISTORY_BATCH_ADVANCE_SQL),1); + if value <= maximum-3 { + assert_eq!(statements.total(),1,"canonical range allocation is one statement"); + } + candidate.rollback().await.unwrap(); + let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") + .bind(run_id).fetch_one(&self.pool).await.unwrap(); + assert_eq!(DB::number(&row,"last_history_sequence").unwrap(),value); + } + let missing = "00000000000000000000000000"; + let mut original = self.begin().await.unwrap(); + let expected = allocation_outcome(Self::allocate_history_sequence_fallback(&mut original,missing).await).unwrap_err(); + original.rollback().await.unwrap(); + let mut candidate = self.begin().await.unwrap(); + assert_eq!(batch_allocation_outcome(Self::allocate_history_sequences(&mut candidate,missing,3).await).unwrap_err(),expected); + let (empty,statements) = super::sql_counts::measure(Self::allocate_history_sequences(&mut candidate,missing,0)).await; + assert!(empty.unwrap().is_empty()); + assert_eq!(statements.total(),0); + candidate.rollback().await.unwrap(); + if DB::NAME != "MySQL" { + Self::query("UPDATE workflow_runs SET last_history_sequence=-1 WHERE id=$1") + .bind(run_id).execute(&self.pool).await.unwrap(); + let mut candidate = self.begin().await.unwrap(); + assert_eq!(Self::allocate_history_sequences(&mut candidate,run_id,3).await.unwrap(),vec![0,1,2]); + candidate.rollback().await.unwrap(); + } + if DB::NAME == "SQLite" { + for sql in ["UPDATE workflow_runs SET last_history_sequence=1.5 WHERE id=$1", + "UPDATE workflow_runs SET last_history_sequence='not-an-integer' WHERE id=$1"] { + Self::query(sql).bind(run_id).execute(&self.pool).await.unwrap(); + let mut original = self.begin().await.unwrap(); + let expected = allocation_outcome(Self::allocate_history_sequence_fallback(&mut original,run_id).await).unwrap_err(); + original.rollback().await.unwrap(); + let mut candidate = self.begin().await.unwrap(); + assert_eq!(batch_allocation_outcome(Self::allocate_history_sequences(&mut candidate,run_id,3).await).unwrap_err(),expected); + candidate.rollback().await.unwrap(); + } + } + Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2") + .bind(original_sequence).bind(run_id).execute(&self.pool).await.unwrap(); } pub(super) async fn qualify_authored_sequence_reads(&self) { diff --git a/rust/src/runtime/local_activities.rs b/rust/src/runtime/local_activities.rs index 1c00ac578..1655c7748 100644 --- a/rust/src/runtime/local_activities.rs +++ b/rust/src/runtime/local_activities.rs @@ -1041,13 +1041,13 @@ store_methods! { let queue = DB::optional_string(run, "queue")?; let started_at = attempts[0].started_at; tx.charge_rows(1)?; - Self::query("INSERT INTO activity_executions(id,workflow_run_id,sequence,activity_class,activity_type,status,payload_codec,arguments,result,exception,connection,queue,attempt_count,current_attempt_id,retry_policy,activity_options,started_at,closed_at,last_heartbeat_at,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21)") - .bind(&execution_id).bind(&run_id).bind(sequence).bind(&record.activity_type).bind(&record.activity_type).bind(status) - .bind(&codec).bind(record.arguments.as_deref()).bind(record.result.as_deref()).bind(exception) - .bind(connection.as_deref()).bind(queue.as_deref()).bind(attempts.len() as i64).bind(&last.id) - .bind(policy.as_ref().map(DB::document)).bind(DB::document(&options)).bind(DB::bind_time(started_at)) - .bind(DB::bind_time(at)).bind(DB::optional_time(latest_heartbeat)).bind(DB::bind_time(at)).bind(DB::bind_time(at)) - .execute(&mut **tx).await?; + tx.local_writes.executions.push(Execution { + id: execution_id.clone(), run_id: run_id.clone(), sequence, activity_type: record.activity_type.clone(), status, + codec: codec.clone(), arguments: record.arguments.clone(), result: record.result.clone(), + exception: exception.map(str::to_owned), connection: connection.clone(), queue: queue.clone(), + attempt_count: attempts.len() as i64, attempt_id: last.id.clone(), retry_policy: policy.clone(), options, + started_at, at, last_heartbeat_at: latest_heartbeat, + }); // `ActivitySnapshot::fromExecution`. let wire = |blob: &Option| blob.as_deref().map_or(Value::Null, |blob| json!({"codec": codec, "blob": blob})); let terminal_activity = filtered(&[ @@ -1081,16 +1081,17 @@ store_methods! { } payload }; - Self::append(tx, &run_id, "ActivityScheduled", event(scheduled_activity, json!({})), Some(&task_id), None).await?; + Self::hold_event(tx, &run_id, &task_id, "ActivityScheduled", event(scheduled_activity, json!({})))?; let mut emitted: Option> = None; for (position, attempt) in attempts.iter().enumerate() { let worker_attempt_id = attempt.report.get("attempt_id").and_then(Value::as_str); tx.charge_rows(1)?; - Self::query("INSERT INTO activity_attempts(id,workflow_run_id,activity_execution_id,workflow_task_id,worker_attempt_id,attempt_number,status,lease_owner,started_at,last_heartbeat_at,lease_expires_at,closed_at,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,NULL,$11,$12,$13)") - .bind(&attempt.id).bind(&run_id).bind(&execution_id).bind(&task_id).bind(worker_attempt_id) - .bind(attempt.number()).bind(attempt.status()).bind(lease_owner.as_deref()).bind(DB::bind_time(attempt.started_at)) - .bind(DB::optional_time(attempt.last_heartbeat_at)).bind(DB::bind_time(attempt.closed_at)) - .bind(DB::bind_time(at)).bind(DB::bind_time(at)).execute(&mut **tx).await?; + tx.local_writes.attempts.push(AttemptRow { + id: attempt.id.clone(), run_id: run_id.clone(), execution_id: execution_id.clone(), task_id: task_id.clone(), + worker_attempt_id: worker_attempt_id.map(str::to_owned), number: attempt.number(), status: attempt.status(), + lease_owner: lease_owner.clone(), started_at: attempt.started_at, last_heartbeat_at: attempt.last_heartbeat_at, + closed_at: attempt.closed_at, at, + }); let terminal_attempt = filtered(&[ ("id", json!(attempt.id)), ("worker_attempt_id", attempt.worker_id()), ("attempt_number", json!(attempt.number())), ("status", json!(attempt.status())), ("task_id", json!(task_id)), ("lease_owner", json!(lease_owner)), @@ -1108,7 +1109,7 @@ store_methods! { } let started = json!({"activity_attempt_id": attempt.id, "worker_attempt_id": attempt.worker_id(), "attempt_number": attempt.number(), "activity_attempt": running_attempt}); - Self::append(tx, &run_id, "ActivityStarted", event(running_activity.clone(), started.clone()), Some(&task_id), None).await?; + Self::hold_event(tx, &run_id, &task_id, "ActivityStarted", event(running_activity.clone(), started.clone()))?; for heartbeat in attempt.heartbeats() { let heartbeat_at = checked_add(attempt.started_at, Duration::try_milliseconds(milliseconds(heartbeat.get("elapsed_ms"))))?; if emitted.is_none_or(|latest| heartbeat_at > latest) { @@ -1123,7 +1124,7 @@ store_methods! { if let Some(details) = heartbeat.get("details") { fields["progress"] = details.clone(); } - Self::append(tx, &run_id, "ActivityHeartbeatRecorded", event(activity, fields), Some(&task_id), None).await?; + Self::hold_event(tx, &run_id, &task_id, "ActivityHeartbeatRecorded", event(activity, fields))?; } if position + 1 < attempts.len() { let mut activity = running_activity.clone(); @@ -1141,7 +1142,7 @@ store_methods! { "exception_type": report.get("exception_type").cloned().unwrap_or(Value::Null), "retry_reason": report.get("retry_reason").cloned().unwrap_or_else(|| json!("failure")), "retry_backoff_seconds": report.get("backoff_seconds").cloned().unwrap_or_else(|| json!(0))}); - Self::append(tx, &run_id, "ActivityRetryScheduled", event(activity, retry), Some(&task_id), None).await?; + Self::hold_event(tx, &run_id, &task_id, "ActivityRetryScheduled", event(activity, retry))?; } } let mut terminal = json!({"activity_attempt_id": last.id, "worker_attempt_id": last.worker_id(), "attempt_number": last.number(), @@ -1189,11 +1190,230 @@ store_methods! { } } } - Self::append(tx, &run_id, event_type, event(terminal_activity, terminal), Some(&task_id), None).await?; + Self::hold_event(tx, &run_id, &task_id, event_type, event(terminal_activity, terminal)) + } + + /// Charges an event as `Store::record` does and holds it for + /// `write_local_activities`. Each event carries its task's snapshot and + /// no command, and closes nothing. + fn hold_event(tx: &mut Transaction<'_, DB>, run_id: &str, task_id: &str, event_type: &'static str, payload: Value) -> Result<()> { + let recorded_at = now(); + tx.charge_event(event_type, &payload)?; + tx.local_writes.events.push(Event { id: id(), run_id: run_id.to_owned(), task_id: task_id.to_owned(), event_type, payload, recorded_at }); + Ok(()) + } + + /// Writes the rows `record_local_activity` holds, in the order it made + /// them: each table's INSERTs have both row and encoded-size bounds, and a + /// run's events at sequences from one allocation. A run whose sequence + /// cannot take them all allocates one at a time, as `Store::record` does, + /// and so refuses as it would. Then each event is projected as + /// `Store::record` projects it. + pub(super) async fn write_local_activities(tx: &mut Transaction<'_, DB>) -> Result<()> { + if let Some(refusal) = tx.meter.as_ref().and_then(super::history_meter::Meter::tripped) { + return Err(refusal); + } + let writes = std::mem::take(&mut tx.local_writes); + for rows in bounded_rows(&writes.executions, EXECUTION_INSERT, Execution::bound_bytes) { + let sql = insert_rows::(EXECUTION_INSERT, + "(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", rows.len()); + #[cfg(test)] + super::sql_counts::record(EXECUTION_INSERT); + let mut query = sqlx::query::(sqlx::AssertSqlSafe(sql)); + for row in rows { + query = query.bind(row.id.as_str()).bind(row.run_id.as_str()).bind(row.sequence).bind(row.activity_type.as_str()) + .bind(row.activity_type.as_str()).bind(row.status).bind(row.codec.as_str()).bind(row.arguments.as_deref()) + .bind(row.result.as_deref()).bind(row.exception.as_deref()).bind(row.connection.as_deref()).bind(row.queue.as_deref()) + .bind(row.attempt_count).bind(row.attempt_id.as_str()).bind(row.retry_policy.as_ref().map(DB::document)) + .bind(DB::document(&row.options)).bind(DB::bind_time(row.started_at)).bind(DB::bind_time(row.at)) + .bind(DB::optional_time(row.last_heartbeat_at)).bind(DB::bind_time(row.at)).bind(DB::bind_time(row.at)); + } + query.execute(&mut **tx).await?; + } + for rows in bounded_rows(&writes.attempts, ATTEMPT_INSERT, AttemptRow::bound_bytes) { + let sql = insert_rows::(ATTEMPT_INSERT, + "(?,?,?,?,?,?,?,?,?,?,NULL,?,?,?)", rows.len()); + #[cfg(test)] + super::sql_counts::record(ATTEMPT_INSERT); + let mut query = sqlx::query::(sqlx::AssertSqlSafe(sql)); + for row in rows { + query = query.bind(row.id.as_str()).bind(row.run_id.as_str()).bind(row.execution_id.as_str()).bind(row.task_id.as_str()) + .bind(row.worker_attempt_id.as_deref()).bind(row.number).bind(row.status).bind(row.lease_owner.as_deref()) + .bind(DB::bind_time(row.started_at)).bind(DB::optional_time(row.last_heartbeat_at)).bind(DB::bind_time(row.closed_at)) + .bind(DB::bind_time(row.at)).bind(DB::bind_time(row.at)); + } + query.execute(&mut **tx).await?; + } + for events in writes.events.chunk_by(|left, right| left.run_id == right.run_id) { + let run_id = events[0].run_id.as_str(); + let sequences = Self::allocate_history_sequences(tx, run_id, events.len()).await?; + let mut offset = 0; + for rows in bounded_rows(events, HISTORY_INSERT, Event::bound_bytes) { + let sequences = &sequences[offset..offset + rows.len()]; + offset += rows.len(); + let sql = insert_rows::(HISTORY_INSERT, + "(?,?,?,?,?,?,NULL,?)", rows.len()); + #[cfg(test)] + super::sql_counts::record(HISTORY_INSERT); + let mut query = sqlx::query::(sqlx::AssertSqlSafe(sql)); + for (row, sequence) in rows.iter().zip(sequences) { + query = query.bind(row.id.as_str()).bind(run_id).bind(*sequence).bind(row.event_type) + .bind(DB::document(&row.payload)).bind(row.task_id.as_str()).bind(DB::bind_time(row.recorded_at)); + } + query.execute(&mut **tx).await?; + } + for event in events { + Self::project_event(tx, run_id, event.event_type, &event.payload, event.recorded_at).await?; + } + tx.summary_changed(run_id); + } Ok(()) } } +/// The most rows one INSERT of held local activity rows writes. +const ROWS_PER_INSERT: usize = 100; +const BYTES_PER_INSERT: usize = 1024 * 1024; +const EXECUTION_INSERT: &str = "INSERT INTO activity_executions(id,workflow_run_id,sequence,activity_class,activity_type,status,payload_codec,arguments,result,exception,connection,queue,attempt_count,current_attempt_id,retry_policy,activity_options,started_at,closed_at,last_heartbeat_at,created_at,updated_at)"; +const ATTEMPT_INSERT: &str = "INSERT INTO activity_attempts(id,workflow_run_id,activity_execution_id,workflow_task_id,worker_attempt_id,attempt_number,status,lease_owner,started_at,last_heartbeat_at,lease_expires_at,closed_at,created_at,updated_at)"; +const HISTORY_INSERT: &str = "INSERT INTO workflow_history_events(id,workflow_run_id,sequence,event_type,payload,workflow_task_id,workflow_command_id,recorded_at)"; + +// Values remain bound parameters. Each bound includes the UTF-8 strings and +// serialized JSON plus 128 bytes per parameter for its placeholder, framing, +// type/length metadata and fixed-width numeric/timestamp representation. +// The shared row cap bounds placeholder widths as well as parameter counts. +fn string_bytes(values: &[Option<&str>]) -> usize { + values.iter().flatten().fold(0usize, |total, value| total.saturating_add(value.len())) +} + +fn bounded_rows<'a, T>(rows: &'a [T], head: &str, size: impl Fn(&T) -> usize) -> Vec<&'a [T]> { + let mut chunks = Vec::new(); + let mut start = 0; + let prefix = head.len().saturating_add(" VALUES ".len()); + let mut bytes = prefix; + for (index, row) in rows.iter().enumerate() { + let row_bytes = size(row); + if index > start && (index - start == ROWS_PER_INSERT + || bytes.saturating_add(row_bytes) > BYTES_PER_INSERT) + { + chunks.push(&rows[start..index]); + start = index; + bytes = prefix; + } + bytes = bytes.saturating_add(row_bytes); + } + if start < rows.len() { chunks.push(&rows[start..]); } + // A legal row larger than the batching target is written alone using + // the existing bindings. It never enlarges a multi-row statement or + // introduces a new product refusal for a previously legal single row. + chunks +} + +/// The rows `record_local_activity` makes, held in the transaction +/// (`Transaction::local_writes`) while a completion applies a run of local +/// activity records, so that each table takes one INSERT for the run and +/// its history one sequence allocation (`Store::write_local_activities`). +#[derive(Default)] +pub(super) struct Writes { + executions: Vec, + attempts: Vec, + events: Vec, +} + +struct Execution { + id: String, + run_id: String, + sequence: i64, + activity_type: String, + status: &'static str, + codec: String, + arguments: Option, + result: Option, + exception: Option, + connection: Option, + queue: Option, + attempt_count: i64, + attempt_id: String, + retry_policy: Option, + options: Value, + started_at: DateTime, + at: DateTime, + last_heartbeat_at: Option>, +} + +impl Execution { + fn bound_bytes(&self) -> usize { + (21usize * 128).saturating_add(string_bytes(&[ + Some(&self.id), Some(&self.run_id), Some(&self.activity_type), Some(&self.activity_type), + Some(self.status), Some(&self.codec), self.arguments.as_deref(), self.result.as_deref(), + self.exception.as_deref(), self.connection.as_deref(), self.queue.as_deref(), Some(&self.attempt_id), + ])).saturating_add(self.retry_policy.as_ref().map_or(0, |value| value.to_string().len())) + .saturating_add(self.options.to_string().len()) + } +} + +struct AttemptRow { + id: String, + run_id: String, + execution_id: String, + task_id: String, + worker_attempt_id: Option, + number: i64, + status: &'static str, + lease_owner: Option, + started_at: DateTime, + last_heartbeat_at: Option>, + closed_at: DateTime, + at: DateTime, +} + +impl AttemptRow { + fn bound_bytes(&self) -> usize { + (13usize * 128).saturating_add(string_bytes(&[ + Some(&self.id), Some(&self.run_id), Some(&self.execution_id), Some(&self.task_id), + self.worker_attempt_id.as_deref(), Some(self.status), self.lease_owner.as_deref(), + ])) + } +} + +struct Event { + id: String, + run_id: String, + task_id: String, + event_type: &'static str, + payload: Value, + recorded_at: DateTime, +} + +impl Event { + fn bound_bytes(&self) -> usize { + (7usize * 128).saturating_add(string_bytes(&[ + Some(&self.id), Some(&self.run_id), Some(&self.task_id), Some(self.event_type), + ])).saturating_add(self.payload.to_string().len()) + } +} + +/// `head` with `rows` copies of the placeholder row `row`, numbered as +/// the shared templates are except on MySQL, which binds '?' in order. +fn insert_rows(head: &str, row: &str, rows: usize) -> String { + let mut sql = format!("{head} VALUES "); + let mut position = 0; + for index in 0..rows { + if index > 0 { + sql.push(','); + } + for character in row.chars() { + if character == '?' && DB::NAME != "MySQL" { + position += 1; + sql.push_str(&format!("${position}")); + } else { + sql.push(character); + } + } + } + sql +} + #[cfg(test)] mod tests { use super::*; diff --git a/rust/src/runtime/local_write_tests.rs b/rust/src/runtime/local_write_tests.rs new file mode 100644 index 000000000..f0f852c93 --- /dev/null +++ b/rust/src/runtime/local_write_tests.rs @@ -0,0 +1,114 @@ +//! Physical batch counts, byte splitting and completion-budget refusals. +use super::{RuntimeError, store::Store}; +use serde_json::json; +use std::{collections::BTreeMap, sync::Arc}; + +store_methods! { + pub(super) async fn qualify_local_write_batches(&self) { + use crate::codec::{Value as Payload, ValueCodec}; + let codec = ValueCodec::new().unwrap(); + let empty = codec.encode(&Payload::Array(vec![])).unwrap(); + self.register(json!({"worker_id":"local-batch-worker", "task_queue":"local-batches", "runtime":"php", + "supported_workflow_types":["local-batches"], "supported_activity_types":[], "capabilities":["local_activities"], + "capability_manifest":{"local_activities":{"supported":true}}}), + &super::QueryTaskConfig::default(), Some("1.20")).await.unwrap(); + for (name, records, result_bytes) in [("small",40,0), ("split",4,250_000), ("single",1,900_000)] { + let result = codec.encode(&Payload::String("x".repeat(result_bytes))).unwrap(); + let started = self.start(json!({"workflow_id":format!("local-batch-{name}"), "workflow_type":"local-batches", + "task_queue":"local-batches", "input":{"codec":"avro","blob":empty}}), + BTreeMap::new(), None, None).await.unwrap(); + let run_id = started["run_id"].as_str().unwrap(); + let task = self.poll(json!({"worker_id":"local-batch-worker","task_queue":"local-batches","timeout_seconds":0}), + "workflow", "1.20").await.unwrap()["task"].clone(); + assert_eq!(task["run_id"],run_id); + let mut commands: Vec<_> = (0..records).map(|index| json!({"type":"record_local_activity", + "activity_type":format!("local.batch-{index}"), "arguments":{"codec":"avro","blob":empty}, + "result":{"codec":"avro","blob":result}, "outcome":"completed", + "attempts":[{"attempt_number":1,"outcome":"completed","duration_ms":1}]})).collect(); + commands.push(json!({"type":"complete_workflow","result":{"codec":"avro","blob":empty}})); + let (answer, statements) = super::sql_counts::measure(self.complete_or_checkpoint( + task["task_id"].as_str().unwrap(),json!({"lease_owner":"local-batch-worker", + "workflow_task_attempt":task["workflow_task_attempt"],"commands":commands}), + Arc::new(tokio::sync::Semaphore::new(1)),BTreeMap::new(),None, + super::carried_polls::Config {protocol:Some("1.20"),pages:Default::default(),default_wait:30}, + Default::default(),None,None)).await; + assert_eq!(answer.unwrap()["recorded"],true); + let executions = statements.count_prefix("INSERT INTO activity_executions("); + let attempts = statements.count_prefix("INSERT INTO activity_attempts("); + let history = statements.count_prefix("INSERT INTO workflow_history_events("); + assert_eq!(attempts,1,"{name}: small attempt rows stay together"); + assert_eq!(statements.count(DB::HISTORY_BATCH_ADVANCE_SQL),1,"{name}: one range allocation"); + if name == "small" { + assert_eq!(executions,1,"40 executions share one INSERT"); + assert_eq!(history,3,"120 local events split at 100, followed by the terminal event"); + } else if name == "split" { + assert!(executions > 1 && executions < records,"large legal results split before the row cap"); + assert!(history > 3,"large event payloads also split before the row cap"); + } else { + assert_eq!(executions,1,"a legal oversized execution is written alone"); + assert_eq!(history,3,"the oversized terminal activity event is written alone"); + } + let stored = Self::query("SELECT sequence,status,result FROM activity_executions WHERE workflow_run_id=$1 ORDER BY sequence") + .bind(run_id).fetch_all(&self.pool).await.unwrap(); + assert_eq!(stored.len(),records); + for (index,row) in stored.iter().enumerate() { + assert_eq!(DB::number(row,"sequence").unwrap(),index as i64 + 1); + assert_eq!(DB::string(row,"status").unwrap(),"completed"); + assert_eq!(DB::optional_string(row,"result").unwrap().as_deref(),Some(result.as_str())); + } + let history = Self::query("SELECT sequence,event_type FROM workflow_history_events WHERE workflow_run_id=$1 ORDER BY sequence") + .bind(run_id).fetch_all(&self.pool).await.unwrap(); + assert_eq!(history.len(),2 + 3 * records + 1); + assert_eq!(history.iter().map(|row| DB::number(row,"sequence").unwrap()).collect::>(), + (1..=history.len() as i64).collect::>()); + assert_eq!(DB::string(history.last().unwrap(),"event_type").unwrap(),"WorkflowCompleted"); + } + for (name, limits, reason) in [ + ("events",super::memo_limits::StructuralLimits {history_events:Some(3),..Default::default()},"completion_budget_exceeded"), + ("rows",super::memo_limits::StructuralLimits {history_rows:Some(7),..Default::default()},"completion_budget_exceeded"), + ("bytes",super::memo_limits::StructuralLimits {history_bytes:Some(1),..Default::default()},"history_transaction_limit_exceeded"), + ] { + let started = self.start(json!({"workflow_id":format!("local-batch-budget-{name}"),"workflow_type":"local-batches", + "task_queue":"local-batches","input":{"codec":"avro","blob":empty}}), + BTreeMap::new(),None,None).await.unwrap(); + let run_id = started["run_id"].as_str().unwrap(); + let task = self.poll(json!({"worker_id":"local-batch-worker","task_queue":"local-batches","timeout_seconds":0}), + "workflow","1.20").await.unwrap()["task"].clone(); + let commands = json!([{"type":"record_local_activity","activity_type":"local.budget", + "arguments":{"codec":"avro","blob":empty},"result":{"codec":"avro","blob":empty},"outcome":"completed", + "attempts":[{"attempt_number":1,"outcome":"completed","duration_ms":1}]}, + {"type":"complete_workflow","result":{"codec":"avro","blob":empty}}]); + let body = json!({"lease_owner":"local-batch-worker","workflow_task_attempt":task["workflow_task_attempt"],"commands":commands}); + let (answer,statements) = super::sql_counts::measure(self.complete_or_checkpoint( + task["task_id"].as_str().unwrap(),body.clone(),Arc::new(tokio::sync::Semaphore::new(1)),BTreeMap::new(),None, + super::carried_polls::Config {protocol:Some("1.20"),pages:Default::default(),default_wait:30},limits,None,None)).await; + let Err(RuntimeError::Protocol {status,response}) = answer else { panic!("{name}: a refused budget"); }; + assert_eq!(status,axum::http::StatusCode::CONFLICT); + assert_eq!(response["reason"],reason); + assert_eq!(response["recorded"],false); + for prefix in ["INSERT INTO activity_executions(","INSERT INTO activity_attempts(","INSERT INTO workflow_history_events("] { + assert_eq!(statements.count_prefix(prefix),0,"{name}: refusal before the first local write"); + } + assert_eq!(statements.count(DB::HISTORY_BATCH_ADVANCE_SQL),0); + let task_row = Self::query("SELECT status FROM workflow_tasks WHERE id=$1") + .bind(task["task_id"].as_str().unwrap()).fetch_one(&self.pool).await.unwrap(); + assert_eq!(DB::string(&task_row,"status").unwrap(),"leased"); + let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") + .bind(run_id).fetch_one(&self.pool).await.unwrap(); + assert_eq!(DB::number(&row,"last_history_sequence").unwrap(),2); + let accepted = self.complete_or_checkpoint(task["task_id"].as_str().unwrap(),body, + Arc::new(tokio::sync::Semaphore::new(1)),BTreeMap::new(),None, + super::carried_polls::Config {protocol:Some("1.20"),pages:Default::default(),default_wait:30}, + Default::default(),None,None).await.unwrap(); + assert_eq!(accepted["recorded"],true,"{name}: the refused claim remains usable"); + } + } +} + +#[tokio::test] +async fn sqlite_local_writes_are_bounded_and_keep_completion_budgets() { + let directory = tempfile::tempdir().unwrap(); + let store = Store::::new(super::sqlite::open(directory.path().join("local-batches.sqlite").to_str().unwrap()).await.unwrap()); + store.qualify_local_write_batches().await; + store.close().await; +} diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 9b41f5089..3ce56bf9f 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -163,6 +163,8 @@ mod commit_handoff_tests; mod sql_counts; #[cfg(test)] mod poll_write_tests; +#[cfg(test)] +mod local_write_tests; mod schema; pub mod schema_upgrade; mod search_attribute_upserts; diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index 7d0b5fa02..935019a7a 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -1255,6 +1255,7 @@ mod tests { let store = super::super::store::Store::new(MySqlStorage::open(db.options.clone()).await.unwrap().into_pool()); store.qualify_commit_handoff().await; store.qualify_history_sequence_allocation().await; + store.qualify_local_write_batches().await; store.qualify_authored_sequence_reads().await; store.close().await; db.remove().await; diff --git a/rust/src/runtime/postgres.rs b/rust/src/runtime/postgres.rs index 4165471f7..4156fc099 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -785,6 +785,7 @@ mod tests { let store = super::super::store::Store::new(PostgresStorage::open(db.options.clone()).await.unwrap().into_pool()); store.qualify_commit_handoff().await; store.qualify_history_sequence_allocation().await; + store.qualify_local_write_batches().await; store.qualify_authored_sequence_reads().await; store.close().await; db.remove().await; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 0bffe8a42..52adb626c 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -3474,6 +3474,9 @@ where .chain((0..commands.len()).filter(|index| is_terminal(&commands[*index]))); for index in order { let command = &commands[index]; + if command["type"] != super::local_activities::COMMAND { + Self::write_local_activities(&mut tx).await?; + } if command["type"] == "fail_workflow" { // Applied after the loop. continue; @@ -3541,6 +3544,7 @@ where } } } + Self::write_local_activities(&mut tx).await?; let failed = if let Some(command) = commands .iter() .find(|command| command["type"] == "fail_workflow") @@ -3634,6 +3638,9 @@ where created: &mut Vec, ) -> Result<()> { let Apply { run, task, task_id, run_id, attempt, limits, external, local, exceptions, start_turn } = *apply; + if command["type"] != super::local_activities::COMMAND { + Self::write_local_activities(tx).await?; + } let mut sequence = *sequence_ref; if matches!( command["type"].as_str(), @@ -4773,6 +4780,39 @@ where Self::record(tx, run_id, event_type, payload, task_id, Some(command_id), None, false).await.map(|_| ()) } + pub(super) async fn allocate_history_sequences( + tx: &mut Transaction<'_, DB>, run_id: &str, count: usize, + ) -> Result> { + if count == 0 { return Ok(Vec::new()); } + let count_i64 = i64::try_from(count) + .map_err(|_| refuse(StatusCode::CONFLICT, "history_sequence_exhausted"))?; + if let Some(maximum) = DB::HISTORY_SEQUENCE_MAX.checked_sub(count_i64).filter(|value| *value >= 0) { + let last = if DB::HISTORY_ADVANCE_RETURNS_ROW { + Self::query(DB::HISTORY_BATCH_ADVANCE_SQL).bind(count_i64).bind(run_id).bind(maximum) + .fetch_optional(&mut **tx).await? + .map(|row| DB::number(&row, "last_history_sequence")).transpose()? + } else { + let result = Self::query(DB::HISTORY_BATCH_ADVANCE_SQL) + .bind(count_i64).bind(run_id).bind(maximum).execute(&mut **tx).await?; + let last = DB::advanced_history_sequence(&result); + if DB::affected(result) == 1 { + Some(last.filter(|value| *value >= count_i64) + .ok_or_else(|| refuse(StatusCode::INTERNAL_SERVER_ERROR, "invalid_history_sequence"))?) + } else { None } + }; + if let Some(last) = last { + return Ok((last - count_i64 + 1..=last).collect()); + } + } + // Retain the single-event path's typed reads and refusals for a + // missing, noncanonical or exhausted counter. No partial batch commits. + let mut sequences = Vec::with_capacity(count); + for _ in 0..count { + sequences.push(Self::allocate_history_sequence(tx, run_id).await?); + } + Ok(sequences) + } + pub(super) async fn allocate_history_sequence( tx: &mut Transaction<'_, DB>, run_id: &str, ) -> Result { @@ -4854,12 +4894,7 @@ where let event_id = id(); Self::query("INSERT INTO workflow_history_events(id,workflow_run_id,sequence,event_type,payload,workflow_task_id,workflow_command_id,recorded_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)") .bind(&event_id).bind(run_id).bind(sequence).bind(event_type).bind(DB::document(&payload)).bind(task_id).bind(command_id).bind(DB::bind_time(recorded_at)).execute(&mut **tx).await?; - Self::project_activity_wait_event(tx, run_id, event_type, &payload, recorded_at).await?; - Self::project_timer_wait_event(tx, run_id, event_type, &payload, recorded_at).await?; - Self::project_signal_wait_event(tx, run_id, event_type, &payload, recorded_at).await?; - Self::project_condition_wait_event(tx, run_id, event_type, &payload, recorded_at).await?; - Self::project_child_wait_event(tx, run_id, event_type, &payload, recorded_at).await?; - Self::project_update_wait_event(tx, run_id, event_type, &payload, recorded_at).await?; + Self::project_event(tx, run_id, event_type, &payload, recorded_at).await?; if let Some(reason) = closed_reason(event_type) { Self::settle_closed_wait_timeouts(tx, run_id, recorded_at).await?; let task = payload.get("task").cloned(); @@ -4869,6 +4904,18 @@ where Ok(event_id) } + pub(super) async fn project_event( + tx: &mut Transaction<'_, DB>, run_id: &str, event_type: &str, + payload: &Value, recorded_at: DateTime, + ) -> Result<()> { + Self::project_activity_wait_event(tx, run_id, event_type, payload, recorded_at).await?; + Self::project_timer_wait_event(tx, run_id, event_type, payload, recorded_at).await?; + Self::project_signal_wait_event(tx, run_id, event_type, payload, recorded_at).await?; + Self::project_condition_wait_event(tx, run_id, event_type, payload, recorded_at).await?; + Self::project_child_wait_event(tx, run_id, event_type, payload, recorded_at).await?; + Self::project_update_wait_event(tx, run_id, event_type, payload, recorded_at).await + } + /// `PendingSignalCloser::closeForTerminalEvent`: once a run closes, each /// signal it received but never applied is rejected, in arrival order. /// The original command keeps its acceptance; `SignalRejected` records diff --git a/rust/src/runtime/transaction.rs b/rust/src/runtime/transaction.rs index 6249b4071..aea8c121f 100644 --- a/rust/src/runtime/transaction.rs +++ b/rust/src/runtime/transaction.rs @@ -40,6 +40,7 @@ pub(super) struct Transaction<'a, DB: Backend> { pub(super) meter: Option, /// The run history the cancellation scope engine has read (`scope_events`). pub(super) scope_history: Option, + pub(super) local_writes: super::local_activities::Writes, // An implicit failure drops the database transaction before this guard. _queue_guard: Option>, #[cfg(test)] @@ -60,6 +61,7 @@ impl<'a, DB: Backend> Transaction<'a, DB> { history_budget, meter: None, scope_history: None, + local_writes: Default::default(), _queue_guard: queue_guard, #[cfg(test)] commit_gate: None, @@ -139,6 +141,7 @@ where /// a partial claim. Budget charges stay consumed and discarded reads /// cannot survive in the transaction's cancellation-history cache. pub(super) async fn poll_savepoint(&mut self) -> Result { + Store::::write_local_activities(self).await?; // MySQL does not support savepoint commands in its prepared-statement // protocol. Send these fixed transaction-control literals directly. (&mut **self).execute("SAVEPOINT dw_poll_claim").await?; @@ -150,6 +153,7 @@ where self.summaries = saved.summaries; self.workflow_poll_replayed = saved.workflow_poll_replayed; self.scope_history = None; + self.local_writes = Default::default(); self.run_tags.clear(); self.release_poll_claim().await } @@ -165,6 +169,7 @@ where if let Some(refusal) = self.meter.as_ref().and_then(super::history_meter::Meter::tripped) { return Err(refusal); } + Store::::write_local_activities(&mut self).await?; for run_id in std::mem::take(&mut self.summaries) { Store::::order_waits(&mut self, &run_id).await?; Store::::project_run_summary(&mut self, &run_id).await?; diff --git a/rust/tests/execution/local_activities.rs b/rust/tests/execution/local_activities.rs index 36aecc763..64399010e 100644 --- a/rust/tests/execution/local_activities.rs +++ b/rust/tests/execution/local_activities.rs @@ -175,6 +175,55 @@ async fn a_recorded_local_activity_writes_its_attempts_history_and_failure() { assert_eq!((&described["status"], &described["wait_kind"]), (&json!("waiting"), &json!("signal")), "{described}"); } +/// Records written together, more of them than one statement takes, keep the +/// order and consecutive sequences of records written one at a time, and a +/// command between them is recorded between their histories. +#[tokio::test] +async fn a_long_run_of_local_activity_records_keeps_its_order() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + local_worker(&app, "local", json!(["local_activities"])).await; + let run = start(&app, "local-long").await; + let task = poll(&app, "local", "workflow").await; + let mut commands: Vec = (0..40) + .map(|index| record(json!({"activity_type": format!("local.step-{index}"), + "attempts": [{"attempt_id": format!("w-{index}"), "attempt_number": 1, "outcome": "completed", "duration_ms": 5}]}))) + .collect(); + commands.insert(35, json!({"type": "record_side_effect", "result": DONE, "payload_codec": "avro"})); + commands.push(wait()); + let (status, body) = complete(&app, &task, Value::Array(commands), "1.20").await; + assert_eq!(status, StatusCode::OK, "{body}"); + + let (status, history) = request(&app, "GET", &format!("/api/workflows/local-long/runs/{}/history?page_size=1000", + run["run_id"].as_str().unwrap()), Value::Null).await; + assert_eq!(status, StatusCode::OK, "{history}"); + let events = history["events"].as_array().unwrap(); + assert_eq!(events.len(), 2 + 40 * 3 + 2, "{history}"); + let sequences: Vec = events.iter().map(|event| event["sequence"].as_i64().unwrap()).collect(); + assert_eq!(sequences, (1..=events.len() as i64).collect::>()); + let local = |index: usize| { + let at = 2 + 3 * index + usize::from(index >= 35); + let types: Vec<_> = events[at..at + 3].iter().map(|event| event["event_type"].as_str().unwrap()).collect(); + assert_eq!(types, ["ActivityScheduled", "ActivityStarted", "ActivityCompleted"], "record {index}"); + assert_eq!(events[at]["payload"]["activity_type"], format!("local.step-{index}")); + assert_eq!(events[at + 1]["payload"]["worker_attempt_id"], format!("w-{index}")); + assert_eq!(events[at]["payload"]["sequence"], index as i64 + 1 + i64::from(index >= 35)); + }; + (0..40).for_each(local); + assert_eq!((&events[2 + 3 * 35]["event_type"], &events[2 + 3 * 35]["payload"]["sequence"]), (&json!("SideEffectRecorded"), &json!(36))); + assert_eq!(events.last().unwrap()["event_type"], "SignalWaitOpened"); + + let (status, view) = request(&app, "GET", &format!("/api/workflows/local-long/runs/{}/activities", + run["run_id"].as_str().unwrap()), Value::Null).await; + assert_eq!(status, StatusCode::OK, "{view}"); + let activities = view["activities"].as_array().unwrap(); + assert_eq!(activities.len(), 40); + for (index, activity) in activities.iter().enumerate() { + assert_eq!((&activity["status"], &activity["attempt_count"]), (&json!("completed"), &json!(1)), "{activity}"); + assert_eq!(activity["attempts"][0]["worker_attempt_id"], format!("w-{index}"), "{activity}"); + } +} + /// The prepared local-activity routes answer as a PHP server whose bridge lacks /// the prepared role: refused with 409 before any task is read, whether or not /// the task exists, in the worker envelope.