Native: write a run of local activity records with one statement per table #1068

Activity

1. rmcdaniel

   added

   kind:featureA public product capability or experience is requestedA public product capability or experience is requested

   priority:P0Immediate release or data-integrity riskImmediate release or data-integrity risk

   repo:serverOwned by the standalone server repositoryOwned by the standalone server repository

   agent:server

   on Nov 6, 2026
2. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0027-t17-Write-a-run-of-local-activity-records-with-one-state.patch, part 1/1.

   This is an unverified, cumulative draft from the external report. Mail author/date/commit-message metadata was removed; the unified-diff hunks were retained. No patch was applied or executed, and no build/test or compatibility result is implied. The known gaps and acceptance criteria in this issue remain mandatory.

   Complete sanitized patch: 29,811 bytes, SHA-256 3d04e5a0f1b3fa31d8a1a374a634326dc24b426cad58322dd0ea2c5f0d5a1640. This part: 29,811 bytes, SHA-256 3d04e5a0f1b3fa31d8a1a374a634326dc24b426cad58322dd0ea2c5f0d5a1640. Parts concatenate in order; a part boundary is between complete file diffs.

   diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
   index 7e5fc6eb..0df7fd59 100644
   --- a/rust/src/runtime/backend.rs
   +++ b/rust/src/runtime/backend.rs
   @@ -185,10 +185,10 @@ pub(super) trait Backend: Database {
        fn document_row(row: &Self::Row, field: &str) -> Result<Value>;
        fn optional_document_row(row: &Self::Row, field: &str) -> Result<Option<Value>>;
        fn begin(pool: &Pool<Self>) -> impl Future<Output = Result<Transaction<'static, Self>>> + Send;
   -    /// The run's next history sequence, allocated in one statement under the
   -    /// transition lock, or `None` when it allocated none: no such run, or a
   -    /// sequence already at its largest.
   -    fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> impl Future<Output = Result<Option<i64>>> + Send;
   +    /// The last of the run's next `count` history sequences, allocated in one
   +    /// statement under the transition lock, or `None` when it allocated none:
   +    /// no such run, or a sequence too near its largest to take them all.
   +    fn next_history_sequences(connection: &mut Self::Connection, run_id: &str, count: i64) -> impl Future<Output = Result<Option<i64>>> + Send;
        fn history_ready(pool: &Pool<Self>) -> impl Future<Output = bool> + Send;
    }
    
   @@ -403,9 +403,11 @@ DELETE FROM dw_poll_receipts WHERE (namespace,worker_id,kind,request_id) IN (
        async fn begin(pool: &Pool<Self>) -> Result<Transaction<'static, Self>> {
            Ok(pool.begin_with("BEGIN IMMEDIATE").await?)
        }
   -    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
   -        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+1 WHERE id=$1 AND last_history_sequence<9223372036854775807 RETURNING last_history_sequence")
   +    async fn next_history_sequences(connection: &mut Self::Connection, run_id: &str, count: i64) -> Result<Option<i64>> {
   +        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+$1 WHERE id=$2 AND last_history_sequence<=9223372036854775807-$3 RETURNING last_history_sequence")
   +            .bind(count)
                .bind(run_id)
   +            .bind(count)
                .fetch_optional(connection)
                .await?
                .map(|row| Self::number(&row, "last_history_sequence"))
   @@ -531,9 +533,11 @@ impl Backend for Postgres {
            // round trip; a lock that fails rolls the transaction back.
            Ok(pool.begin_with("BEGIN; SELECT pg_advisory_xact_lock(4924501473315676163)").await?)
        }
   -    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
   -        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+1 WHERE id=$1 AND last_history_sequence<9223372036854775807 RETURNING last_history_sequence")
   +    async fn next_history_sequences(connection: &mut Self::Connection, run_id: &str, count: i64) -> Result<Option<i64>> {
   +        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+$1 WHERE id=$2 AND last_history_sequence<=9223372036854775807-$3 RETURNING last_history_sequence")
   +            .bind(count)
                .bind(run_id)
   +            .bind(count)
                .fetch_optional(connection)
                .await?
                .map(|row| Self::number(&row, "last_history_sequence"))
   @@ -785,11 +789,13 @@ impl Backend for MySql {
                .bind(mysql::VERSION).fetch_one(&mut *tx).await?;
            Ok(tx)
        }
   -    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
   +    async fn next_history_sequences(connection: &mut Self::Connection, run_id: &str, count: i64) -> Result<Option<i64>> {
            // MySQL has no UPDATE RETURNING: `LAST_INSERT_ID(expr)` returns the
            // incremented counter in the statement's own result.
   -        let result = sqlx::query("UPDATE workflow_runs SET last_history_sequence=LAST_INSERT_ID(last_history_sequence+1) WHERE id=? AND last_history_sequence<9223372036854775807")
   +        let result = sqlx::query("UPDATE workflow_runs SET last_history_sequence=LAST_INSERT_ID(last_history_sequence+?) WHERE id=? AND last_history_sequence<=9223372036854775807-?")
   +            .bind(count)
                .bind(run_id)
   +            .bind(count)
                .execute(connection)
                .await?;
            Ok((result.rows_affected() == 1).then(|| result.last_insert_id() as i64))
   diff --git a/rust/src/runtime/local_activities.rs b/rust/src/runtime/local_activities.rs
   index 1c00ac57..c91c9914 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<String>| 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<DateTime<Utc>> = 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,9 +1190,163 @@ 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 in INSERTs of up to `ROWS_PER_INSERT` rows, 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<()> {
   +        let writes = std::mem::take(&mut tx.local_writes);
   +        for rows in writes.executions.chunks(ROWS_PER_INSERT) {
   +            let sql = insert_rows::<DB>("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)",
   +                "(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", rows.len());
   +            let mut query = sqlx::query::<DB>(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 writes.attempts.chunks(ROWS_PER_INSERT) {
   +            let sql = insert_rows::<DB>("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)",
   +                "(?,?,?,?,?,?,?,?,?,?,NULL,?,?,?)", rows.len());
   +            let mut query = sqlx::query::<DB>(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 count = events.len() as i64;
   +            let sequences: Vec<i64> = match DB::next_history_sequences(&mut **tx, run_id, count).await? {
   +                Some(last) => (last - count + 1..=last).collect(),
   +                None => {
   +                    let mut sequences = Vec::with_capacity(events.len());
   +                    for _ in events {
   +                        sequences.push(Self::next_history_sequence(tx, run_id).await?);
   +                    }
   +                    sequences
   +                }
   +            };
   +            for (rows, sequences) in events.chunks(ROWS_PER_INSERT).zip(sequences.chunks(ROWS_PER_INSERT)) {
   +                let sql = insert_rows::<DB>("INSERT INTO workflow_history_events(id,workflow_run_id,sequence,event_type,payload,workflow_task_id,workflow_command_id,recorded_at)",
   +                    "(?,?,?,?,?,?,NULL,?)", rows.len());
   +                let mut query = sqlx::query::<DB>(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;
   +
   +/// 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<Execution>,
   +    attempts: Vec<AttemptRow>,
   +    events: Vec<Event>,
   +}
   +
   +struct Execution {
   +    id: String,
   +    run_id: String,
   +    sequence: i64,
   +    activity_type: String,
   +    status: &'static str,
   +    codec: String,
   +    arguments: Option<String>,
   +    result: Option<String>,
   +    exception: Option<String>,
   +    connection: Option<String>,
   +    queue: Option<String>,
   +    attempt_count: i64,
   +    attempt_id: String,
   +    retry_policy: Option<Value>,
   +    options: Value,
   +    started_at: DateTime<Utc>,
   +    at: DateTime<Utc>,
   +    last_heartbeat_at: Option<DateTime<Utc>>,
   +}
   +
   +struct AttemptRow {
   +    id: String,
   +    run_id: String,
   +    execution_id: String,
   +    task_id: String,
   +    worker_attempt_id: Option<String>,
   +    number: i64,
   +    status: &'static str,
   +    lease_owner: Option<String>,
   +    started_at: DateTime<Utc>,
   +    last_heartbeat_at: Option<DateTime<Utc>>,
   +    closed_at: DateTime<Utc>,
   +    at: DateTime<Utc>,
   +}
   +
   +struct Event {
   +    id: String,
   +    run_id: String,
   +    task_id: String,
   +    event_type: &'static str,
   +    payload: Value,
   +    recorded_at: DateTime<Utc>,
   +}
   +
   +/// `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<DB: sqlx::Database>(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)]
   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index 9755588a..88d0abc9 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -3371,6 +3371,11 @@ where
                .chain((0..commands.len()).filter(|index| is_terminal(&commands[*index])));
            for index in order {
                let command = &commands[index];
   +            // A run of local activity records is written as one before any
   +            // other command applies.
   +            if command["type"] != super::local_activities::COMMAND {
   +                Self::write_local_activities(&mut tx).await?;
   +            }
                if command["type"] == "fail_workflow" {
                    // Applied after the loop.
                    continue;
   @@ -3438,6 +3443,7 @@ where
                    }
                }
            }
   +        Self::write_local_activities(&mut tx).await?;
            let failed = if let Some(command) = commands
                .iter()
                .find(|command| command["type"] == "fail_workflow")
   @@ -4802,27 +4808,7 @@ where
            recorded_at: Option<DateTime<Utc>>,
            command_snapshot: bool,
        ) -> Result<String> {
   -        // The backend transition lock protects the allocation. One that
   -        // allocated nothing fails as this read of a missing or exhausted
   -        // sequence does.
   -        let sequence = match DB::next_history_sequence(&mut **tx, run_id).await? {
   -            Some(sequence) => sequence,
   -            None => {
   -                let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
   -                    .bind(run_id)
   -                    .fetch_one(&mut **tx)
   -                    .await?;
   -                let sequence = DB::number(&row, "last_history_sequence")?
   -                    .checked_add(1)
   -                    .ok_or_else(|| refuse(StatusCode::CONFLICT, "history_sequence_exhausted"))?;
   -                Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2")
   -                    .bind(sequence)
   -                    .bind(run_id)
   -                    .execute(&mut **tx)
   -                    .await?;
   -                sequence
   -            }
   -        };
   +        let sequence = Self::next_history_sequence(tx, run_id).await?;
            let recorded_at = recorded_at.unwrap_or_else(now);
            let mut payload = payload;
            // `WorkflowHistoryEvent::record`: an event PHP records with its task
   @@ -4851,12 +4837,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();
   @@ -4866,6 +4847,39 @@ where
            Ok(event_id)
        }
    
   +    /// The run's next history sequence. The backend transition lock protects
   +    /// the allocation. One that allocated nothing fails as this read of a
   +    /// missing or exhausted sequence does.
   +    pub(super) async fn next_history_sequence(tx: &mut Transaction<'_, DB>, run_id: &str) -> Result<i64> {
   +        if let Some(sequence) = DB::next_history_sequences(&mut **tx, run_id, 1).await? {
   +            return Ok(sequence);
   +        }
   +        let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
   +            .bind(run_id)
   +            .fetch_one(&mut **tx)
   +            .await?;
   +        let sequence = DB::number(&row, "last_history_sequence")?
   +            .checked_add(1)
   +            .ok_or_else(|| refuse(StatusCode::CONFLICT, "history_sequence_exhausted"))?;
   +        Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2")
   +            .bind(sequence)
   +            .bind(run_id)
   +            .execute(&mut **tx)
   +            .await?;
   +        Ok(sequence)
   +    }
   +
   +    /// The projections `Store::record` runs for an event it has stored.
   +    pub(super) async fn project_event(tx: &mut Transaction<'_, DB>, run_id: &str, event_type: &str, payload: &Value,
   +        recorded_at: DateTime<Utc>) -> 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 720c652b..e0cbd5d9 100644
   --- a/rust/src/runtime/transaction.rs
   +++ b/rust/src/runtime/transaction.rs
   @@ -34,6 +34,8 @@ pub(super) struct Transaction<'a, DB: Backend> {
        /// The tags of the runs this transaction has read whole, by run id
        /// (`Store::keep_run_tags`).
        pub(super) run_tags: HashMap<String, RunTags>,
   +    /// Local activity rows not yet written (`Store::write_local_activities`).
   +    pub(super) local_writes: super::local_activities::Writes,
    }
    
    impl<'a, DB: Backend> Transaction<'a, DB> {
   @@ -50,6 +52,7 @@ impl<'a, DB: Backend> Transaction<'a, DB> {
                scope_history: None,
                replayed_poll: None,
                run_tags: HashMap::new(),
   +            local_writes: Default::default(),
            }
        }
    
   @@ -118,6 +121,7 @@ where
            if let Some(refusal) = self.meter.as_ref().and_then(super::history_meter::Meter::tripped) {
                return Err(refusal);
            }
   +        Store::<DB>::write_local_activities(&mut self).await?;
            for run_id in std::mem::take(&mut self.summaries) {
                Store::<DB>::order_waits(&mut self, &run_id).await?;
                Store::<DB>::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 36aecc76..64399010 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<Value> = (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<i64> = events.iter().map(|event| event["sequence"].as_i64().unwrap()).collect();
   +    assert_eq!(sequences, (1..=events.len() as i64).collect::<Vec<_>>());
   +    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.
3. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized rebased reference for report topic 17: 0027-t17-Write-a-run-of-local-activity-records-with-one-state.patch.

   The supplied revision targets Server b8f4dd13623ce645de77ba5bd89ae8dd8eea2f7c. It is cumulative external reference material, not independently qualified against today's maintained source. Mail headers and commit-message/application instructions were removed; the unified diff is retained. The reporter now claims every rebased commit builds and the tip passes SQLite/MySQL/PostgreSQL tests. Those are attributed claims, not a new maintainer test result. Existing acceptance criteria, security findings and completed implementation receipts remain authoritative. Earlier attached patches are retained as historical references.

   Sanitized patch: 29,811 bytes, SHA-256 cb8670bef17cb6ac772b21bd7ba7d44dcafc8076c804e3a7b1480d9eceaf4f90. Original-base sanitized identity: 3d04e5a0f1b3fa31d8a1a374a634326dc24b426cad58322dd0ea2c5f0d5a1640. No patch was applied or executed during this update.

   diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
   index 7e5fc6eb..0df7fd59 100644
   --- a/rust/src/runtime/backend.rs
   +++ b/rust/src/runtime/backend.rs
   @@ -185,10 +185,10 @@ pub(super) trait Backend: Database {
        fn document_row(row: &Self::Row, field: &str) -> Result<Value>;
        fn optional_document_row(row: &Self::Row, field: &str) -> Result<Option<Value>>;
        fn begin(pool: &Pool<Self>) -> impl Future<Output = Result<Transaction<'static, Self>>> + Send;
   -    /// The run's next history sequence, allocated in one statement under the
   -    /// transition lock, or `None` when it allocated none: no such run, or a
   -    /// sequence already at its largest.
   -    fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> impl Future<Output = Result<Option<i64>>> + Send;
   +    /// The last of the run's next `count` history sequences, allocated in one
   +    /// statement under the transition lock, or `None` when it allocated none:
   +    /// no such run, or a sequence too near its largest to take them all.
   +    fn next_history_sequences(connection: &mut Self::Connection, run_id: &str, count: i64) -> impl Future<Output = Result<Option<i64>>> + Send;
        fn history_ready(pool: &Pool<Self>) -> impl Future<Output = bool> + Send;
    }
    
   @@ -403,9 +403,11 @@ DELETE FROM dw_poll_receipts WHERE (namespace,worker_id,kind,request_id) IN (
        async fn begin(pool: &Pool<Self>) -> Result<Transaction<'static, Self>> {
            Ok(pool.begin_with("BEGIN IMMEDIATE").await?)
        }
   -    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
   -        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+1 WHERE id=$1 AND last_history_sequence<9223372036854775807 RETURNING last_history_sequence")
   +    async fn next_history_sequences(connection: &mut Self::Connection, run_id: &str, count: i64) -> Result<Option<i64>> {
   +        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+$1 WHERE id=$2 AND last_history_sequence<=9223372036854775807-$3 RETURNING last_history_sequence")
   +            .bind(count)
                .bind(run_id)
   +            .bind(count)
                .fetch_optional(connection)
                .await?
                .map(|row| Self::number(&row, "last_history_sequence"))
   @@ -531,9 +533,11 @@ impl Backend for Postgres {
            // round trip; a lock that fails rolls the transaction back.
            Ok(pool.begin_with("BEGIN; SELECT pg_advisory_xact_lock(4924501473315676163)").await?)
        }
   -    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
   -        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+1 WHERE id=$1 AND last_history_sequence<9223372036854775807 RETURNING last_history_sequence")
   +    async fn next_history_sequences(connection: &mut Self::Connection, run_id: &str, count: i64) -> Result<Option<i64>> {
   +        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+$1 WHERE id=$2 AND last_history_sequence<=9223372036854775807-$3 RETURNING last_history_sequence")
   +            .bind(count)
                .bind(run_id)
   +            .bind(count)
                .fetch_optional(connection)
                .await?
                .map(|row| Self::number(&row, "last_history_sequence"))
   @@ -785,11 +789,13 @@ impl Backend for MySql {
                .bind(mysql::VERSION).fetch_one(&mut *tx).await?;
            Ok(tx)
        }
   -    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
   +    async fn next_history_sequences(connection: &mut Self::Connection, run_id: &str, count: i64) -> Result<Option<i64>> {
            // MySQL has no UPDATE RETURNING: `LAST_INSERT_ID(expr)` returns the
            // incremented counter in the statement's own result.
   -        let result = sqlx::query("UPDATE workflow_runs SET last_history_sequence=LAST_INSERT_ID(last_history_sequence+1) WHERE id=? AND last_history_sequence<9223372036854775807")
   +        let result = sqlx::query("UPDATE workflow_runs SET last_history_sequence=LAST_INSERT_ID(last_history_sequence+?) WHERE id=? AND last_history_sequence<=9223372036854775807-?")
   +            .bind(count)
                .bind(run_id)
   +            .bind(count)
                .execute(connection)
                .await?;
            Ok((result.rows_affected() == 1).then(|| result.last_insert_id() as i64))
   diff --git a/rust/src/runtime/local_activities.rs b/rust/src/runtime/local_activities.rs
   index 1c00ac57..c91c9914 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<String>| 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<DateTime<Utc>> = 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,9 +1190,163 @@ 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 in INSERTs of up to `ROWS_PER_INSERT` rows, 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<()> {
   +        let writes = std::mem::take(&mut tx.local_writes);
   +        for rows in writes.executions.chunks(ROWS_PER_INSERT) {
   +            let sql = insert_rows::<DB>("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)",
   +                "(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", rows.len());
   +            let mut query = sqlx::query::<DB>(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 writes.attempts.chunks(ROWS_PER_INSERT) {
   +            let sql = insert_rows::<DB>("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)",
   +                "(?,?,?,?,?,?,?,?,?,?,NULL,?,?,?)", rows.len());
   +            let mut query = sqlx::query::<DB>(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 count = events.len() as i64;
   +            let sequences: Vec<i64> = match DB::next_history_sequences(&mut **tx, run_id, count).await? {
   +                Some(last) => (last - count + 1..=last).collect(),
   +                None => {
   +                    let mut sequences = Vec::with_capacity(events.len());
   +                    for _ in events {
   +                        sequences.push(Self::next_history_sequence(tx, run_id).await?);
   +                    }
   +                    sequences
   +                }
   +            };
   +            for (rows, sequences) in events.chunks(ROWS_PER_INSERT).zip(sequences.chunks(ROWS_PER_INSERT)) {
   +                let sql = insert_rows::<DB>("INSERT INTO workflow_history_events(id,workflow_run_id,sequence,event_type,payload,workflow_task_id,workflow_command_id,recorded_at)",
   +                    "(?,?,?,?,?,?,NULL,?)", rows.len());
   +                let mut query = sqlx::query::<DB>(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;
   +
   +/// 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<Execution>,
   +    attempts: Vec<AttemptRow>,
   +    events: Vec<Event>,
   +}
   +
   +struct Execution {
   +    id: String,
   +    run_id: String,
   +    sequence: i64,
   +    activity_type: String,
   +    status: &'static str,
   +    codec: String,
   +    arguments: Option<String>,
   +    result: Option<String>,
   +    exception: Option<String>,
   +    connection: Option<String>,
   +    queue: Option<String>,
   +    attempt_count: i64,
   +    attempt_id: String,
   +    retry_policy: Option<Value>,
   +    options: Value,
   +    started_at: DateTime<Utc>,
   +    at: DateTime<Utc>,
   +    last_heartbeat_at: Option<DateTime<Utc>>,
   +}
   +
   +struct AttemptRow {
   +    id: String,
   +    run_id: String,
   +    execution_id: String,
   +    task_id: String,
   +    worker_attempt_id: Option<String>,
   +    number: i64,
   +    status: &'static str,
   +    lease_owner: Option<String>,
   +    started_at: DateTime<Utc>,
   +    last_heartbeat_at: Option<DateTime<Utc>>,
   +    closed_at: DateTime<Utc>,
   +    at: DateTime<Utc>,
   +}
   +
   +struct Event {
   +    id: String,
   +    run_id: String,
   +    task_id: String,
   +    event_type: &'static str,
   +    payload: Value,
   +    recorded_at: DateTime<Utc>,
   +}
   +
   +/// `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<DB: sqlx::Database>(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)]
   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index b6245837..7ad5a108 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -3384,6 +3384,11 @@ where
                .chain((0..commands.len()).filter(|index| is_terminal(&commands[*index])));
            for index in order {
                let command = &commands[index];
   +            // A run of local activity records is written as one before any
   +            // other command applies.
   +            if command["type"] != super::local_activities::COMMAND {
   +                Self::write_local_activities(&mut tx).await?;
   +            }
                if command["type"] == "fail_workflow" {
                    // Applied after the loop.
                    continue;
   @@ -3451,6 +3456,7 @@ where
                    }
                }
            }
   +        Self::write_local_activities(&mut tx).await?;
            let failed = if let Some(command) = commands
                .iter()
                .find(|command| command["type"] == "fail_workflow")
   @@ -4816,27 +4822,7 @@ where
            recorded_at: Option<DateTime<Utc>>,
            command_snapshot: bool,
        ) -> Result<String> {
   -        // The backend transition lock protects the allocation. One that
   -        // allocated nothing fails as this read of a missing or exhausted
   -        // sequence does.
   -        let sequence = match DB::next_history_sequence(&mut **tx, run_id).await? {
   -            Some(sequence) => sequence,
   -            None => {
   -                let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
   -                    .bind(run_id)
   -                    .fetch_one(&mut **tx)
   -                    .await?;
   -                let sequence = DB::number(&row, "last_history_sequence")?
   -                    .checked_add(1)
   -                    .ok_or_else(|| refuse(StatusCode::CONFLICT, "history_sequence_exhausted"))?;
   -                Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2")
   -                    .bind(sequence)
   -                    .bind(run_id)
   -                    .execute(&mut **tx)
   -                    .await?;
   -                sequence
   -            }
   -        };
   +        let sequence = Self::next_history_sequence(tx, run_id).await?;
            let recorded_at = recorded_at.unwrap_or_else(now);
            let mut payload = payload;
            // `WorkflowHistoryEvent::record`: an event PHP records with its task
   @@ -4865,12 +4851,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();
   @@ -4880,6 +4861,39 @@ where
            Ok(event_id)
        }
    
   +    /// The run's next history sequence. The backend transition lock protects
   +    /// the allocation. One that allocated nothing fails as this read of a
   +    /// missing or exhausted sequence does.
   +    pub(super) async fn next_history_sequence(tx: &mut Transaction<'_, DB>, run_id: &str) -> Result<i64> {
   +        if let Some(sequence) = DB::next_history_sequences(&mut **tx, run_id, 1).await? {
   +            return Ok(sequence);
   +        }
   +        let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
   +            .bind(run_id)
   +            .fetch_one(&mut **tx)
   +            .await?;
   +        let sequence = DB::number(&row, "last_history_sequence")?
   +            .checked_add(1)
   +            .ok_or_else(|| refuse(StatusCode::CONFLICT, "history_sequence_exhausted"))?;
   +        Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2")
   +            .bind(sequence)
   +            .bind(run_id)
   +            .execute(&mut **tx)
   +            .await?;
   +        Ok(sequence)
   +    }
   +
   +    /// The projections `Store::record` runs for an event it has stored.
   +    pub(super) async fn project_event(tx: &mut Transaction<'_, DB>, run_id: &str, event_type: &str, payload: &Value,
   +        recorded_at: DateTime<Utc>) -> 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 720c652b..e0cbd5d9 100644
   --- a/rust/src/runtime/transaction.rs
   +++ b/rust/src/runtime/transaction.rs
   @@ -34,6 +34,8 @@ pub(super) struct Transaction<'a, DB: Backend> {
        /// The tags of the runs this transaction has read whole, by run id
        /// (`Store::keep_run_tags`).
        pub(super) run_tags: HashMap<String, RunTags>,
   +    /// Local activity rows not yet written (`Store::write_local_activities`).
   +    pub(super) local_writes: super::local_activities::Writes,
    }
    
    impl<'a, DB: Backend> Transaction<'a, DB> {
   @@ -50,6 +52,7 @@ impl<'a, DB: Backend> Transaction<'a, DB> {
                scope_history: None,
                replayed_poll: None,
                run_tags: HashMap::new(),
   +            local_writes: Default::default(),
            }
        }
    
   @@ -118,6 +121,7 @@ where
            if let Some(refusal) = self.meter.as_ref().and_then(super::history_meter::Meter::tripped) {
                return Err(refusal);
            }
   +        Store::<DB>::write_local_activities(&mut self).await?;
            for run_id in std::mem::take(&mut self.summaries) {
                Store::<DB>::order_waits(&mut self, &run_id).await?;
                Store::<DB>::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 36aecc76..64399010 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<Value> = (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<i64> = events.iter().map(|event| event["sequence"].as_i64().unwrap()).collect();
   +    assert_eq!(sequences, (1..=events.len() as i64).collect::<Vec<_>>());
   +    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.
4. rmcdaniel commented on Nov 8, 2026

   

   rmcdaniel

   on Nov 8, 2026

   MemberAuthor

   More actions

   Operator nudge: this benchmax ticket (P0) has had no linked PR and no activity for about 32 h. Please pick it up next, after the current work on #1057. Suggested order for the stalled server tickets: #1063, #1068, #1075, #1077, #1078, #1079, then #1081 (P2); this one is 2 of 7. Open a draft PR as soon as work starts so the tracker shows it. If it is blocked, add the status:blocked label with a one-line reason.
5. rmcdaniel

   added

   agent:general

   and removed

   agent:server

   on Nov 8, 2026
6. rmcdaniel commented on Nov 9, 2026

   

   rmcdaniel

   on Nov 9, 2026

   MemberAuthor

   More actions

   General has started the maintained-source implementation for this P0, independently of the frozen3.3.1 release cut. The original WIP refs remain preserved.

   The reference draft does not apply to current backend/store/transaction code. Its local-record and order-test hunks have been imported into a separate worktree and are being ported. The new range allocator keeps the existing single-event allocator and typed fallback; pending rows flush at command/checkpoint/history-read/commit boundaries. Multi-row writes now have both the100-row limit and a1MiB conservative encoded-size target. A previously legal oversized single row stays a single-row write.

   Next: physical statement-count controls, large-result splitting and oversized-row coverage, sequence-boundary/fallback equivalence, and the unchanged whole-completion persistence-budget refusal/rollback behavior. Then a coherent draft PR and normal hosted compile/conformance. No code has been compiled or pushed yet, and this work is not in3.3.1 or a benchmark result.
7. rmcdaniel commented on Nov 9, 2026

   

   rmcdaniel

   on Nov 9, 2026

   MemberAuthor

   More actions

   General selects one ordinary draft source train for #1068's maintained-source batching implementation. #1109 has merged; its subject-only main checks remain under observation. There is one new heavy source train, with no P0 queue or release selected.

   Head 514dfcdabd316870afeea275fff82e7a3bfec5b9, sole parent/current main 603f9f5cbf4684fc3fba7dbf20b4ee115fe40453, tree 3d69cdac3ab5480b2cd526ef3f89bcc95f463c8d. Eleven Rust files,536additions/23deletions; full binary delta 49041B / SHA-256 043f24bfcb6ea8dccde0d526ec3b768ee5f9a47a604d448f24270ccff135f29e. The rebase over the runner-subject merge is byte-identical in its Rust delta. Preserved operator WIP refs are untouched.

   Implementation: hold local execution/attempt/history rows; flush before other commands, history-dependent scope reads, poll savepoints and commit; discard held state on rollback. Multi-row statements have100-row and conservative1MiB encoded-size bounds. A legal oversized single row stays a single-row write. A separate range allocator retains backend counter limits, SQLite canonical-type protection and the unchanged single-event typed fallback. Logical completion-budget charges remain unchanged.

   Added controls:40-record order with a side effect; physical execution/attempt/history INSERT and range counts; large-result byte splitting; oversized legal rows; event/row/byte budget refusal before local writes with the claim usable; range-boundary, missing/noncanonical and zero-count controls. Existing retry/heartbeat/failure and full source corpus coverage remain.

   Only source/whitespace/secret/ancestry checks ran here. Uncompiled and unexecuted locally; normal hosted CI is the first compile and database execution. One must-be-absent branch push and agent:general draft PR. Refuse main/head/diff drift, existing branch/PR or a secret finding. Observe all checks, including SQLite, PG, MySQL and MariaDB. Preserve any NONPASS; no automatic rerun, cancellation or successor push. This is for a later patch, with no benchmark speedup or3.3.1-content claim.
8. rmcdaniel

   closed this as completedon Nov 9, 2026