From b07164e843c86586229f77a9393dec62e84faa87 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 03:22:29 +0000 Subject: [PATCH 1/2] Give codec work under the transition lock its own permit Five transitions run Avro work while holding the global transition lock: the rejected-signal command, signal arguments applied by a resumed task, cooperative cancellation requests, task repair and fenced deregistration. They shared the single codec permit with describe, query and listing reads, so under read load a transition waited on the lock for unrelated decodes. A per-call-site probe on the CPU deadline-health cell measured 0.58 to 0.80 s of such waiting in a cancellation request. These transitions now take a separate transition_codec permit. Reads keep signal_codec. Encodings, values and refusals are unchanged. New tests hold the read permit and require a cancellation request, a signal-resumed completion, a fenced deregistration and a range repair to finish within one second. They time out with the old wiring. Part of #325. Co-Authored-By: Claude Opus 5.5 --- docs/rust-server-transition-locking.md | 25 ++- rust/src/runtime/execution.rs | 41 +++-- rust/src/runtime/store.rs | 12 +- rust/src/runtime/transition_codec_tests.rs | 202 +++++++++++++++++++++ 4 files changed, 254 insertions(+), 26 deletions(-) create mode 100644 rust/src/runtime/transition_codec_tests.rs diff --git a/docs/rust-server-transition-locking.md b/docs/rust-server-transition-locking.md index e593eab1a..89eeabe65 100644 --- a/docs/rust-server-transition-locking.md +++ b/docs/rust-server-transition-locking.md @@ -36,6 +36,21 @@ scheduler from sleeping after each pass to a fixed tick made no measurable difference. Deadline health and throughput under load are bounded by how long other transitions hold the lock. +A scratch build that logged lock wait and hold per call site then ran the same +cell on a host shared with other work, so absolute times are approximate: + +- A cancellation request held the lock 0.65 to 0.86 s, almost all of it + waiting for the single codec permit behind describe decodes of the dense + input (0.58 to 0.80 s). With a permit of its own the hold fell to 0.05 to + 0.09 s. Codec work under the lock now has its own permit. +- Activity status checks spent about 1.6 s per run waiting for the lock. They + no longer take it. +- Cancellation delivery and expiry, workflow completions, heartbeats and + activity expiry held the lock 0.2 to 0.9 s during the read wave without codec + work, several times longer than the same transitions before it. Their summary + projections took under 30 ms each. Lowering the decode thread's CPU priority + changed nothing measurable. This remaining cost is not yet attributed. + ## What lengthens lock hold time - **Summary projection at commit.** `Transaction::commit` reorders open waits @@ -43,10 +58,12 @@ other transitions hold the lock. `summary_history` pages through the run's entire history to do so (`terminal_summary.rs`). Every heartbeat and claim therefore costs time proportional to history length while holding the lock. -- **Codec work inside transactions.** Several paths take the single codec - permit and run Avro work on the blocking pool while the transaction is open, - for example the rejected-signal path, continue-as-new, cooperative requests - and repair. +- **Codec work inside transactions.** Five paths run Avro work on the blocking + pool while the transaction is open: the rejected-signal command, signal + arguments applied by a resumed workflow task, cooperative cancellation + requests, task repair and fenced worker deregistration. They take + `transition_codec`, never the read permit `signal_codec`, so they do not + queue behind reads. The CPU work itself still runs under the lock. - **Claim reads.** Workflow claims load the first history page (the requested page size, 100 by default) and compute the history budget under the lock. Query claims parse cached query JSON values. diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index edb69ac8e..5715137b8 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -25,6 +25,10 @@ enum Storage { #[path = "query_codec_tests.rs"] mod query_codec_tests; +#[cfg(test)] +#[path = "transition_codec_tests.rs"] +mod transition_codec_tests; + #[derive(Clone)] pub struct Runtime { storage: Arc, @@ -32,6 +36,10 @@ pub struct Runtime { pub(crate) role_tokens: [Option>; 3], scheduler: Arc, signal_codec: Arc, + // Codec work done while holding the transition lock takes this permit, + // never `signal_codec`, so a transition never queues behind reads for + // codec capacity. + transition_codec: Arc, pub(super) external: Arc, queries: Arc, } @@ -194,6 +202,7 @@ impl Runtime { role_tokens: [None, None, None], scheduler: Arc::new(TimerScheduler::new()), signal_codec: Arc::new(Semaphore::new(1)), + transition_codec: Arc::new(Semaphore::new(1)), external: Arc::new(external), queries: Arc::new(queries), }; @@ -262,6 +271,7 @@ impl Runtime { role_tokens: self.role_tokens.clone(), scheduler: self.scheduler.clone(), signal_codec: self.signal_codec.clone(), + transition_codec: self.transition_codec.clone(), external: self.external.clone(), queries: self.queries.clone(), } @@ -376,7 +386,7 @@ impl Runtime { name, body, prepared, - self.signal_codec.clone(), + self.transition_codec.clone(), ) .await } @@ -388,7 +398,7 @@ impl Runtime { name, body, prepared, - self.signal_codec.clone(), + self.transition_codec.clone(), ) .await } @@ -400,7 +410,7 @@ impl Runtime { name, body, prepared, - self.signal_codec.clone(), + self.transition_codec.clone(), ) .await } @@ -421,7 +431,7 @@ impl Runtime { run_id, body, context, - self.signal_codec.clone(), + self.transition_codec.clone(), ) .await } @@ -432,7 +442,7 @@ impl Runtime { run_id, body, context, - self.signal_codec.clone(), + self.transition_codec.clone(), ) .await } @@ -443,7 +453,7 @@ impl Runtime { run_id, body, context, - self.signal_codec.clone(), + self.transition_codec.clone(), ) .await } @@ -658,9 +668,8 @@ impl Runtime { ) -> Result<(Vec, Vec)> ); pub(crate) async fn activities(&self, workflow_id: &str, run_id: &str) -> Result { - // Writers take this permit while holding the writer transaction. Finish - // every read before taking it so a listing can never wait on a writer - // that is itself waiting on the permit. + // Finish every read before taking the codec permit, so a listing never + // holds codec capacity while it waits on the database. let (events, executions) = self.activity_metadata_input(workflow_id, run_id).await?; let permit = self .signal_codec @@ -700,7 +709,7 @@ impl Runtime { delegate!(worker_heartbeat(body: Value) -> Result); delegate!(deregister(worker_id: &str) -> Result); pub(crate) async fn deregister_fenced(&self, worker_id: &str, body: Value) -> Result { - let codec = self.signal_codec.clone(); + let codec = self.transition_codec.clone(); match &*self.storage { Storage::Sqlite(store) => store.deregister_fenced(worker_id, body, codec).await, Storage::Postgres(store) => store.deregister_fenced(worker_id, body, codec).await, @@ -761,17 +770,17 @@ impl Runtime { match &*self.storage { Storage::Sqlite(store) => { store - .complete_workflow(task_id, body, self.signal_codec.clone(), external) + .complete_workflow(task_id, body, self.transition_codec.clone(), external) .await } Storage::Postgres(store) => { store - .complete_workflow(task_id, body, self.signal_codec.clone(), external) + .complete_workflow(task_id, body, self.transition_codec.clone(), external) .await } Storage::MySql(store) => { store - .complete_workflow(task_id, body, self.signal_codec.clone(), external) + .complete_workflow(task_id, body, self.transition_codec.clone(), external) .await } } @@ -817,7 +826,7 @@ impl Runtime { run_id, reason, budget, - self.signal_codec.clone(), + self.transition_codec.clone(), ) .await } @@ -828,7 +837,7 @@ impl Runtime { run_id, reason, budget, - self.signal_codec.clone(), + self.transition_codec.clone(), ) .await } @@ -839,7 +848,7 @@ impl Runtime { run_id, reason, budget, - self.signal_codec.clone(), + self.transition_codec.clone(), ) .await } diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 6490d0f95..24ae515ad 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -402,7 +402,7 @@ where name: &str, body: Value, mut prepared: PreparedControlArguments, - signal_codec: Arc, + transition_codec: Arc, ) -> Result { reject_fields(&body, &["input", "request_id"])?; if name.trim().is_empty() || name.len() > 255 { @@ -505,7 +505,7 @@ where "message":format!("Workflow signal [{name}] is unknown for workflow type [{workflow_type}]: the durable command contract does not declare that handler. Command contract source [durable_history], declared signals [{labels}].")}); } prepared.command = prepare_rejected_signal_command( - signal_codec, + transition_codec, name.to_owned(), prepared.values.clone(), validation_errors.clone(), @@ -696,7 +696,7 @@ where run_id: &str, task_id: &str, task: &DB::Row, - signal_codec: Arc, + transition_codec: Arc, ) -> Result<()> { let payload = DB::document_row(task, "payload")?; if payload["resume_source_kind"] != "workflow_signal" { @@ -723,7 +723,7 @@ where return Err(refuse(StatusCode::CONFLICT, "signal_wait_mismatch")); } let prepared = prepare_control_arguments( - signal_codec, + transition_codec, name.clone(), DB::string(&signal, "arguments")?, "invalid_signal_arguments", @@ -1499,7 +1499,7 @@ where &self, task_id: &str, body: Value, - signal_codec: Arc, + transition_codec: Arc, external: super::external_resolution::PreparedCommands, ) -> Result { let commands = body["commands"] @@ -1699,7 +1699,7 @@ where return Err(refuse(StatusCode::CONFLICT, "update_task_mismatch")); } let mut sequence = Self::authored_sequence(&mut tx, &run_id).await?; - Self::apply_task_signal(&mut tx, &run_id, task_id, &task, signal_codec).await?; + Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec).await?; for (index, command) in commands.iter().enumerate() { if command["type"] == "fail_workflow" { // Published parseCommands applies every nonterminal command diff --git a/rust/src/runtime/transition_codec_tests.rs b/rust/src/runtime/transition_codec_tests.rs new file mode 100644 index 000000000..becf90d87 --- /dev/null +++ b/rust/src/runtime/transition_codec_tests.rs @@ -0,0 +1,202 @@ +//! Transitions that encode or decode while holding the transition lock use +//! their own codec permit. Each test holds the read codec permit, as a long +//! describe decode would, and requires the transition to finish anyway. +use super::*; +use crate::codec::{Value as Payload, ValueCodec}; +use serde_json::json; + +fn envelope(value: Payload) -> Value { + json!({"codec":"avro","blob":ValueCodec::new().unwrap().encode(&value).unwrap()}) +} + +const DECISIVE: Duration = Duration::from_secs(1); + +async fn runtime(directory: &tempfile::TempDir) -> Runtime { + let runtime = Runtime::open( + directory.path().join("runtime.sqlite").to_str().unwrap(), + "test-token".into(), + ) + .await + .unwrap(); + runtime + .register(json!({"worker_id":"codec-worker","task_queue":"test","runtime":"php", + "supported_workflow_types":["echo"],"supported_activity_types":[], + "workflow_command_contracts":{"echo":{"signals":["payload"],"signal_contracts":[{"name":"payload", + "parameters":[{"name":"value","position":0,"type":"array","required":true,"variadic":false, + "default_available":false,"allows_null":false}]}]}}})) + .await + .unwrap(); + runtime +} + +async fn start(runtime: &Runtime, workflow_id: &str) { + runtime + .start(json!({"workflow_id":workflow_id,"workflow_type":"echo","task_queue":"test", + "input":envelope(Payload::Array(vec![]))})) + .await + .unwrap(); +} + +async fn poll(runtime: &Runtime) -> Value { + let task = runtime + .poll( + json!({"worker_id":"codec-worker","task_queue":"test","timeout_seconds":0}), + "workflow", + "1.20", + ) + .await + .unwrap()["task"] + .clone(); + assert!(!task.is_null(), "workflow task is ready"); + task +} + +async fn complete(runtime: &Runtime, task: &Value, commands: Value) -> Result { + runtime + .complete_workflow( + task["task_id"].as_str().unwrap(), + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"], + "commands":commands}), + ) + .await +} + +#[tokio::test] +async fn cancellation_request_does_not_wait_for_read_codec_capacity() { + let directory = tempfile::tempdir().unwrap(); + let runtime = runtime(&directory).await; + start(&runtime, "codec-cancel").await; + let _held = runtime.signal_codec.clone().acquire_owned().await.unwrap(); + let (status, accepted) = tokio::time::timeout( + DECISIVE, + runtime.request_cancellation( + "codec-cancel", + None, + json!({"reason":"held read permit","cleanup_timeout_seconds":30}), + ), + ) + .await + .expect("the cancellation encode must not wait for read codec capacity") + .unwrap(); + assert_eq!(status, StatusCode::ACCEPTED, "{accepted}"); + runtime.close().await; +} + +#[tokio::test] +async fn signal_resumed_completion_does_not_wait_for_read_codec_capacity() { + let directory = tempfile::tempdir().unwrap(); + let runtime = runtime(&directory).await; + start(&runtime, "codec-signal").await; + let waiting = poll(&runtime).await; + complete( + &runtime, + &waiting, + json!([{"type":"open_signal_wait","signal_name":"payload"}]), + ) + .await + .unwrap(); + runtime + .signal( + "codec-signal", + None, + "payload", + json!({"input":envelope(Payload::Array(vec![Payload::Map( + std::collections::BTreeMap::from([("count".into(), Payload::Long(7))]), + )]))}), + ) + .await + .unwrap(); + let resumed = poll(&runtime).await; + let _held = runtime.signal_codec.clone().acquire_owned().await.unwrap(); + let completed = tokio::time::timeout( + DECISIVE, + complete( + &runtime, + &resumed, + json!([{"type":"complete_workflow","result":envelope(Payload::Long(7))}]), + ), + ) + .await + .expect("applying the signal must not wait for read codec capacity") + .unwrap(); + assert_eq!(completed["recorded"], true, "{completed}"); + runtime.close().await; +} + +#[tokio::test] +async fn fenced_deregistration_repair_does_not_wait_for_read_codec_capacity() { + let directory = tempfile::tempdir().unwrap(); + let runtime = Runtime::open( + directory.path().join("runtime.sqlite").to_str().unwrap(), + "test-token".into(), + ) + .await + .unwrap(); + let registration = runtime + .register(json!({"worker_id":"codec-worker","task_queue":"test","runtime":"php", + "supported_workflow_types":["echo"],"supported_activity_types":[]})) + .await + .unwrap(); + start(&runtime, "codec-repair").await; + poll(&runtime).await; + let _held = runtime.signal_codec.clone().acquire_owned().await.unwrap(); + let deregistered = tokio::time::timeout( + DECISIVE, + runtime.deregister_fenced( + "codec-worker", + json!({"registration_token":registration["registration_token"]}), + ), + ) + .await + .expect("repairing the leased task must not wait for read codec capacity") + .unwrap(); + assert_eq!(deregistered["recovered_workflow_task_count"], 1, "{deregistered}"); + runtime.close().await; +} + +#[tokio::test] +async fn range_repair_does_not_wait_for_read_codec_capacity() { + use axum::{ + body::{Body, to_bytes}, + http::Request, + }; + use tower::ServiceExt; + let directory = tempfile::tempdir().unwrap(); + let runtime = runtime(&directory).await; + start(&runtime, "codec-range").await; + let task = poll(&runtime).await; + let reported = runtime + .fail_workflow_task( + task["task_id"].as_str().unwrap(), + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"], + "failure":{"message":"Recorded version 1 is outside supported range 2..2.", + "type":"DurableWorkflow\\Exception\\NonDeterministicWorkflow", + "reason":"version_marker_incompatible_range","sequence":1}}), + ) + .await + .unwrap(); + assert_eq!(reported["recorded"], true, "{reported}"); + // The repair context comes from the authenticated request, so use the route. + let request = Request::builder() + .method("POST") + .uri("/api/workflows/codec-range/repair") + .header("authorization", "Bearer test-token") + .header("content-type", "application/json") + .header("x-namespace", "default") + .header("x-durable-workflow-control-plane-version", "2") + .body(Body::empty()) + .unwrap(); + let _held = runtime.signal_codec.clone().acquire_owned().await.unwrap(); + let response = tokio::time::timeout( + DECISIVE, + super::super::router(runtime.clone()).oneshot(request), + ) + .await + .expect("the repair payload encode must not wait for read codec capacity") + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let repaired: Value = + serde_json::from_slice(&to_bytes(response.into_body(), 1 << 20).await.unwrap()).unwrap(); + assert_eq!(repaired["outcome"], "repair_dispatched", "{repaired}"); + runtime.close().await; +} From 7e4919313a0d5ac0512e22c5e181988a4ddb9902 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 07:50:39 +0000 Subject: [PATCH 2/2] Import the codec test names explicitly Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/transition_codec_tests.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/rust/src/runtime/transition_codec_tests.rs b/rust/src/runtime/transition_codec_tests.rs index becf90d87..9f02385d6 100644 --- a/rust/src/runtime/transition_codec_tests.rs +++ b/rust/src/runtime/transition_codec_tests.rs @@ -1,9 +1,11 @@ //! Transitions that encode or decode while holding the transition lock use //! their own codec permit. Each test holds the read codec permit, as a long //! describe decode would, and requires the transition to finish anyway. -use super::*; +use super::{Result, Runtime}; use crate::codec::{Value as Payload, ValueCodec}; -use serde_json::json; +use axum::http::StatusCode; +use serde_json::{Value, json}; +use std::time::Duration; fn envelope(value: Payload) -> Value { json!({"codec":"avro","blob":ValueCodec::new().unwrap().encode(&value).unwrap()})