From 514dfcdabd316870afeea275fff82e7a3bfec5b9 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 9 Nov 2026 03:22:50 +0000 Subject: [PATCH 1/2] Batch local activity records with row and byte limits Hold local execution, attempt and history rows in the transaction, then flush bounded INSERTs before other commands and history-dependent reads. Allocate canonical history ranges in one statement while keeping the existing typed single-event fallback and logical persistence charges. Add order, physical-write, large-result, oversized-row, budget refusal and sequence-boundary controls across the existing database suites. Co-Authored-By: Codex --- rust/src/runtime/backend.rs | 3 + rust/src/runtime/cancellation_scopes.rs | 1 + rust/src/runtime/commit_handoff_tests.rs | 71 +++++++ rust/src/runtime/local_activities.rs | 254 +++++++++++++++++++++-- rust/src/runtime/local_write_tests.rs | 113 ++++++++++ rust/src/runtime/mod.rs | 2 + rust/src/runtime/mysql.rs | 1 + rust/src/runtime/postgres.rs | 1 + rust/src/runtime/store.rs | 59 +++++- rust/src/runtime/transaction.rs | 5 + rust/tests/execution/local_activities.rs | 49 +++++ 11 files changed, 536 insertions(+), 23 deletions(-) create mode 100644 rust/src/runtime/local_write_tests.rs 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..2000b1125 --- /dev/null +++ b/rust/src/runtime/local_write_tests.rs @@ -0,0 +1,113 @@ +//! 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"]}), + &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. From f40ac3b5964f3f291d199c3de7533bf86c9997bf Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 9 Nov 2026 03:46:40 +0000 Subject: [PATCH 2/2] Declare local activity capability in the batching fixture The direct Store registration requires the canonical capability manifest that HTTP registration normally supplies. Keep the production guard and all batching assertions unchanged. Co-Authored-By: Codex --- rust/src/runtime/local_write_tests.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/rust/src/runtime/local_write_tests.rs b/rust/src/runtime/local_write_tests.rs index 2000b1125..f0f852c93 100644 --- a/rust/src/runtime/local_write_tests.rs +++ b/rust/src/runtime/local_write_tests.rs @@ -9,7 +9,8 @@ store_methods! { 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"]}), + "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();