From 7a4ed05a6a0a152246a4d403ccedbf51c58b39c4 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 26 Oct 2026 13:56:03 +0000 Subject: [PATCH 1/2] Store chronological wait positions and write only waits that moved Native ranked only open waits and rewrote every open wait's position on each transition, even when nothing moved. Resolved waits kept stale ordinals. Workflow #941's projector stores one chronological ordinal over every retained wait (opened milliseconds, then wait id) and applies the status ranking only when reading. Rank all of the run's waits in the database and return only rows whose stored position differs, then update those rows by primary key. Resolving a wait no longer moves any other wait, and a new wait is normally inserted at its final ordinal, so most transitions write no positions. Addresses #783. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/backend.rs | 19 ++-- rust/src/runtime/transaction.rs | 2 +- rust/src/runtime/wait_projection.rs | 31 ++++-- rust/tests/execution.rs | 2 + rust/tests/execution/wait_positions.rs | 139 +++++++++++++++++++++++++ 5 files changed, 174 insertions(+), 19 deletions(-) create mode 100644 rust/tests/execution/wait_positions.rs diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 1229d146c..ba98fea67 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -19,7 +19,9 @@ pub(super) trait Backend: Database { // deferred read transaction already keeps one WAL snapshot. const READ_SNAPSHOT_BEGIN: &str = "BEGIN"; const ACTIVITY_METADATA_HISTORY: &str = "SELECT sequence,event_type,CASE WHEN projection_bytes<=8388608 THEN payload ELSE NULL END AS payload,workflow_task_id,recorded_at,projection_count FROM (SELECT sequence,event_type,payload,workflow_task_id,recorded_at,COUNT(*) OVER () AS projection_count,SUM(octet_length(CAST(payload AS TEXT))) OVER (ORDER BY sequence) AS projection_bytes FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') ORDER BY sequence LIMIT 10001) AS page ORDER BY sequence"; - const OPEN_WAIT_ORDER: &str = "WITH clocks AS (SELECT id,wait_id,opened_at,EXTRACT(EPOCH FROM opened_at) AS seconds FROM workflow_run_waits WHERE workflow_run_id=$1 AND status='open') SELECT id FROM clocks ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN seconds<0 THEN 2*FLOOR(seconds)-seconds ELSE seconds END)*1000),wait_id LIMIT 100 OFFSET $2"; + // Waits whose stored position differs from their chronological ordinal + // over every retained wait of the run, as Workflow #941's projector stores. + const WAIT_POSITION_MOVES: &str = "WITH clocks AS (SELECT id,wait_id,position,opened_at,EXTRACT(EPOCH FROM opened_at) AS seconds FROM workflow_run_waits WHERE workflow_run_id=$1), ordinals AS (SELECT id,position,ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN seconds<0 THEN 2*FLOOR(seconds)-seconds ELSE seconds END)*1000),wait_id)-1 AS ordinal FROM clocks) SELECT id,ordinal FROM ordinals WHERE position IS DISTINCT FROM ordinal ORDER BY ordinal LIMIT 100"; const SUMMARY_HISTORY_SQL: &str = "SELECT id,sequence,event_type,CAST(payload AS TEXT) AS payload_json FROM workflow_history_events WHERE workflow_run_id=$1 AND sequence>$2 ORDER BY sequence LIMIT 16"; const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND payload->>'failure_id'=$2 AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1"; const OBSERVATION_ATTRIBUTES: &str = "SELECT \"key\" AS attribute_key,type,value_string,value_keyword,value_int,value_float,CASE WHEN value_bool IS NULL THEN NULL WHEN value_bool THEN 1 ELSE 0 END AS value_bool,value_datetime,value_keyword_list FROM workflow_search_attributes WHERE workflow_run_id=$1 AND workflow_instance_id=$2 AND \"key\" IN ($3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22) ORDER BY \"key\" LIMIT 20"; @@ -157,20 +159,23 @@ impl Backend for Sqlite { // whole UTC seconds with SQLite's clock functions and round integer // microseconds explicitly; SQLite's fractional-date rounding differs at ties. // Preserve Carbon's signed U.u decimal key for pre-epoch dates as well. - const OPEN_WAIT_ORDER: &str = r#" + const WAIT_POSITION_MOVES: &str = r#" WITH utc_waits AS ( - SELECT id,wait_id,opened_at, + SELECT id,wait_id,position,opened_at, replace(replace(replace(opened_at,'T',' '),'Z',''),'+00:00','') AS utc_opened - FROM workflow_run_waits WHERE workflow_run_id=$1 AND status='open' + FROM workflow_run_waits WHERE workflow_run_id=$1 ), clocks AS ( SELECT *,unixepoch(substr(utc_opened,1,19))*1000000 +(CASE WHEN unixepoch(substr(utc_opened,1,19))<0 THEN -1 ELSE 1 END) *CAST(CASE WHEN substr(utc_opened,20,1)='.' THEN substr(substr(utc_opened,21)||'000000',1,6) ELSE '000000' END AS INTEGER) AS micros FROM utc_waits +), ordinals AS ( + SELECT id,position,ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END, + CASE WHEN micros<0 THEN (micros-500)/1000 ELSE (micros+500)/1000 END,wait_id)-1 AS ordinal + FROM clocks ) -SELECT id FROM clocks ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END, - CASE WHEN micros<0 THEN (micros-500)/1000 ELSE (micros+500)/1000 END,wait_id LIMIT 100 OFFSET $2 +SELECT id,ordinal FROM ordinals WHERE position IS NOT ordinal ORDER BY ordinal LIMIT 100 "#; const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND json_extract(payload,'$.failure_id') COLLATE BINARY=$2 AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1"; const ACTIVITY_RESULT_EVENT_SQL: &str = "SELECT sequence,event_type,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') AND (json_extract(payload,'$.activity_execution_id') COLLATE BINARY=$2 OR json_extract(payload,'$.activity.id') COLLATE BINARY=$3) ORDER BY sequence DESC LIMIT 1"; @@ -450,7 +455,7 @@ impl Backend for MySql { const READ_SNAPSHOT_BEGIN: &str = "START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY"; const ACTIVITY_METADATA_HISTORY: &str = "SELECT h.sequence,h.event_type,CASE WHEN page.projection_bytes<=8388608 THEN h.payload ELSE NULL END AS payload,h.workflow_task_id,h.recorded_at,page.projection_count FROM (SELECT id,sequence,COUNT(*) OVER () AS projection_count,SUM(payload_bytes) OVER (ORDER BY sequence) AS projection_bytes FROM (SELECT id,sequence,octet_length(CAST(payload AS CHAR CHARACTER SET utf8mb4)) AS payload_bytes FROM workflow_history_events FORCE INDEX (workflow_history_events_workflow_run_id_sequence_unique) WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled')) AS sizes ORDER BY sequence LIMIT 10001) AS page JOIN workflow_history_events h ON h.id=page.id ORDER BY page.sequence"; const PHP_CONNECTION: &str = "mysql"; - const OPEN_WAIT_ORDER: &str = "WITH clocks AS (SELECT id,wait_id,opened_at,TIMESTAMPDIFF(MICROSECOND,'1970-01-01 00:00:00',opened_at) AS micros FROM workflow_run_waits WHERE workflow_run_id=? AND status='open') SELECT id FROM clocks ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN micros<0 THEN 2*FLOOR(micros/1000000)*1000000-micros ELSE micros END)/1000),wait_id LIMIT 100 OFFSET ?"; + const WAIT_POSITION_MOVES: &str = "WITH clocks AS (SELECT id,wait_id,position,opened_at,TIMESTAMPDIFF(MICROSECOND,'1970-01-01 00:00:00',opened_at) AS micros FROM workflow_run_waits WHERE workflow_run_id=?), ordinals AS (SELECT id,position,CAST(ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN micros<0 THEN 2*FLOOR(micros/1000000)*1000000-micros ELSE micros END)/1000),wait_id) AS SIGNED)-1 AS ordinal FROM clocks) SELECT id,ordinal FROM ordinals WHERE NOT (position <=> ordinal) ORDER BY ordinal LIMIT 100"; // The run-and-sequence index keeps MySQL from sorting whole rows: a // filesort of one large payload exceeds the default sort buffer and // fails the transaction ("Out of sort memory"). diff --git a/rust/src/runtime/transaction.rs b/rust/src/runtime/transaction.rs index 21bc4c851..a0bd59035 100644 --- a/rust/src/runtime/transaction.rs +++ b/rust/src/runtime/transaction.rs @@ -96,7 +96,7 @@ where return Err(refusal); } for run_id in std::mem::take(&mut self.summaries) { - Store::::order_open_waits(&mut self, &run_id).await?; + Store::::order_waits(&mut self, &run_id).await?; Store::::project_run_summary(&mut self, &run_id).await?; } self.inner.commit().await?; diff --git a/rust/src/runtime/wait_projection.rs b/rust/src/runtime/wait_projection.rs index 15077a2c0..a41e772d3 100644 --- a/rust/src/runtime/wait_projection.rs +++ b/rust/src/runtime/wait_projection.rs @@ -18,32 +18,41 @@ where for<'r> i64: Decode<'r, DB>, usize: ColumnIndex, { - pub(super) async fn order_open_waits(tx: &mut Transaction<'_, DB>, run: &str) -> Result<()> { - let mut position = 0_i64; + /// Store each wait's chronological ordinal over all of the run's retained + /// waits, as Workflow #941's projector does, writing only waits that moved. + /// Resolving a wait moves nothing, so most transitions write no positions. + pub(super) async fn order_waits(tx: &mut Transaction<'_, DB>, run: &str) -> Result<()> { + let mut moved = std::collections::HashSet::new(); loop { - // Match the published projector's open-wait order without - // rebuilding missing rows or hydrating the complete collection. - let rows = Self::query(DB::OPEN_WAIT_ORDER) + // The database ranks the run's waits; only rows whose stored + // position differs come back, at most a page at a time. + let rows = Self::query(DB::WAIT_POSITION_MOVES) .bind(run) - .bind(position) .fetch_all(&mut **tx) .await?; if rows.is_empty() { - break; + return Ok(()); } for row in rows { + let id = DB::string(&row, "id")?; + // A wait already written in this pass that still differs + // means the write did not apply; stop instead of looping. + if !moved.insert(id.clone()) { + return Err(super::refuse( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + "wait_position_not_applied", + )); + } Self::query( "UPDATE workflow_run_waits SET position=$1 WHERE id=$2 AND workflow_run_id=$3", ) - .bind(position) - .bind(DB::string(&row, "id")?) + .bind(DB::number(&row, "ordinal")?) + .bind(id) .bind(run) .execute(&mut **tx) .await?; - position += 1; } } - Ok(()) } pub(super) async fn project_update_wait_event( diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index cf9f88e60..9ea23ae06 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -244,6 +244,8 @@ mod run_observation_signals; mod run_observation_summaries; #[path = "execution/run_observation_timers.rs"] mod run_observation_timers; +#[path = "execution/wait_positions.rs"] +mod wait_positions; #[path = "execution/run_observation_updates.rs"] mod run_observation_updates; diff --git a/rust/tests/execution/wait_positions.rs b/rust/tests/execution/wait_positions.rs new file mode 100644 index 000000000..d19f41ce3 --- /dev/null +++ b/rust/tests/execution/wait_positions.rs @@ -0,0 +1,139 @@ +//! Stored wait positions are chronological over every retained wait of a run, +//! as Workflow #941's projector stores them, and a transition writes only the +//! waits whose position moved (Server #783). +use super::run_observation_signals::{claim, finish, register_signal_worker, signal}; +use super::*; + +const RUN: &str = "wait-positions"; + +/// Log position writes. SQLite and PostgreSQL log every write of the column, +/// including a rewrite of the same value; MySQL has no column-scoped update +/// trigger, so it logs only writes that change the value. +async fn log_position_writes(database: &TestDatabase) { + database.execute("CREATE TABLE wait_position_writes (wait_id VARCHAR(191) NOT NULL, old_position BIGINT NOT NULL, new_position BIGINT NOT NULL)").await; + match database { + TestDatabase::Sqlite(_) => { + database.execute("CREATE TRIGGER log_wait_position AFTER UPDATE OF position ON workflow_run_waits BEGIN INSERT INTO wait_position_writes VALUES (new.wait_id, old.position, new.position); END").await; + } + TestDatabase::Postgres { .. } => { + database.execute("CREATE FUNCTION log_wait_position() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN INSERT INTO wait_position_writes VALUES (NEW.wait_id, OLD.position, NEW.position); RETURN NEW; END $$").await; + database.execute("CREATE TRIGGER log_wait_position AFTER UPDATE OF position ON workflow_run_waits FOR EACH ROW EXECUTE FUNCTION log_wait_position()").await; + } + TestDatabase::MySql { .. } => { + database.execute("CREATE TRIGGER log_wait_position AFTER UPDATE ON workflow_run_waits FOR EACH ROW INSERT INTO wait_position_writes SELECT NEW.wait_id, OLD.position, NEW.position FROM DUAL WHERE NOT (OLD.position <=> NEW.position)").await; + } + } +} + +/// Logged position writes as `(wait_id, "old>new")`, sorted by wait id. +async fn position_writes(database: &TestDatabase) -> Vec<(String, String)> { + sorted(text_rows(database, + "SELECT wait_id, old_position||'>'||new_position FROM wait_position_writes", + "SELECT wait_id, old_position::text||'>'||new_position::text FROM wait_position_writes", + "SELECT CAST(wait_id AS CHAR), CAST(CONCAT(old_position,'>',new_position) AS CHAR) FROM wait_position_writes").await) +} + +/// The run's waits as `(wait_id, "position:status")`, sorted by wait id. +async fn waits(database: &TestDatabase) -> Vec<(String, String)> { + sorted(text_rows(database, + "SELECT wait_id, position||':'||status FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'", + "SELECT wait_id, position::text||':'||status FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'", + "SELECT CAST(wait_id AS CHAR), CAST(CONCAT(position,':',status) AS CHAR) FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'").await) +} + +/// The run's waits as `(wait_id, kind)`, sorted by wait id. +async fn kinds(database: &TestDatabase) -> Vec<(String, String)> { + sorted(text_rows(database, + "SELECT wait_id, kind FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'", + "SELECT wait_id, kind FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'", + "SELECT CAST(wait_id AS CHAR), CAST(kind AS CHAR) FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'").await) +} + +fn sorted(mut rows: Vec<(String, String)>) -> Vec<(String, String)> { + rows.sort(); + rows +} + +fn expected(rows: &[(&str, &str)]) -> Vec<(String, String)> { + sorted(rows.iter().map(|(id, value)| ((*id).to_owned(), (*value).to_owned())).collect()) +} + +async fn set_wait(database: &TestDatabase, wait_id: &str, opened_at: &str, position: i64) { + database.execute(sqlx::AssertSqlSafe(format!( + "UPDATE workflow_run_waits SET opened_at='{opened_at}',position={position} WHERE workflow_instance_id='{RUN}' AND wait_id='{wait_id}'" + ))).await; +} + +#[tokio::test] +async fn wait_positions_are_chronological_and_written_only_when_a_wait_moves() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register_signal_worker(&app, "default").await; + let run = namespaces::start_scoped(&app, "default", RUN).await; + let task = claim(&app, "default").await; + let opened = finish(&app, "default", &task, json!([ + {"type":"start_timer","delay_seconds":3600}, + {"type":"start_timer","delay_seconds":7200}, + {"type":"open_signal_wait","signal_name":"payload"}])).await; + assert_eq!(opened.0, StatusCode::OK, "{}", opened.1); + let opened = kinds(&database).await; + let timers: Vec<&str> = opened.iter().filter(|(_, kind)| kind == "timer").map(|(id, _)| id.as_str()).collect(); + let signals: Vec<&str> = opened.iter().filter(|(_, kind)| kind == "signal").map(|(id, _)| id.as_str()).collect(); + assert_eq!((timers.len(), signals.len(), opened.len()), (2, 1, 3), "{opened:?}"); + let (first_signal, early_timer, late_timer) = (signals[0].to_owned(), timers[0].to_owned(), timers[1].to_owned()); + // Retained clocks that run against wait-id order, stored in chronological + // order: the later timer id opened first and the earlier one last. + set_wait(&database, &late_timer, "2026-01-01 00:00:01", 0).await; + set_wait(&database, &first_signal, "2026-01-01 00:00:02", 1).await; + set_wait(&database, &early_timer, "2026-01-01 00:00:03", 2).await; + log_position_writes(&database).await; + + // Resolving a wait moves no wait: the open-wait ordinal would have moved + // the last timer from 2 to 1 and rewritten the first in place. + let received = signal(&app, "default", &run).await; + assert_eq!(received.0, StatusCode::ACCEPTED, "{}", received.1); + let task = claim(&app, "default").await; + assert_eq!( + waits(&database).await, + expected(&[(&late_timer, "0:open"), (&first_signal, "1:resolved"), (&early_timer, "2:open")]) + ); + assert_eq!(position_writes(&database).await, Vec::new()); + + // A new wait opens after every retained wait and is the only one that + // may be written. + let reopened = finish(&app, "default", &task, json!([{"type":"open_signal_wait","signal_name":"payload"}])).await; + assert_eq!(reopened.0, StatusCode::OK, "{}", reopened.1); + let current = kinds(&database).await; + let second_signal = current.iter().find(|(id, kind)| kind == "signal" && *id != first_signal).map(|(id, _)| id.clone()).unwrap(); + assert_eq!( + waits(&database).await, + expected(&[(&late_timer, "0:open"), (&first_signal, "1:resolved"), (&early_timer, "2:open"), (&second_signal, "3:open")]) + ); + let writes = position_writes(&database).await; + assert!(writes.iter().all(|(id, _)| *id == second_signal), "{writes:?}"); + + // Positions stored by another projector converge on the chronological + // ordinal, and only those rows are written. + set_wait(&database, &late_timer, "2026-01-01 00:00:01", 9).await; + set_wait(&database, &early_timer, "2026-01-01 00:00:03", 7).await; + database.execute("DELETE FROM wait_position_writes").await; + let received = signal(&app, "default", &run).await; + assert_eq!(received.0, StatusCode::ACCEPTED, "{}", received.1); + let task = claim(&app, "default").await; + let closed = finish(&app, "default", &task, json!([{"type":"complete_workflow","result":envelope(Payload::Null)}])).await; + assert_eq!(closed.0, StatusCode::OK, "{}", closed.1); + let positions: Vec<(String, String)> = waits(&database) + .await + .into_iter() + .map(|(id, value)| (id, value.split(':').next().unwrap().to_owned())) + .collect(); + assert_eq!( + positions, + expected(&[(&late_timer, "0"), (&first_signal, "1"), (&early_timer, "2"), (&second_signal, "3")]) + ); + assert_eq!( + position_writes(&database).await, + expected(&[(&early_timer, "7>2"), (&late_timer, "9>0")]) + ); +} From b7094d6a6316963388350134ac133eefa76fa5a1 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 26 Oct 2026 18:38:33 +0000 Subject: [PATCH 2/2] Rust #325: answer prepared local-activity routes as PHP refuses them without the role Native advertises prepared_local_activities false, no groups and no cancellation policies, like a published PHP server whose installed bridge lacks the prepared role, so SDKs record local activities with record_local_activity. The eight prepared local-activity worker routes (prepare, recover, checkpoint, checkpoint-group, and the attempt's control, heartbeat, outcome and acknowledge-cancellation) now answer what that PHP server answers instead of 501: PreparedLocalActivityController::refused's 409 prepared_local_activity[_groups]_not_supported, listing request_protocol below 1.20 and installed_runtime_prepared_local_activity (plus the groups runtime for the group checkpoint), before any task is read. The cancellation-scope checkpoint stays unserved: PHP gates it on the same bridge role as the scope opening native serves, so refusing it would be a state PHP cannot be in. Serving the family is planned on #325. Co-Authored-By: Claude Opus 5.5 --- docs/rust-server-port.md | 22 +++- rust/src/runtime/http.rs | 31 +++++ rust/src/runtime/mod.rs | 1 + rust/src/runtime/prepared_local.rs | 115 ++++++++++++++++++ rust/tests/execution/local_activities.rs | 41 +++++++ .../fixtures/unserved-php-routes.v2.11.1.txt | 8 -- 6 files changed, 206 insertions(+), 12 deletions(-) create mode 100644 rust/src/runtime/prepared_local.rs diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 9fc053771..971c9f3d4 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -82,11 +82,13 @@ Updated October 24, 2026. ([#737](https://github.com/durable-workflow/server/issues/737)). No expectation is edited to make native pass. -- **Parity surface.** Native serves 146 of the 175 API route methods in +- **Parity surface.** Native answers 154 of the 175 API route methods in published 2.11.1's route table, with message streams, update validation, activity cancellation acknowledgement, worker sessions, workflow-task cancellation scopes, caller-indexed Nexus operation reads, bridge adapter - webhooks, and embedded v2 and Waterline v1 workflow import. An execution + webhooks, and embedded v2 and Waterline v1 workflow import. Eight of those + are the prepared local-activity routes, which it answers with PHP's + refusal for a bridge without the prepared role. An execution test sends each one through the native router, and the ones still answering `501 rust_capability_not_implemented` must match [`unserved-php-routes.v2.11.1.txt`](../rust/tests/fixtures/unserved-php-routes.v2.11.1.txt). @@ -138,8 +140,20 @@ any other unknown path or method answers PHP's own 404 or 405 `repair-pass` fixture; a recovery native does not perform yet is reported as its candidate's failure, with 207. - **Worker protocol features.** - - Local activities (prepare, checkpoint, recover, heartbeat, control, - outcome and cancellation acknowledgement). + - Prepared local activities (prepare, checkpoint, group checkpoint, + recover, heartbeat, control, outcome and cancellation acknowledgement) + are answered as a PHP server whose bridge lacks the prepared role answers + them. Native advertises `prepared_local_activities: false`, no groups and + no policies, so SDKs record local activities with `record_local_activity`. + A worker that calls one anyway gets PHP's 409 + `prepared_local_activity[_groups]_not_supported`, listing + `request_protocol` below 1.20 and `installed_runtime_prepared_local_activity` + (plus `…_groups` for the group checkpoint), before any task is read. + No recorded fixture: published 2.11.1 has the role, so its answers can't + be recorded. `prepared_local_routes_are_refused_without_the_prepared_role` + pins the answers to `PreparedLocalActivityController::refused`. Serving + them is planned in + [#325 comment 6423707058](https://github.com/durable-workflow/server/issues/325#issuecomment-6423707058). - Cancellation scope checkpoints. (Open, prepare and deliver are served, with the `cancellation-scopes` fixture.) - (Worker sessions are served: create, heartbeat, close, list and detail, diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 667cc8622..0faa8a1a2 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -277,6 +277,19 @@ pub fn router(runtime: Runtime) -> Router { "/api/worker/workflow-tasks/{task_id}/cancellation-scopes/deliver", post(deliver_cancellation_scope), ) + // Native lacks the prepared local-activity role: refused as PHP + // refuses them without it (prepared_local.rs). + .route("/api/worker/workflow-tasks/{task_id}/local-activities/prepare", post(prepared_local_refused)) + .route("/api/worker/workflow-tasks/{task_id}/local-activities/recover", post(prepared_local_refused)) + .route("/api/worker/workflow-tasks/{task_id}/local-activities/checkpoint", post(prepared_local_refused)) + .route("/api/worker/workflow-tasks/{task_id}/local-activities/checkpoint-group", post(prepared_local_group_refused)) + .route("/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/control", post(prepared_local_refused)) + .route("/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/heartbeat", post(prepared_local_refused)) + .route("/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/outcome", post(prepared_local_refused)) + .route( + "/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/acknowledge-cancellation", + post(prepared_local_refused), + ) .route( "/api/worker/workflow-tasks/{task_id}/complete", post(complete_workflow), @@ -811,6 +824,8 @@ fn worker_route(rest: &str) -> Option { | ["update-validation-tasks", _, "approve" | "reject"] | ["workflow-tasks", _, "complete" | "fail" | "history" | "heartbeat" | "deliver-cancellation"] | ["workflow-tasks", _, "cancellation-scopes", "open" | "prepare" | "deliver"] + | ["workflow-tasks", _, "local-activities", "prepare" | "recover" | "checkpoint" | "checkpoint-group"] + | ["workflow-tasks", _, "local-activities", _, "control" | "heartbeat" | "outcome" | "acknowledge-cancellation"] | ["activity-tasks", _, "complete" | "fail" | "heartbeat" | "status" | "renew-lease" | "acknowledge-cancellation"] ) .then_some(Allowed::Any) @@ -2326,6 +2341,22 @@ async fn deliver_cancellation_scope( Ok((status, Json(body))) } +/// POST …/local-activities/*: refused before any task is read, as a PHP +/// server whose bridge lacks the prepared role refuses them. +async fn prepared_local_refused(headers: HeaderMap, body: axum::body::Bytes) -> Result<(StatusCode, Json)> { + prepared_local(&headers, &body, super::prepared_local::Capability::Prepared) +} + +async fn prepared_local_group_refused(headers: HeaderMap, body: axum::body::Bytes) -> Result<(StatusCode, Json)> { + prepared_local(&headers, &body, super::prepared_local::Capability::Groups) +} + +fn prepared_local(headers: &HeaderMap, body: &[u8], capability: super::prepared_local::Capability) -> Result<(StatusCode, Json)> { + let protocol = headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok()); + let (status, response) = super::prepared_local::refusal(&request_json(body)?, protocol, capability); + Ok((status, Json(response))) +} + async fn acknowledge_activity_cancellation( Extension(runtime): Extension, Path(task_id): Path, diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 9f05a7dcf..7d8dbdfe3 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -66,6 +66,7 @@ mod history_presentation; mod http; mod laravel_validation; mod local_activities; +mod prepared_local; mod memo_json; mod memo_limits; mod history_meter; diff --git a/rust/src/runtime/prepared_local.rs b/rust/src/runtime/prepared_local.rs new file mode 100644 index 000000000..5b4f04eec --- /dev/null +++ b/rust/src/runtime/prepared_local.rs @@ -0,0 +1,115 @@ +//! The prepared local-activity worker routes, as a published PHP server +//! answers them when its installed Workflow bridge lacks the prepared role +//! (`PreparedLocalActivityController::unavailable` and `refused`, and +//! `WorkerController::checkpointLocalActivityPrefix` and +//! `checkpointLocalActivityGroup`). +//! +//! Native advertises `prepared_local_activities: false`, no groups and no +//! cancellation policies, so SDKs record local activities with +//! `record_local_activity` instead. A worker that calls one of these routes +//! anyway is refused before any task is read, with PHP's 409 body: +//! - `request_protocol` when its protocol version is below 1.20; +//! - `installed_runtime_prepared_local_activity`, always; +//! - for `checkpoint-group`, also `installed_runtime_prepared_local_activity_groups`. +//! +//! A version header native does not support at all is refused earlier, by the +//! worker protocol gate, with 400 as in PHP. PHP's own server-protocol reason +//! never applies: native serves cooperative cancellation. +//! +//! PHP reports `(string) $request->input('lease_owner', '')` as `worker_id`. An +//! array or object there is an array-to-string conversion, which Laravel turns +//! into a 500. Native reports an empty `worker_id` instead. +use super::laravel_rules; +use axum::http::StatusCode; +use serde_json::{Value, json}; + +/// `PreparedLocalActivityPolicy::MINIMUM_PROTOCOL_VERSION`. +const MINIMUM_PROTOCOL_VERSION: &str = "1.20"; + +/// Which prepared capability the refused route needs. +#[derive(Clone, Copy)] +pub(crate) enum Capability { + /// Prepare, recover, control, heartbeat, outcome, acknowledge-cancellation + /// and the local prefix checkpoint. + Prepared, + /// The group checkpoint. + Groups, +} + +/// PHP's 409 for a route whose prepared role the installed runtime lacks. +pub(crate) fn refusal(body: &Value, protocol: Option<&str>, capability: Capability) -> (StatusCode, Value) { + let mut unavailable = Vec::new(); + if !super::worker_admission::at_least(protocol, MINIMUM_PROTOCOL_VERSION) { + unavailable.push("request_protocol"); + } + unavailable.push("installed_runtime_prepared_local_activity"); + let (reason, required) = match capability { + Capability::Prepared => ("prepared_local_activity_not_supported", "prepared_local_activities"), + Capability::Groups => { + unavailable.push("installed_runtime_prepared_local_activity_groups"); + ("prepared_local_activity_groups_not_supported", "prepared_local_activity_groups") + } + }; + let body = laravel_rules::normalize(body); + let worker_id = match &body["lease_owner"] { + Value::String(text) => text.clone(), + Value::Number(number) => php_number(number), + Value::Bool(true) => "1".into(), + _ => String::new(), + }; + let response = json!({ + "reason": reason, + "required_capability": required, + "minimum_protocol_version": MINIMUM_PROTOCOL_VERSION, + "worker_id": worker_id, + "unavailable": unavailable, + "remediation": "Use a prepared-local backend and a claim issued to a capable protocol 1.20 worker.", + }); + (StatusCode::CONFLICT, super::cancellation_scopes::enveloped(response)) +} + +/// PHP's `(string)` of a JSON number. +fn php_number(number: &serde_json::Number) -> String { + match number.as_i64() { + Some(whole) => whole.to_string(), + None => { + let float = number.as_f64().unwrap_or_default(); + if float.fract() == 0.0 && float.abs() < 1e15 { format!("{float:.0}") } else { float.to_string() } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_current_worker_is_refused_for_the_missing_runtime_role_only() { + let (status, body) = refusal(&json!({"lease_owner": " worker-1 "}), Some("1.20"), Capability::Prepared); + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!(body["reason"], "prepared_local_activity_not_supported"); + assert_eq!(body["required_capability"], "prepared_local_activities"); + assert_eq!(body["worker_id"], "worker-1"); + assert_eq!(body["unavailable"], json!(["installed_runtime_prepared_local_activity"])); + } + + #[test] + fn an_older_worker_is_also_refused_for_its_protocol() { + let (_, body) = refusal(&json!({}), Some("1.19"), Capability::Groups); + assert_eq!(body["reason"], "prepared_local_activity_groups_not_supported"); + assert_eq!(body["required_capability"], "prepared_local_activity_groups"); + assert_eq!(body["worker_id"], ""); + assert_eq!( + body["unavailable"], + json!(["request_protocol", "installed_runtime_prepared_local_activity", "installed_runtime_prepared_local_activity_groups"]) + ); + } + + #[test] + fn worker_id_is_php_string_cast_of_lease_owner() { + for (owner, expected) in [(json!(7), "7"), (json!(true), "1"), (json!(false), ""), (json!(null), ""), (json!(["x"]), "")] { + let (_, body) = refusal(&json!({"lease_owner": owner}), Some("1.20"), Capability::Prepared); + assert_eq!(body["worker_id"], expected, "{owner}"); + } + } +} diff --git a/rust/tests/execution/local_activities.rs b/rust/tests/execution/local_activities.rs index 41490976b..36aecc763 100644 --- a/rust/tests/execution/local_activities.rs +++ b/rust/tests/execution/local_activities.rs @@ -174,3 +174,44 @@ async fn a_recorded_local_activity_writes_its_attempts_history_and_failure() { assert_eq!(status, StatusCode::OK, "{described}"); assert_eq!((&described["status"], &described["wait_kind"]), (&json!("waiting"), &json!("signal")), "{described}"); } + +/// 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. +#[tokio::test] +async fn prepared_local_routes_are_refused_without_the_prepared_role() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let base = "/api/worker/workflow-tasks/no-such-task/local-activities"; + let prepared = ["prepare", "recover", "checkpoint", "attempt-1/control", "attempt-1/heartbeat", "attempt-1/outcome", + "attempt-1/acknowledge-cancellation"]; + for route in prepared { + let (status, body) = request_protocol(&app, "POST", &format!("{base}/{route}"), json!({"lease_owner": " worker-1 "}), "1.20").await; + assert_eq!(status, StatusCode::CONFLICT, "{route}: {body}"); + assert_eq!(body["reason"], "prepared_local_activity_not_supported", "{route}: {body}"); + assert_eq!(body["required_capability"], "prepared_local_activities", "{route}"); + assert_eq!(body["minimum_protocol_version"], "1.20", "{route}"); + assert_eq!(body["worker_id"], "worker-1", "{route}"); + assert_eq!(body["unavailable"], json!(["installed_runtime_prepared_local_activity"]), "{route}"); + assert_eq!(body["remediation"], "Use a prepared-local backend and a claim issued to a capable protocol 1.20 worker."); + assert_eq!(body["protocol_version"], "1.20", "{route}"); + assert_eq!(body["server_capabilities"]["prepared_local_activities"], false, "{route}"); + + // An older worker is refused for its protocol too. + let (status, body) = request_protocol(&app, "POST", &format!("{base}/{route}"), Value::Null, "1.19").await; + assert_eq!(status, StatusCode::CONFLICT, "{route}: {body}"); + assert_eq!(body["worker_id"], "", "{route}"); + assert_eq!(body["unavailable"], json!(["request_protocol", "installed_runtime_prepared_local_activity"]), "{route}"); + } + + let (status, body) = request_protocol(&app, "POST", &format!("{base}/checkpoint-group"), json!({"lease_owner": 7}), "1.20").await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!(body["reason"], "prepared_local_activity_groups_not_supported", "{body}"); + assert_eq!(body["required_capability"], "prepared_local_activity_groups"); + assert_eq!(body["worker_id"], "7"); + assert_eq!(body["unavailable"], json!(["installed_runtime_prepared_local_activity", "installed_runtime_prepared_local_activity_groups"])); + + // A version native does not support at all is the protocol gate's 400. + let (status, body) = request_protocol(&app, "POST", &format!("{base}/prepare"), json!({}), "2.0").await; + assert_eq!((status, &body["reason"]), (StatusCode::BAD_REQUEST, &json!("unsupported_protocol_version")), "{body}"); +} diff --git a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt index a67da9d4a..c7b368ced 100644 --- a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt +++ b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt @@ -20,13 +20,5 @@ POST /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{op GET /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{operationName}/service-calls/{serviceCallId} POST /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{operationName}/service-calls/{serviceCallId}/cancel POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/checkpoint -POST /api/worker/workflow-tasks/{taskId}/local-activities/checkpoint -POST /api/worker/workflow-tasks/{taskId}/local-activities/checkpoint-group -POST /api/worker/workflow-tasks/{taskId}/local-activities/prepare -POST /api/worker/workflow-tasks/{taskId}/local-activities/recover -POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/acknowledge-cancellation -POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/control -POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/heartbeat -POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/outcome POST /api/workflows/{workflowId}/runs/{runId}/streams/{streamName}/close POST /api/workflows/{workflowId}/runs/{runId}/streams/{streamName}/items