Give codec work under the transition lock its own permit

docs/rust-server-transition-locking.md

@@ -46,17 +46,34 @@ 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
   and re-projects the run summary for each touched run (`transaction.rs`).
   `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.

rust/src/runtime/execution.rs

@@ -25,13 +25,21 @@ 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<Storage>,
     pub(crate) token: Arc<str>,
     pub(crate) role_tokens: [Option<Arc<str>>; 3],
     scheduler: Arc<TimerScheduler>,
     signal_codec: Arc<Semaphore>,
+    // 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<Semaphore>,
     pub(super) external: Arc<super::ExternalPayloadConfig>,
     queries: Arc<super::QueryTaskConfig>,
 }
@@ -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<Value>, Vec<Value>)>
     );
     pub(crate) async fn activities(&self, workflow_id: &str, run_id: &str) -> Result<Value> {
-        // 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<Value>);
     delegate!(deregister(worker_id: &str) -> Result<Value>);
     pub(crate) async fn deregister_fenced(&self, worker_id: &str, body: Value) -> Result<Value> {
-        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
             }

rust/src/runtime/store.rs

@@ -402,7 +402,7 @@ where
         name: &str,
         body: Value,
         mut prepared: PreparedControlArguments,
-        signal_codec: Arc<Semaphore>,
+        transition_codec: Arc<Semaphore>,
     ) -> Result<Value> {
         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<Semaphore>,
+        transition_codec: Arc<Semaphore>,
     ) -> 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",
@@ -1545,7 +1545,7 @@ where
         &self,
         task_id: &str,
         body: Value,
-        signal_codec: Arc<Semaphore>,
+        transition_codec: Arc<Semaphore>,
         external: super::external_resolution::PreparedCommands,
     ) -> Result<Value> {
         let commands = body["commands"]
@@ -1745,7 +1745,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