From 4dcbfabda434bfac4af010c2eca24716991b23d8 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 22:57:30 -0700 Subject: [PATCH 01/11] Vendored handler progress and told new callers of old records. The API protos take the NexusHandler progress fields and the notification service WIT. After a caller attaches, Core notifies the newest record, or arms an empty stream, so a caller that only waits for progress hears of every record. --- .../nexus/notification-service.wit | 95 +++++++ .../api_upstream/openapi/openapiv2.json | 6 +- .../api_upstream/openapi/openapiv3.yaml | 12 + .../temporal/api/common/v1/message.proto | 11 + .../v1/request_response.proto | 23 +- crates/protos/src/protos/mod.rs | 16 ++ crates/sdk-core/CHANGELOG.md | 4 + crates/streams/src/handle.rs | 5 + crates/streams/src/nexus/operation.rs | 232 +++++++++++++++++- crates/streams/src/notify.rs | 2 +- crates/streams/src/service.rs | 48 +++- 11 files changed, 435 insertions(+), 19 deletions(-) create mode 100644 crates/protos/protos/api_upstream/nexus/notification-service.wit diff --git a/crates/protos/protos/api_upstream/nexus/notification-service.wit b/crates/protos/protos/api_upstream/nexus/notification-service.wit new file mode 100644 index 0000000000..20ffb0f611 --- /dev/null +++ b/crates/protos/protos/api_upstream/nexus/notification-service.wit @@ -0,0 +1,95 @@ +// Nexus service that a Worker implements to take NexusHandler callbacks. The server calls +// OnComplete for the callback's `operation` and OnProgress for its `progress_operation`. Both +// operations complete synchronously. The records are backed by the messages in +// temporal/api/notificationservice/v1/request_response.proto, which hold the full comments. +package temporal:nexus@1.0.0; + +world system { + export notification-service; +} + +/// @nexus.service-name "temporal.api.notificationservice.v1.NotificationService" +/// @nexus.experimental +interface notification-service { + use nexus:temporal-types/model@1.0.0.{ + failure, + payload, + }; + + /// @nexus.doc "The result of the source operation." + /// @nexus.proto "temporal.api.notificationservice.v1.OnCompleteRequest.result" + variant on-complete-request-result { + success(payload), + failure(failure), + } + + /// @nexus.doc "The completion of the source operation." + /// @nexus.experimental + /// @nexus.proto "temporal.api.notificationservice.v1.OnCompleteRequest" + record on-complete-request { + /// @nexus.doc "The result of the source operation." + %result: on-complete-request-result, + /// @nexus.doc "User-supplied data which was added to the source invocation." + source-context: option, + /// @nexus.doc "The state of the last progress answer the server accepted, if any." + receiver-state: option>, + } + + /// @nexus.doc "The answer to a completion." + /// @nexus.experimental + /// @nexus.proto "temporal.api.notificationservice.v1.OnCompleteResponse" + record on-complete-response { + } + + /// @nexus.doc "The operation that a progress belongs to." + /// @nexus.proto "temporal.api.nexus.v1.NexusOperationProgress.operation" + variant nexus-operation-progress-operation { + scheduled-event-id(s64), + } + + /// @nexus.doc "The non-terminal progress of an operation." + /// @nexus.experimental + /// @nexus.proto "temporal.api.nexus.v1.NexusOperationProgress" + record nexus-operation-progress { + /// @nexus.doc "The operation this progress belongs to. Not set on a delivery." + operation: option, + /// @nexus.doc "Where the operation's output stands, such as a stream cursor." + position: option, + /// @nexus.doc "Orders the progress of one operation. The highest counter is kept." + counter: option, + /// @nexus.doc "Small details for the caller." + metadata: map, + } + + /// @nexus.doc "A progress of the source operation." + /// @nexus.experimental + /// @nexus.proto "temporal.api.notificationservice.v1.OnProgressRequest" + record on-progress-request { + /// @nexus.doc "The progress of the source operation." + progress: option, + /// @nexus.doc "User-supplied data which was added to the source invocation." + source-context: option, + /// @nexus.doc "The handler token of a started standalone source operation." + operation-token: option, + /// @nexus.doc "The state of the last progress answer the server accepted, if any." + receiver-state: option>, + } + + /// @nexus.doc "The answer to a progress." + /// @nexus.experimental + /// @nexus.proto "temporal.api.notificationservice.v1.OnProgressResponse" + record on-progress-response { + /// @nexus.doc "State that the server sends back with the next delivery." + receiver-state: option>, + /// @nexus.doc "Set when the handler did not reach the progress position." + behind: option, + } + + /// @nexus.doc "Receives the completion of the source operation. Called once." + /// @nexus.operation name="OnComplete" + on-complete: func(request: on-complete-request) -> on-complete-response; + + /// @nexus.doc "Receives the progress of the source operation, any number of times." + /// @nexus.operation name="OnProgress" + on-progress: func(request: on-progress-request) -> on-progress-response; +} diff --git a/crates/protos/protos/api_upstream/openapi/openapiv2.json b/crates/protos/protos/api_upstream/openapi/openapiv2.json index 186c16f26c..410461724d 100644 --- a/crates/protos/protos/api_upstream/openapi/openapiv2.json +++ b/crates/protos/protos/api_upstream/openapi/openapiv2.json @@ -10879,9 +10879,13 @@ "$ref": "#/definitions/v1Payload", "description": "There are restrictions on the maxium payload size a single callback can carry, as well as the\ntotal sum of all source context payloads attached to an execution. See dynamic configuration:\n\"callback.nexusHandler.sourceContext.maxSize\", \"callback.nexusHandler.sourceContext.aggregateMaxSize\".", "title": "Arbitrary user-supplied data from the source operation's callsite. (As applicable, not all operations\nsupport attaching context data.)" + }, + "progressOperation": { + "type": "string", + "description": "Target operation for progress of the source operation, e.g. \"OnProgress\". Empty means the\ncallback takes the completion only." } }, - "description": "The targeted Nexus service must be registered within the same namespace as the source operation\nthe callback is attached to. (While Nexus allows for cross-namespace operations, NexusHandler callbacks\nare strictly caller-side.)\n\nNexusHandler callbacks are only supported for certain types of operations, e.g. standalone Nexus operations.\nAttempting to attach a Worker callback for an unsupported operation will result in an INVALID_ARGUMENT\nerror from the server.", + "description": "When `progress_operation` is set, the server also invokes that operation of the same service for\nthe source operation's progress. That operation must have the following:\n- Input: temporal.api.notificationservice.v1.OnProgressRequest\n- Output: temporal.api.notificationservice.v1.OnProgressResponse\nIt may be invoked any number of times before, during or after the single completion. It must\ncomplete synchronously.\n\nThe targeted Nexus service must be registered within the same namespace as the source operation\nthe callback is attached to. (While Nexus allows for cross-namespace operations, NexusHandler callbacks\nare strictly caller-side.)\n\nNexusHandler callbacks are only supported for certain types of operations, e.g. standalone Nexus operations.\nAttempting to attach a Worker callback for an unsupported operation will result in an INVALID_ARGUMENT\nerror from the server.", "title": "NexusHandler callbacks are requests to invoke a specific shape of Nexus operation on a Temporal worker.\nThe specified Nexus operation must have the following:\n- Input: temporal.api.notificationservice.v1.OnCompleteRequest\n- Output: temporal.api.notificationservice.v1.OnCompleteResponse" }, "ComputeStatusProviderValidationStatus": { diff --git a/crates/protos/protos/api_upstream/openapi/openapiv3.yaml b/crates/protos/protos/api_upstream/openapi/openapiv3.yaml index 5b99461822..8f7ac7f425 100644 --- a/crates/protos/protos/api_upstream/openapi/openapiv3.yaml +++ b/crates/protos/protos/api_upstream/openapi/openapiv3.yaml @@ -10935,12 +10935,24 @@ components: There are restrictions on the maxium payload size a single callback can carry, as well as the total sum of all source context payloads attached to an execution. See dynamic configuration: "callback.nexusHandler.sourceContext.maxSize", "callback.nexusHandler.sourceContext.aggregateMaxSize". + progressOperation: + type: string + description: |- + Target operation for progress of the source operation, e.g. "OnProgress". Empty means the + callback takes the completion only. description: |- NexusHandler callbacks are requests to invoke a specific shape of Nexus operation on a Temporal worker. The specified Nexus operation must have the following: - Input: temporal.api.notificationservice.v1.OnCompleteRequest - Output: temporal.api.notificationservice.v1.OnCompleteResponse + When `progress_operation` is set, the server also invokes that operation of the same service for + the source operation's progress. That operation must have the following: + - Input: temporal.api.notificationservice.v1.OnProgressRequest + - Output: temporal.api.notificationservice.v1.OnProgressResponse + It may be invoked any number of times before, during or after the single completion. It must + complete synchronously. + The targeted Nexus service must be registered within the same namespace as the source operation the callback is attached to. (While Nexus allows for cross-namespace operations, NexusHandler callbacks are strictly caller-side.) diff --git a/crates/protos/protos/api_upstream/temporal/api/common/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/common/v1/message.proto index 908c5c80b8..95256a4ddf 100644 --- a/crates/protos/protos/api_upstream/temporal/api/common/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/common/v1/message.proto @@ -207,6 +207,13 @@ message Callback { // - Input: temporal.api.notificationservice.v1.OnCompleteRequest // - Output: temporal.api.notificationservice.v1.OnCompleteResponse // + // When `progress_operation` is set, the server also invokes that operation of the same service for + // the source operation's progress. That operation must have the following: + // - Input: temporal.api.notificationservice.v1.OnProgressRequest + // - Output: temporal.api.notificationservice.v1.OnProgressResponse + // It may be invoked any number of times before, during or after the single completion. It must + // complete synchronously. + // // The targeted Nexus service must be registered within the same namespace as the source operation // the callback is attached to. (While Nexus allows for cross-namespace operations, NexusHandler callbacks // are strictly caller-side.) @@ -232,6 +239,10 @@ message Callback { // total sum of all source context payloads attached to an execution. See dynamic configuration: // "callback.nexusHandler.sourceContext.maxSize", "callback.nexusHandler.sourceContext.aggregateMaxSize". temporal.api.common.v1.Payload source_context = 4; + + // Target operation for progress of the source operation, e.g. "OnProgress". Empty means the + // callback takes the completion only. + string progress_operation = 5; } reserved 1; // For a generic callback mechanism to be added later. diff --git a/crates/protos/protos/api_upstream/temporal/api/notificationservice/v1/request_response.proto b/crates/protos/protos/api_upstream/temporal/api/notificationservice/v1/request_response.proto index cd79e13a65..98bbf2cf1d 100644 --- a/crates/protos/protos/api_upstream/temporal/api/notificationservice/v1/request_response.proto +++ b/crates/protos/protos/api_upstream/temporal/api/notificationservice/v1/request_response.proto @@ -32,6 +32,10 @@ message OnCompleteRequest { // User-supplied data which was added to the source invocation. (As applicable.) temporal.api.common.v1.Payload source_context = 3; + + // The `receiver_state` of the last OnProgressResponse that the server accepted for this + // callback. Empty when no progress was accepted. + bytes receiver_state = 4; } // OnCompleteResponse is the return type of the OnComplete operation. @@ -48,7 +52,24 @@ message OnProgressRequest { // User-supplied data which was added to the source invocation. (As applicable.) temporal.api.common.v1.Payload source_context = 2; + + // The handler token of the source operation, when the source is a standalone Nexus operation + // that has started. Empty otherwise. + string operation_token = 3; + + // The `receiver_state` of the last OnProgressResponse that the server accepted for this + // callback. Empty when no progress was accepted. + bytes receiver_state = 4; } // OnProgressResponse is the return type of the OnProgress operation. -message OnProgressResponse {} +message OnProgressResponse { + // Opaque state for the handler, such as a read cursor. The server stores it on the callback + // and sends it in the next OnProgressRequest and in the OnCompleteRequest. The server refuses + // further progress for the callback when the state is larger than its configured limit. + bytes receiver_state = 1; + + // Set when the handler did not reach `progress.position`. The server then sends the same + // progress again at once, unless newer progress replaced it. + bool behind = 2; +} diff --git a/crates/protos/src/protos/mod.rs b/crates/protos/src/protos/mod.rs index b52fc44518..4f2fee01a3 100644 --- a/crates/protos/src/protos/mod.rs +++ b/crates/protos/src/protos/mod.rs @@ -3061,6 +3061,22 @@ mod sdk_helpers { }; use anyhow::anyhow; + mod nexus_handler_callback { + use crate::protos::temporal::api::common::v1::callback::NexusHandler; + use prost::Message; + + // The server reads the progress operation by its field number, so a renumbered field + // would send progress to an operation the Worker never named. + #[test] + fn the_progress_operation_keeps_its_field_number() { + let handler = NexusHandler { + progress_operation: "p".to_string(), + ..Default::default() + }; + assert_eq!(handler.encode_to_vec(), vec![0x2a, 0x01, b'p']); + } + } + mod stream_record_envelope { use crate::protos::temporal::{ api::common::v1::Payload, diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 7d21bc6283..ea2a985df0 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -113,6 +113,10 @@ relevant information. the Nexus calls, so lang's handler raises it as it is. A mistake in the stream, an empty callback URL or request id, and a token that names no stream are `BAD_REQUEST`. An attach for the latest run describes the Workflow fresh, so it never lands on a chain that already ended. +* Experimental: with `notify_on_append`, `AttachCaller` tells the stream's notifier about the + newest record after the attach, or arms an empty stream, so a caller that only waits for + progress hears of the records already there and of the first append. The API protos carry the + `NexusHandler` progress fields and `nexus/notification-service.wit`. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 101404115a..b54fa804ab 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -429,6 +429,11 @@ impl Streams { } } + /// The store the handles serve. + pub(crate) fn store(&self) -> &dyn StreamStore { + self.store.as_ref() + } + /// The cursor of a stream's newest record, empty when it holds none. pub async fn latest(&self, request: LatestRequest) -> StreamResult { let stream = address(request.stream.as_ref())?; diff --git a/crates/streams/src/nexus/operation.rs b/crates/streams/src/nexus/operation.rs index 6edf7161c1..b92f94e91f 100644 --- a/crates/streams/src/nexus/operation.rs +++ b/crates/streams/src/nexus/operation.rs @@ -21,9 +21,10 @@ use std::{io, sync::Arc, time::SystemTime}; pub(crate) async fn attach_caller( streams: &Streams, client: Option<&Arc>, + announce: Option<&Arc>, request: AttachCallerRequest, ) -> AttachCallerResponse { - match attach(streams, client, request).await { + match attach(streams, client, announce, request).await { Ok(operation_token) => AttachCallerResponse { operation_token, failure: None, @@ -51,6 +52,7 @@ pub(crate) async fn detach_caller( async fn attach( streams: &Streams, client: Option<&Arc>, + announce: Option<&Arc>, request: AttachCallerRequest, ) -> StreamResult { if request.callback_url.is_empty() { @@ -78,7 +80,7 @@ async fn attach( // completion that beats the start's response still names the operation. client .attach_caller(CallerAttachment { - chain, + chain: chain.clone(), topic: stream.topic.clone(), request_id: request.request_id, callback_url: request.callback_url, @@ -88,9 +90,46 @@ async fn attach( }) .await .map_err(caller_error)?; + if let Some(notifier) = announce { + announce_newest(streams, notifier, chain, &stream.topic).await; + } Ok(operation_token) } +/// Notifies the newest record on `topic`, so a caller that just attached hears of the records +/// already there. The server keeps a notifier only once a caller attached, so the notifications +/// sent before were dropped. It goes through the notifier the appends use, so a failed send is +/// tried again and a flush waits for it. A notification the retries can't send is lost as an +/// append's is, and the caller still hears of every record from the next append or the close. +/// +/// On an empty stream it arms the stream instead, in the same store step that finds it empty, +/// since a caller that waits only for progress must hear of the first append. +async fn announce_newest( + streams: &Streams, + notifier: &crate::Notifier, + chain: ChainId, + topic: &str, +) { + let position = match streams.store().newest_or_arm(&chain, topic).await { + Ok(crate::NewestOrArm::Newest(position)) => position, + Ok(crate::NewestOrArm::Armed(_) | crate::NewestOrArm::Unarmed) => return, + Err(error) => { + tracing::warn!( + workflow_id = %chain.workflow_id, + topic, + %error, + "Could not find the newest record to tell a new caller about" + ); + return; + } + }; + if let Some((cursor, counter)) = + crate::notify::notification_at(streams.store(), &chain, topic, &position) + { + notifier.notify(&chain, topic, cursor, counter); + } +} + async fn detach( client: Option<&Arc>, request: DetachCallerRequest, @@ -331,12 +370,23 @@ mod tests { struct Notifier { attached: Mutex>, detached: Mutex>, + notified: Mutex>, + notify_failures: std::sync::atomic::AtomicUsize, answer: Mutex>, } #[async_trait::async_trait] impl NotifierClient for Notifier { - async fn notify(&self, _: StreamNotification) -> Result<(), NotifyError> { + async fn notify(&self, notification: StreamNotification) -> Result<(), NotifyError> { + use std::sync::atomic::Ordering; + if self + .notify_failures + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1)) + .is_ok() + { + return Err(NotifyError::Failed("the server is down".to_string())); + } + self.notified.lock().unwrap().push(notification); Ok(()) } @@ -465,6 +515,182 @@ mod tests { .map_or(Ok(()), |failure| Err(raised(failure))) } + fn append(values: &[&str]) -> Vec { + use crate::proto::{AppendRecord, AppendRequest, NamedProducer, append_request}; + use temporalio_common::protos::temporal::{ + api::common::v1::Payload, sdk::streams::v1::StreamRecordKind, + }; + AppendRequest { + stream: Some(stream("")), + producer: Some(append_request::Producer::Named(NamedProducer { + producer_id: "p".to_string(), + attempt: 1, + })), + sequence: 1, + digest: vec![1; 32], + records: values + .iter() + .map(|value| AppendRecord { + kind: StreamRecordKind::Data as i32, + body: Some(Payload { + data: value.as_bytes().to_vec(), + ..Default::default() + }), + content_hash: vec![value.as_bytes()[0]; 32], + }) + .collect(), + } + .encode_to_vec() + } + + /// `setup()` with the announcement on, sent through Core's folding notifier. + fn announcing() -> Setup { + let mut setup = setup(); + let folding = Arc::new(crate::Notifier::new(setup.notifier.clone(), 16)); + setup.service = setup.service.with_attach_progress(Some(folding)); + setup + } + + /// Waits for the background sends of the folding notifier. + async fn notified(notifier: &Notifier, count: usize) -> Vec { + for _ in 0..50 { + if notifier.notified.lock().unwrap().len() >= count { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + notifier.notified.lock().unwrap().clone() + } + + #[tokio::test] + async fn an_attach_announces_the_records_already_there_when_progress_is_on() { + let setup = announcing(); + setup + .service + .call("Append", &append(&["a", "b"])) + .await + .unwrap(); + attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + let notified = notified(&setup.notifier, 1).await; + let [notification] = notified.as_slice() else { + panic!("one notification after the attach, not {notified:?}"); + }; + assert_eq!( + notification.chain, + setup.notifier.attached.lock().unwrap()[0].chain + ); + assert_eq!(notification.topic, "tokens"); + // The memory store's second record is at offset 1, so its counter is 2. + assert_eq!(notification.counter, 2); + assert!( + notification.position.ends_with(":1"), + "{}", + notification.position + ); + assert!(notification.close.is_none()); + } + + #[tokio::test] + async fn a_failed_announcement_is_sent_again_as_an_append_notification_is() { + // A quiet stream gets no other notification, so a lost announcement would leave a + // caller that only waits for progress with nothing until the close. + let setup = announcing(); + setup.service.call("Append", &append(&["a"])).await.unwrap(); + setup + .notifier + .notify_failures + .store(1, std::sync::atomic::Ordering::SeqCst); + attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + let notified = notified(&setup.notifier, 1).await; + assert_eq!(notified.len(), 1, "the retry went out: {notified:?}"); + assert_eq!(notified[0].counter, 1); + } + + /// Appends one record to the stream through the store, answering the arm it reported. + async fn arm_reported(service: &StreamService) -> Option { + service + .store() + .append(crate::proto::StoreAppendRequest { + chain: Some(ChainId { + namespace: "default".to_string(), + workflow_id: "owner-1".to_string(), + first_run_id: "chain-1".to_string(), + }), + topic: "tokens".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence: 1, + digest: vec![1], + records: vec![b"record".to_vec()], + }) + .await + .unwrap() + .arm + } + + #[tokio::test] + async fn an_attach_to_an_empty_stream_arms_it_for_the_first_append() { + // A follower waits only for progress, so the first append must notify it. + let setup = announcing(); + attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + assert!( + arm_reported(&setup.service) + .await + .is_some_and(|arm| arm > 0) + ); + } + + #[tokio::test] + async fn an_attach_to_a_stream_with_records_announces_without_arming() { + let setup = announcing(); + setup.service.call("Append", &append(&["a"])).await.unwrap(); + attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + assert_eq!(notified(&setup.notifier, 1).await.len(), 1); + let landed = setup + .service + .store() + .append(crate::proto::StoreAppendRequest { + chain: Some(ChainId { + namespace: "default".to_string(), + workflow_id: "owner-1".to_string(), + first_run_id: "chain-1".to_string(), + }), + topic: "tokens".to_string(), + producer_id: "other".to_string(), + attempt: 1, + sequence: 1, + digest: vec![1], + records: vec![b"record".to_vec()], + }) + .await + .unwrap(); + assert_eq!(landed.arm, Some(0)); + } + + #[tokio::test] + async fn an_attach_announces_nothing_on_an_empty_stream_or_with_progress_off() { + let setup = announcing(); + attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + assert!(notified(&setup.notifier, 1).await.is_empty()); + + let setup = self::setup(); + setup.service.call("Append", &append(&["a"])).await.unwrap(); + attach(&setup.service, attach_request("temporal://system")) + .await + .unwrap(); + assert!(notified(&setup.notifier, 1).await.is_empty()); + } + #[tokio::test] async fn a_start_attaches_the_callers_callback_and_hands_back_the_ref() { let setup = setup(); diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index adc517122c..2b7da1f3e0 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -828,7 +828,7 @@ fn cursor_of(store: &dyn StreamStore, chain: &ChainId, topic: &str, position: &s } /// The cursor and counter that tell a reader about the record at `position`. -fn notification_at( +pub(crate) fn notification_at( store: &dyn StreamStore, chain: &ChainId, topic: &str, diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 0299794940..9aafef5e9e 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -33,6 +33,7 @@ pub struct StreamService { subscriptions: Subscriptions, nexus: NexusCalls, callers: Option>, + attach_progress: Option>, } impl std::fmt::Debug for StreamService { @@ -78,25 +79,27 @@ impl StreamService { "stream notifications need a client to send them with", )); } - let store: Arc = match notifier.clone() { - Some(client) => { - let max_notifiers = match config.max_notifiers { - 0 => DEFAULT_MAX_NOTIFIERS, - max => max as usize, - }; - Arc::new( - NotifyingStore::new(store, Arc::new(Notifier::new(client, max_notifiers))) - .with_progress(config.notify_on_append) - .with_every_append(config.notify_every_append), - ) - } + let folding = notifier.as_ref().map(|client| { + let max_notifiers = match config.max_notifiers { + 0 => DEFAULT_MAX_NOTIFIERS, + max => max as usize, + }; + Arc::new(Notifier::new(client.clone(), max_notifiers)) + }); + let store: Arc = match &folding { + Some(folding) => Arc::new( + NotifyingStore::new(store, folding.clone()) + .with_progress(config.notify_on_append) + .with_every_append(config.notify_every_append), + ), None => store, }; let service = Self::new(store, owner, StreamsOptions::for_retention(retention)); Ok(match notifier { Some(client) => service .with_subscriptions(client.clone()) - .with_callers(client), + .with_callers(client) + .with_attach_progress(folding.filter(|_| config.notify_on_append)), None => service, }) } @@ -115,6 +118,7 @@ impl StreamService { subscriptions: Subscriptions::default(), nexus, callers: None, + attach_progress: None, } } @@ -130,6 +134,23 @@ impl StreamService { self } + /// After a caller's attach, notifies the stream's newest record, so the new caller hears of + /// the records already there. A caller that only waits for progress would otherwise hear + /// nothing until the next append. It goes through `notifier`, the one that sends the appends' + /// notifications, so it is retried and flushed as they are. On an empty stream the attach + /// arms the stream instead, so the first append notifies. `None` turns both off, as the + /// config's progress setting does. + pub fn with_attach_progress(mut self, notifier: Option>) -> Self { + self.attach_progress = notifier; + self + } + + /// The store, for tests that append past the stream layer. + #[cfg(test)] + pub(crate) fn store(&self) -> Arc { + self.store.clone() + } + /// Serves one call: `rpc` is a `StreamService` method name and `request` its serialized /// request. Answers with the serialized response, or the failure the call ended with. pub async fn call(&self, rpc: &str, request: &[u8]) -> Result, StreamFailure> { @@ -215,6 +236,7 @@ impl StreamService { "AttachCaller" => Ok(crate::nexus::attach_caller( &self.streams, self.callers.as_ref(), + self.attach_progress.as_ref(), decode::(rpc, request)?, ) .await From 6bb46d659a96412b16aa33cc6f177baa74c2d02f Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 09:02:04 -0700 Subject: [PATCH 02/11] Declared `read_range` and the read's floor holder in the contract. A Workflow that reads through a Local Activity keeps the range of each answer in place of its records. The read must hold that range, and replay must read it again. --- .../nexus/temporal_streams.nexusrpc.yaml | 104 ++++++++++++++++++ crates/streams/src/nexus/mod.rs | 4 + crates/streams/src/nexus/wire.rs | 87 +++++++++++++++ crates/streams/tests/nexus_contract.rs | 55 ++++++++- 4 files changed, 249 insertions(+), 1 deletion(-) diff --git a/crates/streams/nexus/temporal_streams.nexusrpc.yaml b/crates/streams/nexus/temporal_streams.nexusrpc.yaml index 141771bcf6..99bc5e5a9a 100644 --- a/crates/streams/nexus/temporal_streams.nexusrpc.yaml +++ b/crates/streams/nexus/temporal_streams.nexusrpc.yaml @@ -51,6 +51,20 @@ services: state-field: state input: { $ref: "#/$defs/ReadInput" } output: { $ref: "#/$defs/ReadOutput" } + read_range: + fqn: read_range + description: >- + Answer again the records of an earlier read that named a + floor_holder, for a Workflow that replays that read. The call reads + the range its replay names with the read's input state, so the + superseded and stale marks come out the same, and checks the records + against the read's digest. It waits for nothing, asks nothing of the + owner and has no byte budget, since it must answer exactly what the + read answered. Records with other content are refused as + StreamReplayMismatchError, and records the store no longer holds as + StreamReplayRangeLostError. + input: { $ref: "#/$defs/ReadRangeInput" } + output: { $ref: "#/$defs/ReadOutput" } $defs: StreamRef: @@ -196,9 +210,60 @@ $defs: the call waits its whole wait_ms and collects nothing, the endpoint arms the stream, so the next append notifies the stream's callers. A reader with no callback leaves it out. + floor_holder: + type: string + description: >- + Names a reader that keeps the range of each answer in place of its + records and reads the range again through read_range on replay. With + it the endpoint holds the records after the read's start against + retention for this holder, and the answer carries replay. Each read + and read_range of the holder renews the hold, which lapses the + store's floor lease after the last one. At most 512 bytes. Refused + beside latest_only, which answers no records. required: [stream] additionalProperties: false + ReadRangeInput: + type: object + description: >- + One replay of an earlier read that named a floor_holder. Every member + but stream and floor_holder comes from that read's input and answer. + properties: + stream: { $ref: "#/$defs/StreamRef" } + chain_first_run_id: + type: string + description: "The replay's chain_first_run_id." + after_token: + type: string + x-nexus-cursor: StreamCursor + description: "The replay's after_token." + to_token: + type: string + x-nexus-cursor: StreamCursor + description: "The read answer's next_token." + stored_records: + type: integer + minimum: 0 + maximum: 1000 + description: "The replay's stored_records." + state: + type: string + contentEncoding: base64 + description: >- + The state the read sent, so the replay's superseded and stale marks + are the read's. Absent when the read sent none. + records_digest: + type: string + contentEncoding: base64 + description: "The replay's records_digest." + floor_holder: + type: string + description: >- + As on read, so a replay renews the hold, also after a reset that + replays a run whose reads stopped long ago. + required: [stream, chain_first_run_id, after_token, to_token, stored_records, records_digest] + additionalProperties: false + RecordWire: type: object description: "One record on the wire: its cursor and the record." @@ -293,9 +358,48 @@ $defs: The call asked to arm, collected nothing and armed the stream, so the caller waits for progress. Never true with records or with done. An empty answer without it falls back to waiting for progress too. + replay: + $ref: "#/$defs/ReplayRange" + description: >- + Set when the call named a floor_holder. What a replay passes to + read_range to get these records again. required: [records, next_token, done] additionalProperties: true + ReplayRange: + type: object + description: >- + The range one read answered, after its byte budget cut it. A Workflow + keeps it with the answer's next_token and state in place of the records. + properties: + chain_first_run_id: + type: string + description: "The first run of the run chain the read read." + after_token: + type: string + x-nexus-cursor: StreamCursor + description: >- + The cursor the answer reads strictly after, with the read's start + resolved. Empty when the answer starts at the oldest record. An + answer whose only record is an error the store's reader raised has + an empty range at that record's cursor. + stored_records: + type: integer + minimum: 0 + description: >- + How many stored records the answer holds, the ones that crossed as + an error included and supersessions not counted. + records_digest: + type: string + contentEncoding: base64 + description: >- + SHA-256 over the answer's records. Per record it covers the cursor, + the stale mark, and the kind, producer, attempt, sequence and + content hash of a stored record or the producer and both attempts of + a supersession. It leaves out the bodies, so a codec change keeps it. + required: [chain_first_run_id, after_token, stored_records, records_digest] + additionalProperties: true + StreamOperationToken: type: object description: >- diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index c28c130f51..3c8f27b90a 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -312,6 +312,7 @@ impl NexusCalls { done: false, state: None, armed: false, + replay: None, }), Err(error) => failed(failure::of_stream(&error)), }; @@ -645,6 +646,7 @@ impl Budget { done: false, state: Some(Base64::default()), armed: false, + replay: None, }); let state = crossing_size(read.state.encoded_len().min(MAX_STATE_BYTES)); Self { @@ -742,6 +744,7 @@ fn finish_read( done, state: Some(Base64(state.encode_to_vec())), armed, + replay: None, }) } @@ -816,6 +819,7 @@ fn read_failure(error: StreamError, state: Vec) -> NexusStep { done: false, state: (!state.is_empty()).then_some(Base64(state)), armed: false, + replay: None, }), _ => failed(failure::of_stream(&error)), } diff --git a/crates/streams/src/nexus/wire.rs b/crates/streams/src/nexus/wire.rs index dc216e2ad6..41895ae2a6 100644 --- a/crates/streams/src/nexus/wire.rs +++ b/crates/streams/src/nexus/wire.rs @@ -98,6 +98,32 @@ pub(crate) struct ReadInput { pub(crate) state: Option, #[serde(default)] pub(crate) arm: Option, + #[serde(default)] + pub(crate) floor_holder: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(deny_unknown_fields)] +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "the contract declares the model before Core serves it" + ) +)] +pub(crate) struct ReadRangeInput { + pub(crate) stream: StreamRef, + pub(crate) chain_first_run_id: String, + pub(crate) after_token: String, + pub(crate) to_token: String, + #[serde(deserialize_with = "spec_integer")] + pub(crate) stored_records: i64, + /// Left as base64, so an oversized state is refused before it is decoded. + #[serde(default)] + pub(crate) state: Option, + pub(crate) records_digest: Base64, + #[serde(default)] + pub(crate) floor_holder: Option, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] @@ -109,6 +135,16 @@ pub(crate) struct ReadOutput { pub(crate) state: Option, #[serde(skip_serializing_if = "std::ops::Not::not")] pub(crate) armed: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) replay: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ReplayRange { + pub(crate) chain_first_run_id: String, + pub(crate) after_token: String, + pub(crate) stored_records: i64, + pub(crate) records_digest: Base64, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] @@ -212,6 +248,57 @@ mod tests { ); } + #[test] + fn a_read_range_input_takes_its_members_and_refuses_others() { + let input: ReadRangeInput = serde_json::from_str( + r#"{"stream":{"kind":"workflow","workflow_id":"wf","topic":"t"}, + "chain_first_run_id":"run-1","after_token":"","to_token":"c","stored_records":2.0, + "state":null,"records_digest":"AQI=","floor_holder":"ns/run/r"}"#, + ) + .unwrap(); + assert_eq!(input.stored_records, 2); + assert_eq!(input.records_digest.0, [1, 2]); + assert_eq!(input.state, None); + assert_eq!(input.floor_holder.as_deref(), Some("ns/run/r")); + let unknown = serde_json::from_str::( + r#"{"stream":{"kind":"workflow","workflow_id":"wf","topic":"t"}, + "chain_first_run_id":"run-1","after_token":"","to_token":"c","stored_records":0, + "records_digest":"","wait_ms":5}"#, + ) + .unwrap_err(); + assert!(unknown.to_string().contains("unknown field"), "{unknown}"); + let missing = serde_json::from_str::( + r#"{"stream":{"kind":"workflow","workflow_id":"wf","topic":"t"}, + "chain_first_run_id":"run-1","after_token":"","stored_records":0, + "records_digest":""}"#, + ) + .unwrap_err(); + assert!(missing.to_string().contains("to_token"), "{missing}"); + } + + #[test] + fn a_replay_range_crosses_with_its_digest_as_base64() { + let output = ReadOutput { + records: Vec::new(), + next_token: "c".to_string(), + done: false, + state: None, + armed: false, + replay: Some(ReplayRange { + chain_first_run_id: "run-1".to_string(), + after_token: String::new(), + stored_records: 0, + records_digest: Base64(vec![1, 2]), + }), + }; + assert_eq!( + serde_json::to_value(&output).unwrap(), + serde_json::json!({"records": [], "next_token": "c", "done": false, + "replay": {"chain_first_run_id": "run-1", "after_token": "", "stored_records": 0, + "records_digest": "AQI="}}) + ); + } + #[test] fn a_record_crosses_at_four_thirds_of_its_bytes() { assert_eq!(crossing_size(0), 0); diff --git a/crates/streams/tests/nexus_contract.rs b/crates/streams/tests/nexus_contract.rs index 9f0e17a32e..3237f4a023 100644 --- a/crates/streams/tests/nexus_contract.rs +++ b/crates/streams/tests/nexus_contract.rs @@ -60,6 +60,7 @@ fn full_input(contract: &Value, name: &str, workflow_id: &str) -> Value { ("ReadInput", "state") => Value::Null, // So the answer carries `armed`, which the output check holds to the contract. ("ReadInput", "arm") => json!(true), + ("ReadInput", "floor_holder") => json!("ns/run-1/reader"), (model, member) => panic!("the contract added {model}.{member}; teach this test"), }; out.insert(member, value); @@ -101,7 +102,15 @@ fn the_contract_names_the_operations_core_serves() { ); assert_eq!(operations["read"]["input"]["$ref"], "#/$defs/ReadInput"); assert_eq!(operations["read"]["output"]["$ref"], "#/$defs/ReadOutput"); - for input in ["StreamRef", "AppendInput", "ReadInput"] { + assert_eq!( + operations["read_range"]["input"]["$ref"], + "#/$defs/ReadRangeInput" + ); + assert_eq!( + operations["read_range"]["output"]["$ref"], + "#/$defs/ReadOutput" + ); + for input in ["StreamRef", "AppendInput", "ReadInput", "ReadRangeInput"] { assert_eq!( model(&contract, input)["additionalProperties"], false, @@ -130,6 +139,50 @@ fn a_read_carries_its_state_through_the_generated_long_poll() { } } +#[test] +fn a_replay_range_names_what_read_range_takes() { + // A replay passes the answer's range back as it got it, so every member of `ReplayRange` + // is a required member of `ReadRangeInput` of the same type, and the answer names it. + let contract = contract(); + assert_eq!( + model(&contract, "ReadOutput")["properties"]["replay"]["$ref"], + "#/$defs/ReplayRange" + ); + let range = model(&contract, "ReplayRange"); + let input = model(&contract, "ReadRangeInput"); + let mut members = required(range); + members.sort(); + assert_eq!( + members, + [ + "after_token", + "chain_first_run_id", + "records_digest", + "stored_records" + ] + ); + for member in members { + assert!(required(input).contains(&member), "{member}"); + assert_eq!( + range["properties"][&member]["type"], input["properties"][&member]["type"], + "{member}" + ); + } + for member in ["stream", "to_token"] { + assert!(required(input).contains(&member.to_string()), "{member}"); + } + for member in ["state", "floor_holder"] { + assert!(!required(input).contains(&member.to_string()), "{member}"); + } + assert_eq!( + input["properties"]["records_digest"]["contentEncoding"], + "base64" + ); + let holder = &model(&contract, "ReadInput")["properties"]["floor_holder"]; + assert_eq!(holder["type"], "string"); + assert!(!required(model(&contract, "ReadInput")).contains(&"floor_holder".to_string())); +} + #[test] fn the_operation_token_model_is_closed_and_versioned() { // The generated token helpers refuse a token model that isn't closed or that lacks a From 74def5f7f6c8655e47082873f95d16c52f9228e5 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 09:05:11 -0700 Subject: [PATCH 03/11] Held the floor on a Nexus read and answered the range that crossed. A reader that keeps positions needs the records held until replay, and a range that replay can read again. The range covers the records left after the byte cut, since replay must give the records the Workflow got. --- crates/streams/src/nexus/mod.rs | 95 ++++++--- crates/streams/tests/nexus_contract.rs | 5 + crates/streams/tests/nexus_service.rs | 263 ++++++++++++++++++++++++- 3 files changed, 333 insertions(+), 30 deletions(-) diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index 3c8f27b90a..2170e0b17d 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -22,7 +22,7 @@ use crate::{ StreamAddress, StreamOwnerKind, StreamRecord, StreamRecordKind, append_request::Producer, codec_result, nexus_step, read_record, }, - reader::ReadState, + reader::{ReadState, records_digest}, }; use prost::Message; use sha2::{Digest, Sha256}; @@ -37,13 +37,14 @@ use std::{ use temporalio_common::protos::temporal::api::common::v1::Payload; use tokio::time::Instant; use wire::{ - AppendInput, AppendOutput, Base64, ReadInput, ReadOutput, RecordWire, StreamRef, Supersession, - crossing_size, json_size, + AppendInput, AppendOutput, Base64, ReadInput, ReadOutput, RecordWire, ReplayRange, StreamRef, + Supersession, crossing_size, json_size, }; /// The most bytes one read answer takes as JSON past its first record, with its tokens and state. -/// A read answer is a sync Nexus result recorded in the caller's History, so it stays well below -/// the server's blob size limit (2 MiB by default). +/// A Workflow caller records a read answer in its History, unless its read names a floor holder +/// and it keeps only the range. So an answer stays well below the server's blob size limit +/// (2 MiB by default). pub const DEFAULT_MAX_ANSWER_BYTES: u64 = 1 << 20; /// The most bytes one record may take in an answer. One record must fit an answer on its own, so /// a larger one can never be read through the service. @@ -141,6 +142,8 @@ enum Work { bodies: Vec, /// The answer's byte budget left out records of the page before the codec step. cut: bool, + /// The call named a floor holder, so the answer carries its replay range. + positions: bool, }, } @@ -269,7 +272,8 @@ impl NexusCalls { /// The answer carries the read's state as it stood after the last record kept, also when the /// byte budget cut it short, so a reader's marks stay exact across calls and Workers. A record /// the store's reader can't parse crosses as an error at its cursor, which the reader goes on - /// past. + /// past. A call with a floor holder holds the records after its start for that holder, and its + /// answer carries the range of the records that crossed, so replay can read them again. pub async fn read(&self, call: NexusCall) -> NexusStep { let limits = Limits::of(&call); let input: ReadInput = match serde_json::from_slice(&call.input) { @@ -293,12 +297,19 @@ impl NexusCalls { ))); } let after = input.after_token.unwrap_or_default(); + let floor_holder = input.floor_holder.unwrap_or_default(); if input.latest_only.unwrap_or(false) { if !after.is_empty() { return failed(failure::input_mistake( "latest_only starts a read at the newest record; it takes no after_token", )); } + if !floor_holder.is_empty() { + return failed(failure::input_mistake( + "latest_only answers no records, so it holds no floor; start the read that \ + holds one at the cursor latest_only answers", + )); + } return match self .streams .latest(LatestRequest { @@ -353,8 +364,9 @@ impl NexusCalls { max_records: max_records as u32, state: state.clone(), arm: input.arm.unwrap_or(false), - floor_holder: String::new(), + floor_holder: floor_holder.clone(), }; + let positions = !floor_holder.is_empty(); let mut read = match self.streams.read_carried(request).await { Ok(read) => read, Err(error) => { @@ -363,7 +375,7 @@ impl NexusCalls { } else { state }; - return read_failure(error, state); + return read_failure(error, state, positions); } }; // Cut by the stored bodies first, so the codec decodes only what can cross. A codec that @@ -411,6 +423,7 @@ impl NexusCalls { limits, bodies, cut, + positions, }; if payloads.is_empty() { return self.finish(work, Vec::new()).await; @@ -539,7 +552,8 @@ impl NexusCalls { limits, bodies, cut, - } => finish_read(*read, limits, &bodies, cut, results), + positions, + } => finish_read(*read, limits, &bodies, cut, positions, results), } } @@ -673,6 +687,7 @@ fn finish_read( limits: Limits, bodies: &[usize], cut: bool, + positions: bool, results: Vec, ) -> NexusStep { if results.len() != bodies.len() { @@ -697,29 +712,28 @@ fn finish_read( } = read; let mut cut = cut; let mut wires: Vec = Vec::with_capacity(response.records.len()); - let mut kept: Vec = Vec::new(); + // The records that crossed, without their bodies, for the state at a cut and the digest. + let mut kept: Vec = Vec::with_capacity(response.records.len()); for (index, record) in response.records.into_iter().enumerate() { - let ReadRecord { - cursor, - record, - stale, - } = record; - let (wire, stored) = match record { + let wire = match &record.record { Some(read_record::Record::Superseded(superseded)) => { - (superseded_wire(cursor, &superseded), None) - } - Some(read_record::Record::Stored(stored)) => { - let wire = record_wire(cursor, &stored, decoded.remove(&index), stale, limits); - (wire, Some(stored)) + superseded_wire(record.cursor.clone(), superseded) } + Some(read_record::Record::Stored(stored)) => record_wire( + record.cursor.clone(), + stored, + decoded.remove(&index), + record.stale, + limits, + ), None => continue, }; - if !budget.admits(json_size(&wire), stored.is_some()) { + if !budget.admits(json_size(&wire), wire.superseded.is_none()) { cut = true; break; } wires.push(wire); - kept.extend(stored); + kept.push(record); } // An answer the budget cut holds records, so only a whole empty answer stays armed. let armed = response.armed && !cut && wires.is_empty(); @@ -727,8 +741,9 @@ fn finish_read( // A supersession belongs to the record after it, so it goes into the next answer too. if wires.last().is_some_and(|wire| wire.superseded.is_some()) { wires.pop(); + kept.pop(); } - state = state.at_cut(&before, &kept); + state = state.at_cut(&before, kept.iter().filter_map(stored_of)); let last = wires .last() .map(|wire| wire.token.clone()) @@ -738,16 +753,29 @@ fn finish_read( (response.cursor, response.done) }; state.fit_within(MAX_STATE_BYTES); + let replay = positions.then(|| ReplayRange { + chain_first_run_id: response.chain_first_run_id, + after_token: response.start_cursor, + stored_records: kept.iter().filter_map(stored_of).count() as i64, + records_digest: Base64(records_digest(&kept)), + }); output(&ReadOutput { records: wires, next_token, done, state: Some(Base64(state.encode_to_vec())), armed, - replay: None, + replay, }) } +fn stored_of(record: &ReadRecord) -> Option<&StreamRecord> { + match &record.record { + Some(read_record::Record::Stored(stored)) => Some(stored), + _ => None, + } +} + fn superseded_wire(token: String, superseded: &crate::proto::Supersession) -> RecordWire { RecordWire { token, @@ -807,19 +835,28 @@ fn record_wire( } /// How a read's failure crosses. A record the store's reader can't parse ends no read: it -/// crosses as an error at its cursor, and the reader goes on past it with the state it had. -fn read_failure(error: StreamError, state: Vec) -> NexusStep { +/// crosses as an error at its cursor, and the reader goes on past it with the state it had. With +/// `positions` the answer's range is empty at that cursor, since a replay can't read the record +/// either, and the reader keeps the error itself. +fn read_failure(error: StreamError, state: Vec, positions: bool) -> NexusStep { match (error.kind, error.cursor.clone()) { (crate::proto::StreamFailureKind::Record, Some(cursor)) => output(&ReadOutput { records: vec![RecordWire { error: Some(failure::capped(&error.message)), ..stored_wire(cursor.clone(), Base64::default(), false) }], - next_token: cursor, + next_token: cursor.clone(), done: false, + replay: positions.then(|| ReplayRange { + chain_first_run_id: ReadState::decode_from(&state) + .map(|state| state.first_run_id) + .unwrap_or_default(), + after_token: cursor, + stored_records: 0, + records_digest: Base64(records_digest(&[])), + }), state: (!state.is_empty()).then_some(Base64(state)), armed: false, - replay: None, }), _ => failed(failure::of_stream(&error)), } diff --git a/crates/streams/tests/nexus_contract.rs b/crates/streams/tests/nexus_contract.rs index 3237f4a023..f313428853 100644 --- a/crates/streams/tests/nexus_contract.rs +++ b/crates/streams/tests/nexus_contract.rs @@ -327,6 +327,11 @@ async fn every_output_member_is_declared() { "no answer carried {member}" ); } + let mut holding = json!({"stream": stream(wf, "out")}); + holding["floor_holder"] = json!("ns/run-1/reader"); + let held = read(&calls, holding).await.unwrap(); + check_output(&contract, "ReadOutput", &held); + check_output(&contract, "ReplayRange", &held["replay"]); let latest = read( &calls, json!({"stream": stream(wf, "out"), "latest_only": true}), diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index 234fb0d703..3297515c62 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -22,7 +22,10 @@ use temporalio_common::protos::temporal::api::{ use temporalio_streams::{ MemoryStore, OwnerClient, OwnerDescription, OwnerError, StreamStore, Streams, StreamsOptions, nexus::{NexusCalls, NexusOptions}, - proto::{ChainId, NexusFailure, StoreAppendRequest, StreamRecord, StreamRecordKind}, + proto::{ + ChainId, NexusFailure, ReadRangeRequest, ReadRequest, StoreAppendRequest, StreamAddress, + StreamOwnerKind, StreamRecord, StreamRecordKind, + }, }; static IDS: AtomicUsize = AtomicUsize::new(0); @@ -886,3 +889,261 @@ async fn an_expired_cursor_is_told_apart() { ("BAD_REQUEST", false, "StreamExpiredError") ); } + +const HOLDER: &str = "ns/reader-run/1"; + +fn holding(wf: &str, after: &str) -> Value { + let mut input = read_input(wf, after); + input["floor_holder"] = json!(HOLDER); + input +} + +fn chain_of(wf: &str) -> ChainId { + ChainId { + namespace: NAMESPACE.to_string(), + workflow_id: wf.to_string(), + first_run_id: "run-1".to_string(), + } +} + +/// The Nexus calls over `store`, and the stream layer they share. +fn layered(store: Arc) -> (Arc, NexusCalls) { + let streams = Arc::new(Streams::new( + store, + Arc::new(Running), + StreamsOptions::default(), + )); + let calls = NexusCalls::new(streams.clone(), NexusOptions::default()); + (streams, calls) +} + +/// Core's replay read of the range `answer` carries, for the read that sent `state`. +fn replay_of(wf: &str, answer: &Value, state: &[u8]) -> ReadRangeRequest { + use base64::Engine; + let replay = &answer["replay"]; + ReadRangeRequest { + stream: Some(StreamAddress { + namespace: NAMESPACE.to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: wf.to_string(), + run_id: String::new(), + topic: TOPIC.to_string(), + }), + chain_first_run_id: replay["chain_first_run_id"].as_str().unwrap().to_string(), + after: replay["after_token"].as_str().unwrap().to_string(), + to: token(answer).to_string(), + stored_records: replay["stored_records"].as_u64().unwrap() as u32, + state: state.to_vec(), + records_digest: base64::engine::general_purpose::STANDARD + .decode(replay["records_digest"].as_str().unwrap()) + .unwrap(), + floor_holder: String::new(), + start_from_end: false, + } +} + +#[tokio::test] +async fn a_read_without_a_floor_holder_answers_no_range() { + let calls = service(memory(), NexusOptions::default()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + let answer = read_after(&calls, &wf, "").await; + assert!(answer.get("replay").is_none(), "{answer}"); +} + +#[tokio::test] +async fn a_read_with_a_floor_holder_holds_what_it_read_against_retention() { + let memory = Arc::new(MemoryStore::new()); + let calls = service(memory.clone(), NexusOptions::default()); + let held = workflow(); + let free = workflow(); + let mut starts = Vec::new(); + for wf in [&held, &free] { + let first = write(&calls, wf, 1, &["one"]).await; + write(&calls, wf, 2, &["two", "three"]).await; + starts.push(first); + } + let answer = read(&calls, holding(&held, &starts[0])).await.unwrap(); + assert_eq!(texts(&answer), ["two", "three"]); + let replay = &answer["replay"]; + assert_eq!(replay["chain_first_run_id"], "run-1"); + assert_eq!(replay["after_token"], starts[0].as_str()); + assert_eq!(replay["stored_records"], 2); + read_after(&calls, &free, &starts[1]).await; + for wf in [&held, &free] { + memory.truncate(&chain_of(wf), TOPIC, 0); + } + assert_eq!( + texts(&read_after(&calls, &held, &starts[0]).await), + ["two", "three"], + "the floor kept the records after the read's start" + ); + let refused = read(&calls, read_input(&free, &starts[1])) + .await + .unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamExpiredError") + ); +} + +#[cfg(feature = "redis")] +#[tokio::test] +async fn a_read_with_a_floor_holder_holds_what_it_read_on_redis() { + let Ok(url) = std::env::var("STREAMS_REDIS_URL") else { + eprintln!("set STREAMS_REDIS_URL to run the Redis cases"); + return; + }; + let mut options = temporalio_streams::RedisStoreOptions::new(url); + options.key_prefix = workflow(); + options.retention = Duration::from_millis(200); + options.floor_lease = Some(Duration::from_secs(60)); + let store = Arc::new( + temporalio_streams::RedisStore::connect(options) + .await + .unwrap(), + ); + let calls = service(store, NexusOptions::default()); + let wf = workflow(); + let first = write(&calls, &wf, 1, &["one"]).await; + write(&calls, &wf, 2, &["two", "three"]).await; + let answer = read(&calls, holding(&wf, &first)).await.unwrap(); + assert_eq!(answer["replay"]["stored_records"], 2); + // Past the retention, an append's trim would drop the records without the floor. + tokio::time::sleep(Duration::from_millis(400)).await; + write(&calls, &wf, 4, &["four"]).await; + assert_eq!( + texts(&read_after(&calls, &wf, &first).await), + ["two", "three", "four"] + ); +} + +#[tokio::test] +async fn a_cut_answer_ranges_over_the_records_that_crossed() { + let (streams, calls) = layered(memory()); + let wf = workflow(); + let big = "x".repeat(300 * 1024); + write(&calls, &wf, 1, &[&format!("1{big}"), &format!("2{big}")]).await; + // The newer attempt's supersession fits the budget, but its record doesn't, so the cut + // leaves the supersession for the next answer. + append(&calls, append_input(&wf, "p", 2, 1, &[&format!("3{big}")])) + .await + .unwrap(); + let answer = read(&calls, holding(&wf, "")).await.unwrap(); + assert_eq!(records(&answer).len(), 2, "the budget cut the answer"); + assert!( + records(&answer) + .iter() + .all(|r| r.get("superseded").is_none()) + ); + let replay = &answer["replay"]; + assert_eq!(replay["after_token"], ""); + assert_eq!(replay["stored_records"], 2); + let replayed = streams + .read_range(replay_of(&wf, &answer, &[])) + .await + .unwrap(); + assert_eq!(replayed.records.len(), 2); + assert_eq!(replayed.cursor, token(&answer)); + // Core's page went on past the cut, so its digest is not the answer's. + let page = streams + .read(ReadRequest { + stream: replay_of(&wf, &answer, &[]).stream, + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(page.records.len(), 4); + assert_ne!( + json!(base64_of(&page.records_digest)), + replay["records_digest"] + ); +} + +fn base64_of(bytes: &[u8]) -> String { + use base64::Engine; + base64::engine::general_purpose::STANDARD.encode(bytes) +} + +#[tokio::test] +async fn a_range_replays_with_the_state_the_read_sent() { + let (streams, calls) = layered(memory()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + let first = read(&calls, holding(&wf, "")).await.unwrap(); + append(&calls, append_input(&wf, "p", 2, 1, &["two"])) + .await + .unwrap(); + let mut input = holding(&wf, token(&first)); + input["state"] = first["state"].clone(); + let second = read(&calls, input).await.unwrap(); + assert_eq!(values(&second), ["SUPERSEDED p 1->2", "p@2 two"]); + let state = { + use base64::Engine; + base64::engine::general_purpose::STANDARD + .decode(first["state"].as_str().unwrap()) + .unwrap() + }; + let replayed = streams + .read_range(replay_of(&wf, &second, &state)) + .await + .unwrap(); + assert_eq!( + replayed.records.len(), + 2, + "the supersession comes out again" + ); +} + +#[tokio::test] +async fn a_record_the_reader_cannot_parse_has_an_empty_range_at_its_cursor() { + let memory = Arc::new(MemoryStore::new()); + let (streams, calls) = layered(memory.clone()); + let wf = workflow(); + memory + .append(StoreAppendRequest { + chain: Some(chain_of(&wf)), + topic: TOPIC.to_string(), + producer_id: "raw".to_string(), + attempt: 1, + sequence: 1, + digest: vec![1; 32], + records: vec![b"\xff".to_vec()], + }) + .await + .unwrap(); + let answer = read(&calls, holding(&wf, "")).await.unwrap(); + let [wire] = records(&answer).as_slice() else { + panic!("expected one record, got {answer}"); + }; + assert!(wire.get("error").is_some(), "{answer}"); + let replay = &answer["replay"]; + assert_eq!(replay["after_token"], token(&answer)); + assert_eq!(replay["stored_records"], 0); + assert_eq!(replay["chain_first_run_id"], "run-1"); + let replayed = streams + .read_range(replay_of(&wf, &answer, &[])) + .await + .unwrap(); + assert!(replayed.records.is_empty()); +} + +#[tokio::test] +async fn a_floor_holder_is_at_most_512_bytes_and_never_beside_latest_only() { + let calls = service(memory(), NexusOptions::default()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + let mut input = read_input(&wf, ""); + input["floor_holder"] = json!("h".repeat(512)); + read(&calls, input.clone()).await.unwrap(); + input["floor_holder"] = json!("h".repeat(513)); + let refused = read(&calls, input).await.unwrap_err(); + assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); + assert!(refused.message.contains("512"), "{}", refused.message); + let mut latest = holding(&wf, ""); + latest.as_object_mut().unwrap().remove("after_token"); + latest["latest_only"] = json!(true); + let refused = read(&calls, latest).await.unwrap_err(); + assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); + assert!(refused.message.contains("floor"), "{}", refused.message); +} From a58f2375578b2348414a7eac0f8ec341616dd2c1 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 09:10:11 -0700 Subject: [PATCH 04/11] Served `read_range` through the codec step. Replay of a positions read gets its records from the owner's endpoint. The answer has no byte budget, since it must give exactly what the read gave. --- .../temporal/sdk/core/streams/nexus.proto | 2 + .../nexus/temporal_streams.nexusrpc.yaml | 4 +- crates/streams/src/handle.rs | 2 +- crates/streams/src/nexus/mod.rs | 249 ++++++++++++++---- crates/streams/src/nexus/wire.rs | 7 - crates/streams/src/service.rs | 15 +- crates/streams/tests/nexus_contract.rs | 63 +++++ crates/streams/tests/nexus_service.rs | 117 ++++++++ crates/streams/tests/nexus_support/mod.rs | 18 ++ 9 files changed, 416 insertions(+), 61 deletions(-) diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto b/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto index 3c89ae53c2..8ceabee1bc 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto @@ -19,6 +19,8 @@ service StreamNexusService { rpc NexusAppend(NexusCall) returns (NexusStep); // The stream service's Nexus `read` operation, over its JSON transfer form. rpc NexusRead(NexusCall) returns (NexusStep); + // The stream service's Nexus `read_range` operation, over its JSON transfer form. + rpc NexusReadRange(NexusCall) returns (NexusStep); // Goes on with a call that asked for codec work. rpc NexusResume(NexusResumeRequest) returns (NexusStep); // Attaches the caller of a stream-returning Nexus operation's start to the stream's notifier, diff --git a/crates/streams/nexus/temporal_streams.nexusrpc.yaml b/crates/streams/nexus/temporal_streams.nexusrpc.yaml index 99bc5e5a9a..9c05215b17 100644 --- a/crates/streams/nexus/temporal_streams.nexusrpc.yaml +++ b/crates/streams/nexus/temporal_streams.nexusrpc.yaml @@ -62,7 +62,9 @@ services: owner and has no byte budget, since it must answer exactly what the read answered. Records with other content are refused as StreamReplayMismatchError, and records the store no longer holds as - StreamReplayRangeLostError. + StreamReplayRangeLostError. The answer holds the records and + next_token only, since the caller keeps the read's state, done and + replay. input: { $ref: "#/$defs/ReadRangeInput" } output: { $ref: "#/$defs/ReadOutput" } diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index a844f0b17b..9a1cc47685 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -388,7 +388,7 @@ impl Streams { /// Refuses a carried state whose chain is not one of the Workflow's. A caller outside the /// process could otherwise point a read at another run's chain by editing its state. Asked /// once per chain in a process. - async fn check_carried_chain( + pub(crate) async fn check_carried_chain( &self, stream: &StreamAddress, first_run_id: &str, diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index 2170e0b17d..2d86db4b8f 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -1,10 +1,10 @@ //! The stream service's Nexus operations, served by Core. //! -//! Lang forwards each `append` and `read` call as N0a's JSON transfer form and gets back a -//! [NexusStep]. A call whose bodies need lang's payload codec answers first with [CodecWork], and -//! lang resumes it with the results. Core runs every rule of the contract, so each SDK's handler -//! keeps only its authorizer and its codec. The state a reader carries makes the endpoint -//! stateless, so a read goes on at any handler Worker. +//! Lang forwards each `append`, `read` and `read_range` call as N0a's JSON transfer form and gets +//! back a [NexusStep]. A call whose bodies need lang's payload codec answers first with +//! [CodecWork], and lang resumes it with the results. Core runs every rule of the contract, so +//! each SDK's handler keeps only its authorizer and its codec. The state a reader carries makes +//! the endpoint stateless, so a read goes on at any handler Worker. mod failure; mod operation; @@ -18,9 +18,9 @@ use crate::{ handle::CarriedRead, proto::{ AppendRecord, AppendRequest, CodecResult, CodecWork, LatestRequest, NamedProducer, - NexusCall, NexusFailure, NexusResumeRequest, NexusStep, ReadRecord, ReadRequest, - StreamAddress, StreamOwnerKind, StreamRecord, StreamRecordKind, append_request::Producer, - codec_result, nexus_step, read_record, + NexusCall, NexusFailure, NexusResumeRequest, NexusStep, ReadRangeRequest, ReadRecord, + ReadRequest, ReadResponse, StreamAddress, StreamOwnerKind, StreamRecord, StreamRecordKind, + append_request::Producer, codec_result, nexus_step, read_record, }, reader::{ReadState, records_digest}, }; @@ -37,8 +37,8 @@ use std::{ use temporalio_common::protos::temporal::api::common::v1::Payload; use tokio::time::Instant; use wire::{ - AppendInput, AppendOutput, Base64, ReadInput, ReadOutput, RecordWire, ReplayRange, StreamRef, - Supersession, crossing_size, json_size, + AppendInput, AppendOutput, Base64, ReadInput, ReadOutput, ReadRangeInput, RecordWire, + ReplayRange, StreamRef, Supersession, crossing_size, json_size, }; /// The most bytes one read answer takes as JSON past its first record, with its tokens and state. @@ -145,6 +145,12 @@ enum Work { /// The call named a floor holder, so the answer carries its replay range. positions: bool, }, + ReadRange { + response: Box, + limits: Limits, + /// As for [Work::Read]. + bodies: Vec, + }, } #[derive(Debug, Clone, Copy)] @@ -336,20 +342,9 @@ impl NexusCalls { )) }); let wait = shortened(Duration::from_millis(wait_ms as u64), deadline); - let state = match input.state { - None => Vec::new(), - Some(text) if text.len() > crossing_size(MAX_STATE_BYTES) => { - return failed(failure::of_stream(&StreamError::cursor(format!( - "the read state takes {} bytes, above the {} of any state a read answers, so \ - no earlier call of this read returned it", - text.len(), - crossing_size(MAX_STATE_BYTES) - )))); - } - Some(text) => match wire::base64_bytes(&text) { - Some(state) => state, - None => return failed(failure::input_mistake("state must be base64-encoded")), - }, + let state = match carried_state(input.state) { + Ok(state) => state, + Err(failure) => return failed(*failure), }; // A ref's run only picks the chain, so the read names the chain by its state and follows // it past that run's end. @@ -406,18 +401,7 @@ impl NexusCalls { } let cut = keep < read.response.records.len(); read.response.records.truncate(keep); - let mut bodies = Vec::new(); - let mut payloads = Vec::new(); - let mut bytes = 0; - for (index, record) in read.response.records.iter_mut().enumerate() { - if let Some(read_record::Record::Stored(stored)) = &mut record.record - && let Some(body) = stored.body.take() - { - bytes += body.encoded_len(); - bodies.push(index); - payloads.push(body); - } - } + let (bodies, payloads, bytes) = take_bodies(&mut read.response.records); let work = Work::Read { read: Box::new(read), limits, @@ -438,6 +422,79 @@ impl NexusCalls { ) } + /// Serves the `read_range` operation: the records of an earlier read that named a floor + /// holder, read again and checked against that read's digest. + /// + /// The answer has no byte budget, since it must hold exactly the records the read answered, + /// and those fit an answer once. It carries no state and no replay, because the caller keeps + /// the read's own. The chain comes from the caller, so it is checked to be one of the + /// Workflow's, as a carried read state is. + pub async fn read_range(&self, call: NexusCall) -> NexusStep { + let limits = Limits::of(&call); + let input: ReadRangeInput = match serde_json::from_slice(&call.input) { + Ok(input) => input, + Err(error) => return failed(failure::input_mistake(input_error("read_range", error))), + }; + let mut address = match address(&call.namespace, &input.stream) { + Ok(address) => address, + Err(failure) => return failed(*failure), + }; + // The chain names the run chain, so the ref's run picks nothing here. + address.run_id.clear(); + if !(0..=MAX_RECORDS).contains(&input.stored_records) { + return failed(failure::input_mistake(format!( + "stored_records is from 0 to {MAX_RECORDS}, not {}", + input.stored_records + ))); + } + let state = match carried_state(input.state) { + Ok(state) => state, + Err(failure) => return failed(*failure), + }; + if !input.chain_first_run_id.is_empty() + && let Err(error) = self + .streams + .check_carried_chain(&address, &input.chain_first_run_id) + .await + { + return failed(failure::of_stream(&error)); + } + let request = ReadRangeRequest { + stream: Some(address), + chain_first_run_id: input.chain_first_run_id, + after: input.after_token, + to: input.to_token, + stored_records: input.stored_records as u32, + state, + records_digest: input.records_digest.0, + floor_holder: input.floor_holder.unwrap_or_default(), + // A read that names a floor holder never starts at the end, since only latest_only + // does, and it refuses a holder. + start_from_end: false, + }; + let mut response = match self.streams.read_range(request).await { + Ok(response) => response, + Err(error) => return failed(failure::of_stream(&error)), + }; + let (bodies, payloads, bytes) = take_bodies(&mut response.records); + let work = Work::ReadRange { + response: Box::new(response), + limits, + bodies, + }; + if payloads.is_empty() { + return self.finish(work, Vec::new()).await; + } + self.ask_codec( + work, + bytes, + false, + &call.namespace, + &input.stream.workflow_id, + payloads, + ) + } + /// `address` without its run, and a state that names the chain the run picked. A state the /// caller passed back already names its chain. async fn chain_by_state( @@ -554,6 +611,11 @@ impl NexusCalls { cut, positions, } => finish_read(*read, limits, &bodies, cut, positions, results), + Work::ReadRange { + response, + limits, + bodies, + } => finish_read_range(*response, limits, &bodies, results), } } @@ -690,20 +752,10 @@ fn finish_read( positions: bool, results: Vec, ) -> NexusStep { - if results.len() != bodies.len() { - return failed(wrong_count(results.len(), bodies.len())); - } - let mut decoded: HashMap> = HashMap::new(); - for (index, result) in bodies.iter().zip(results) { - decoded.insert( - *index, - match result.result { - Some(codec_result::Result::Payload(body)) => Ok(body), - Some(codec_result::Result::Error(error)) => Err(error), - None => Err("the payload codec answered nothing".to_string()), - }, - ); - } + let mut decoded = match decoded(bodies, results) { + Ok(decoded) => decoded, + Err(failure) => return failed(*failure), + }; let mut budget = Budget::new(limits, &read); let CarriedRead { response, @@ -769,6 +821,85 @@ fn finish_read( }) } +/// A replay's answer. Each record crosses as the read's did, with no budget to cut it. +fn finish_read_range( + response: ReadResponse, + limits: Limits, + bodies: &[usize], + results: Vec, +) -> NexusStep { + let mut decoded = match decoded(bodies, results) { + Ok(decoded) => decoded, + Err(failure) => return failed(*failure), + }; + let records = response + .records + .into_iter() + .enumerate() + .filter_map(|(index, record)| match record.record { + Some(read_record::Record::Superseded(superseded)) => { + Some(superseded_wire(record.cursor, &superseded)) + } + Some(read_record::Record::Stored(stored)) => Some(record_wire( + record.cursor, + &stored, + decoded.remove(&index), + record.stale, + limits, + )), + None => None, + }) + .collect(); + output(&ReadOutput { + records, + next_token: response.cursor, + done: false, + state: None, + armed: false, + replay: None, + }) +} + +/// Takes the bodies out of `records` for lang's codec, answering which records had one, the +/// bodies, and their bytes. +fn take_bodies(records: &mut [ReadRecord]) -> (Vec, Vec, usize) { + let mut bodies = Vec::new(); + let mut payloads = Vec::new(); + let mut bytes = 0; + for (index, record) in records.iter_mut().enumerate() { + if let Some(read_record::Record::Stored(stored)) = &mut record.record + && let Some(body) = stored.body.take() + { + bytes += body.encoded_len(); + bodies.push(index); + payloads.push(body); + } + } + (bodies, payloads, bytes) +} + +/// Lang's codec results by the index of the record each body came from. +fn decoded( + bodies: &[usize], + results: Vec, +) -> Result>, Box> { + if results.len() != bodies.len() { + return Err(Box::new(wrong_count(results.len(), bodies.len()))); + } + Ok(bodies + .iter() + .zip(results) + .map(|(index, result)| { + let body = match result.result { + Some(codec_result::Result::Payload(body)) => Ok(body), + Some(codec_result::Result::Error(error)) => Err(error), + None => Err("the payload codec answered nothing".to_string()), + }; + (*index, body) + }) + .collect()) +} + fn stored_of(record: &ReadRecord) -> Option<&StreamRecord> { match &record.record { Some(read_record::Record::Stored(stored)) => Some(stored), @@ -862,6 +993,24 @@ fn read_failure(error: StreamError, state: Vec, positions: bool) -> NexusSte } } +/// The state a call carries, decoded. One longer than any state a read answers is refused before +/// it is decoded. +fn carried_state(state: Option) -> Result, Box> { + match state { + None => Ok(Vec::new()), + Some(text) if text.len() > crossing_size(MAX_STATE_BYTES) => { + Err(Box::new(failure::of_stream(&StreamError::cursor(format!( + "the read state takes {} bytes, above the {} of any state a read answers, so no \ + earlier call of this read returned it", + text.len(), + crossing_size(MAX_STATE_BYTES) + ))))) + } + Some(text) => wire::base64_bytes(&text) + .ok_or_else(|| Box::new(failure::input_mistake("state must be base64-encoded"))), + } +} + /// The stream address a call names. An owner kind this release can't open is refused before any /// codec work. fn address(namespace: &str, stream: &StreamRef) -> Result> { diff --git a/crates/streams/src/nexus/wire.rs b/crates/streams/src/nexus/wire.rs index 41895ae2a6..987336821a 100644 --- a/crates/streams/src/nexus/wire.rs +++ b/crates/streams/src/nexus/wire.rs @@ -104,13 +104,6 @@ pub(crate) struct ReadInput { #[derive(Debug, Clone, PartialEq, Eq, Deserialize)] #[serde(deny_unknown_fields)] -#[cfg_attr( - not(test), - expect( - dead_code, - reason = "the contract declares the model before Core serves it" - ) -)] pub(crate) struct ReadRangeInput { pub(crate) stream: StreamRef, pub(crate) chain_first_run_id: String, diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index 06be3617fa..06a960bcb6 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -238,6 +238,11 @@ impl StreamService { .read(decode::(rpc, request)?) .await .encode_to_vec()), + "NexusReadRange" => Ok(self + .nexus + .read_range(decode::(rpc, request)?) + .await + .encode_to_vec()), "NexusResume" => Ok(self .nexus .resume(decode::(rpc, request)?) @@ -1285,7 +1290,13 @@ mod tests { .unwrap(); let step = NexusStep::decode(answer.as_slice()).unwrap(); assert!(matches!(step.step, Some(nexus_step::Step::Failure(_)))); - let invalid = service.call("NexusRead", b"\xff").await.unwrap_err(); - assert_eq!(invalid.kind, StreamFailureKind::InvalidArgument as i32); + for rpc in ["NexusRead", "NexusReadRange"] { + let invalid = service.call(rpc, b"\xff").await.unwrap_err(); + assert_eq!( + invalid.kind, + StreamFailureKind::InvalidArgument as i32, + "{rpc}" + ); + } } } diff --git a/crates/streams/tests/nexus_contract.rs b/crates/streams/tests/nexus_contract.rs index f313428853..f22b577dcc 100644 --- a/crates/streams/tests/nexus_contract.rs +++ b/crates/streams/tests/nexus_contract.rs @@ -253,6 +253,69 @@ async fn every_declared_input_member_is_accepted_and_nothing_else() { } } +#[tokio::test] +async fn every_read_range_member_is_accepted_and_nothing_else() { + let contract = contract(); + let calls = service(memory(), NexusOptions::default()); + let wf = "contract-read-range"; + append( + &calls, + json!({"stream": stream(wf, "out"), "producer_id": "p", "attempt": 1, "sequence": 1, + "payloads": [body("a")]}), + ) + .await + .unwrap(); + let first = read( + &calls, + json!({"stream": stream(wf, "out"), "floor_holder": "ns/run-1/reader"}), + ) + .await + .unwrap(); + append( + &calls, + json!({"stream": stream(wf, "out"), "producer_id": "p", "attempt": 1, "sequence": 2, + "payloads": [body("b")]}), + ) + .await + .unwrap(); + let second = read( + &calls, + json!({"stream": stream(wf, "out"), "after_token": first["next_token"], + "state": first["state"], "floor_holder": "ns/run-1/reader"}), + ) + .await + .unwrap(); + let mut full = replay_input(stream(wf, "out"), &second, first["state"].clone()); + full["stream"]["chain_run_id"] = Value::Null; + full["floor_holder"] = json!("ns/run-1/reader"); + let mut declared = properties(model(&contract, "ReadRangeInput")); + let mut sent: Vec = full.as_object().unwrap().keys().cloned().collect(); + declared.sort(); + sent.sort(); + assert_eq!( + sent, declared, + "the contract changed ReadRangeInput; teach this test" + ); + let answer = read_range(&calls, full.clone()).await.unwrap(); + check_output(&contract, "ReadOutput", &answer); + assert_eq!(answer["records"], second["records"]); + let mut unknown = full.clone(); + unknown["surprise"] = json!(1); + let failure = read_range(&calls, unknown).await.unwrap_err(); + assert!(is_contract_mistake(&failure), "{}", failure.message); + assert_eq!(failure.handler_error_type, "BAD_REQUEST"); + for member in required(model(&contract, "ReadRangeInput")) { + let mut missing = full.clone(); + missing.as_object_mut().unwrap().remove(&member); + let failure = read_range(&calls, missing).await.unwrap_err(); + assert!( + is_contract_mistake(&failure), + "{member}: {}", + failure.message + ); + } +} + #[tokio::test] async fn every_output_member_is_declared() { let contract = contract(); diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index 3297515c62..8296c4a4e0 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -1147,3 +1147,120 @@ async fn a_floor_holder_is_at_most_512_bytes_and_never_beside_latest_only() { assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); assert!(refused.message.contains("floor"), "{}", refused.message); } + +/// A cut answer and the answer after it, with a supersession, each read again through +/// `read_range`, cross the same way the read's did. +async fn a_range_replays_as_it_crossed(store: Arc) { + let calls = service(store, NexusOptions::default()); + let wf = workflow(); + let big = "x".repeat(300 * 1024); + write(&calls, &wf, 1, &[&format!("1{big}"), &format!("2{big}")]).await; + append(&calls, append_input(&wf, "p", 2, 1, &[&format!("3{big}")])) + .await + .unwrap(); + let first = read(&calls, holding(&wf, "")).await.unwrap(); + assert_eq!(records(&first).len(), 2, "the budget cut the answer"); + let mut input = holding(&wf, token(&first)); + input["state"] = first["state"].clone(); + let second = read(&calls, input).await.unwrap(); + assert_eq!(values(&second)[0], "SUPERSEDED p 1->2"); + for (answer, state) in [(&first, Value::Null), (&second, first["state"].clone())] { + let replayed = read_range(&calls, replay_input(stream(&wf, TOPIC), answer, state)) + .await + .unwrap(); + assert_eq!(replayed["records"], answer["records"]); + assert_eq!(replayed["next_token"], answer["next_token"]); + assert_eq!(replayed["done"], false); + assert!(replayed.get("state").is_none() && replayed.get("replay").is_none()); + } +} + +#[tokio::test] +async fn a_range_replays_as_it_crossed_on_memory() { + a_range_replays_as_it_crossed(memory()).await; +} + +#[cfg(feature = "redis")] +#[tokio::test] +async fn a_range_replays_as_it_crossed_on_redis() { + let Some(store) = on_redis().await else { + return; + }; + a_range_replays_as_it_crossed(store).await; +} + +/// A held read of two records on `wf`, through `calls`. +async fn held_read(calls: &NexusCalls, wf: &str) -> Value { + write(calls, wf, 1, &["one", "two"]).await; + read(calls, holding(wf, "")).await.unwrap() +} + +#[tokio::test] +async fn a_range_with_other_records_is_a_mismatch() { + let calls = service(memory(), NexusOptions::default()); + let wf = workflow(); + let answer = held_read(&calls, &wf).await; + let mut other_digest = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + other_digest["records_digest"] = json!(base64_of(&[0; 32])); + let mut other_count = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + other_count["stored_records"] = json!(1); + for input in [other_digest, other_count] { + let refused = read_range(&calls, input).await.unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamReplayMismatchError") + ); + } +} + +#[tokio::test] +async fn a_range_the_store_dropped_is_lost_and_a_replay_holds_it_again() { + let memory = Arc::new(MemoryStore::new()); + let (streams, calls) = layered(memory.clone()); + let wf = workflow(); + let answer = held_read(&calls, &wf).await; + let release = |holder: &str| temporalio_streams::proto::ReleaseFloorRequest { + stream: replay_of(&wf, &answer, &[]).stream, + chain_first_run_id: "run-1".to_string(), + floor_holder: holder.to_string(), + }; + streams.release_floor(release(HOLDER)).await.unwrap(); + // A replay that names a holder holds the range again, as after a reset. + let mut renewing = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + renewing["floor_holder"] = json!("ns/reset-run/1"); + read_range(&calls, renewing).await.unwrap(); + memory.truncate(&chain_of(&wf), TOPIC, 0); + let input = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + let replayed = read_range(&calls, input.clone()).await.unwrap(); + assert_eq!(replayed["records"], answer["records"]); + streams + .release_floor(release("ns/reset-run/1")) + .await + .unwrap(); + memory.truncate(&chain_of(&wf), TOPIC, 0); + let refused = read_range(&calls, input).await.unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamReplayRangeLostError") + ); +} + +#[tokio::test] +async fn a_range_names_a_chain_of_the_workflow_and_a_count_a_read_can_answer() { + let calls = service(memory(), NexusOptions::default()); + let wf = workflow(); + let answer = held_read(&calls, &wf).await; + let mut other_chain = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + other_chain["chain_first_run_id"] = json!("run-9"); + let refused = read_range(&calls, other_chain).await.unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamCursorError") + ); + for count in [-1, 1001] { + let mut input = replay_input(stream(&wf, TOPIC), &answer, Value::Null); + input["stored_records"] = json!(count); + let refused = read_range(&calls, input).await.unwrap_err(); + assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); + } +} diff --git a/crates/streams/tests/nexus_support/mod.rs b/crates/streams/tests/nexus_support/mod.rs index e805eb6c3e..bf60dfeb2f 100644 --- a/crates/streams/tests/nexus_support/mod.rs +++ b/crates/streams/tests/nexus_support/mod.rs @@ -171,6 +171,24 @@ pub(crate) async fn read(calls: &NexusCalls, input: Value) -> Result Result { + drive(calls, calls.read_range(call(input)).await, Codec::Sealing).await +} + +/// The `read_range` input that replays `answer`, the answer of a read that sent `state`. +pub(crate) fn replay_input(stream: Value, answer: &Value, state: Value) -> Value { + let replay = &answer["replay"]; + serde_json::json!({ + "stream": stream, + "chain_first_run_id": replay["chain_first_run_id"], + "after_token": replay["after_token"], + "to_token": answer["next_token"], + "stored_records": replay["stored_records"], + "state": state, + "records_digest": replay["records_digest"], + }) +} + /// A stream reference to `topic` of Workflow `workflow_id`. pub(crate) fn stream(workflow_id: &str, topic: &str) -> Value { serde_json::json!({"kind": "workflow", "workflow_id": workflow_id, "topic": topic}) From 4c2c3b848750dd0fa35e9ab8772cee9fdcf44fe4 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 09:12:50 -0700 Subject: [PATCH 05/11] Noted the Nexus floor holder and `read_range` in the changelog. --- crates/sdk-core/CHANGELOG.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index a97bb55d7f..64d58a5663 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -128,6 +128,11 @@ relevant information. newest record after the attach, or arms an empty stream, so a caller that only waits for progress hears of the records already there and of the first append. The API protos carry the `NexusHandler` progress fields and `nexus/notification-service.wit`. +* Experimental: the Nexus `read` takes `floor_holder`. With it the read holds the floor and + answers `replay`, the range of the records that crossed after the byte cut, with Core's digest + over them. The new Nexus `read_range` operation, `NexusReadRange` for lang, reads that range + again through the codec step and checks the digest. Its answer has no byte budget and carries + only the records and `next_token`. It checks that the chain it names is one of the Workflow's. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting From 41d2d86173ef4b1b6661aea43d909741046bcd7d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 09:18:31 -0700 Subject: [PATCH 06/11] Vendored and wired `ExecuteSyncNexusOperation`. A Workflow's direct reads call the owner's stream service with the caller's own credential. Lang reaches the RPC through Core's client. --- crates/client/src/grpc.rs | 9 ++++++ crates/common/build.rs | 1 + .../workflowservice/v1/request_response.proto | 31 +++++++++++++++++++ .../api/workflowservice/v1/service.proto | 14 +++++++++ crates/sdk-core-c-bridge/src/client.rs | 3 ++ crates/sdk-core/CHANGELOG.md | 3 ++ 6 files changed, 61 insertions(+) diff --git a/crates/client/src/grpc.rs b/crates/client/src/grpc.rs index ef9e212e41..5c88b10e0f 100644 --- a/crates/client/src/grpc.rs +++ b/crates/client/src/grpc.rs @@ -1924,6 +1924,15 @@ proxier! { r.extensions_mut().insert(IsUserLongPoll); } ); + ( + execute_sync_nexus_operation, + ExecuteSyncNexusOperationRequest, + ExecuteSyncNexusOperationResponse, + |r| { + let labels = namespaced_request!(r); + r.extensions_mut().insert(labels); + } + ); } proxier! { diff --git a/crates/common/build.rs b/crates/common/build.rs index b8a83a26ef..c66f6255f4 100644 --- a/crates/common/build.rs +++ b/crates/common/build.rs @@ -761,6 +761,7 @@ const BLOB_FIELDS: &[&str] = &[ "temporal.api.protocol.v1.Message.body", // whole Any body; see EXTRA_WHOLE_MESSAGE_LEAVES "temporal.api.query.v1.WorkflowQuery.query_args", "temporal.api.workflow.v1.NewWorkflowExecutionInfo.input", + "temporal.api.workflowservice.v1.ExecuteSyncNexusOperationRequest.input", "temporal.api.workflowservice.v1.NotifyStreamRequest.close_failure", "temporal.api.workflowservice.v1.NotifyStreamRequest.close_result", "temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatByIdRequest.details", diff --git a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto index aaa5bca8bc..eaffa1365a 100644 --- a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto +++ b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto @@ -3813,3 +3813,34 @@ message PollStreamNotifierResponse { // The failure the stream closed with, when `closed` is set and it closed with one. temporal.api.failure.v1.Failure close_failure = 6; } + +message ExecuteSyncNexusOperationRequest { + // The caller namespace. Authorization and the endpoint lookup run against it. + string namespace = 1; + // The identity of the client who sent this request. + string identity = 2; + // A unique identifier for this call. Typically UUIDv4. The handler gets it as the Nexus + // request ID. + string request_id = 3; + // Endpoint name, resolved through the cluster's endpoint registry for the caller namespace. + string endpoint = 4; + // Service name. + string service = 5; + // Operation name. + string operation = 6; + // Serialized input to the operation. Passed as the request payload. + temporal.api.common.v1.Payload input = 7; + // Header to attach to the Nexus request. Keys are lowercased. The server owns the + // `temporal-caller-namespace` key and refuses a request that sets it. + map nexus_header = 8; +} + +message ExecuteSyncNexusOperationResponse { + oneof outcome { + // The result of an operation that completed. + temporal.api.common.v1.Payload result = 1; + // The failure of the call. A handler error has `nexus_handler_failure_info`. An operation + // that failed or was canceled has `application_failure_info` or `canceled_failure_info`. + temporal.api.failure.v1.Failure failure = 2; + } +} diff --git a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto index 25cb8db62b..a27a6edfe1 100644 --- a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto +++ b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto @@ -2088,4 +2088,18 @@ service WorkflowService { // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc PollStreamNotifier (PollStreamNotifierRequest) returns (PollStreamNotifierResponse) { } + + // ExecuteSyncNexusOperation runs one synchronous Nexus operation on an endpoint for a caller in + // `namespace`, and returns its result or its failure. The server persists nothing and does not + // retry. Authorization runs against the caller namespace, and the endpoint must be reachable + // from it. An operation that answers asynchronously is refused with `FailedPrecondition`. Use + // `StartNexusOperationExecution` for those. + // + // The server sets the Nexus header `temporal-caller-namespace` to the caller namespace, so the + // handler can trust it. A request that sets this header is refused. + // + // (-- api-linter: core::0127::http-annotation=disabled + // aip.dev/not-precedent: SDKs call it through gRPC. --) + rpc ExecuteSyncNexusOperation (ExecuteSyncNexusOperationRequest) returns (ExecuteSyncNexusOperationResponse) { + } } diff --git a/crates/sdk-core-c-bridge/src/client.rs b/crates/sdk-core-c-bridge/src/client.rs index 516a04ad9c..9da2ee2919 100644 --- a/crates/sdk-core-c-bridge/src/client.rs +++ b/crates/sdk-core-c-bridge/src/client.rs @@ -719,6 +719,9 @@ async fn call_workflow_service( "ExecuteMultiOperation" => { rpc_call_on_trait!(client, call, WorkflowService, execute_multi_operation) } + "ExecuteSyncNexusOperation" => { + rpc_call_on_trait!(client, call, WorkflowService, execute_sync_nexus_operation) + } "FetchWorkerConfig" => { rpc_call_on_trait!(client, call, WorkflowService, fetch_worker_config) } diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 64d58a5663..c344518b09 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -133,6 +133,9 @@ relevant information. over them. The new Nexus `read_range` operation, `NexusReadRange` for lang, reads that range again through the codec step and checks the digest. Its answer has no byte budget and carries only the records and `next_token`. It checks that the chain it names is one of the Workflow's. +* Experimental: the API protos carry `WorkflowService.ExecuteSyncNexusOperation`, which runs one + synchronous Nexus operation for a caller namespace and persists nothing. The client and the C + bridge call it, and its `input` counts against the blob size limit. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting From d1439c583ce97f5d72fef7fc87f1cdccd7f2715b Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 09:20:03 -0700 Subject: [PATCH 07/11] Keyed the `read_range` operation as `readRange` in the contract. nexgen takes an operation key only in lower camel case. The wire name stays `read_range` through `fqn`. --- .../streams/nexus/temporal_streams.nexusrpc.yaml | 2 +- crates/streams/tests/nexus_contract.rs | 16 ++++++++++++++-- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/crates/streams/nexus/temporal_streams.nexusrpc.yaml b/crates/streams/nexus/temporal_streams.nexusrpc.yaml index 99bc5e5a9a..9dcdbdd741 100644 --- a/crates/streams/nexus/temporal_streams.nexusrpc.yaml +++ b/crates/streams/nexus/temporal_streams.nexusrpc.yaml @@ -51,7 +51,7 @@ services: state-field: state input: { $ref: "#/$defs/ReadInput" } output: { $ref: "#/$defs/ReadOutput" } - read_range: + readRange: fqn: read_range description: >- Answer again the records of an earlier read that named a diff --git a/crates/streams/tests/nexus_contract.rs b/crates/streams/tests/nexus_contract.rs index 3237f4a023..4beb436aeb 100644 --- a/crates/streams/tests/nexus_contract.rs +++ b/crates/streams/tests/nexus_contract.rs @@ -102,14 +102,26 @@ fn the_contract_names_the_operations_core_serves() { ); assert_eq!(operations["read"]["input"]["$ref"], "#/$defs/ReadInput"); assert_eq!(operations["read"]["output"]["$ref"], "#/$defs/ReadOutput"); + assert_eq!(operations["readRange"]["fqn"], "read_range"); assert_eq!( - operations["read_range"]["input"]["$ref"], + operations["readRange"]["input"]["$ref"], "#/$defs/ReadRangeInput" ); assert_eq!( - operations["read_range"]["output"]["$ref"], + operations["readRange"]["output"]["$ref"], "#/$defs/ReadOutput" ); + // nexgen takes an operation key only as a lowercase letter and then letters or digits. The + // wire name comes from `fqn`. + for key in operations.as_object().unwrap().keys() { + let mut chars = key.chars(); + assert!( + chars.next().is_some_and(|c| c.is_ascii_lowercase()) + && key.len() > 1 + && chars.all(|c| c.is_ascii_alphanumeric()), + "nexgen refuses operation key {key}" + ); + } for input in ["StreamRef", "AppendInput", "ReadInput", "ReadRangeInput"] { assert_eq!( model(&contract, input)["additionalProperties"], From cd8f4cfaed6cd115ea395ac101cc29b3f8435f19 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 09:41:10 -0700 Subject: [PATCH 08/11] Told a trimmed start apart from other records in a replay read. A range that began at the stream's start has no cursor to find gone. So a trim showed as a read that ended elsewhere, a mismatch, when it is a lost range. --- crates/streams/src/handle.rs | 41 +++++++++++++++++++++------ crates/streams/tests/nexus_service.rs | 39 +++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 8 deletions(-) diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 9a1cc47685..ed22fccece 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -525,15 +525,9 @@ impl Streams { request.after ))); } - if page.cursor != request.to { - return Err(mismatch(format!( - "the replayed read ends at {:?}, not at {:?}", - page.cursor, request.to - ))); - } - if page.records_digest != request.records_digest { + if page.cursor != request.to || page.records_digest != request.records_digest { // A read from the beginning has no position to check, so a trim since the read shows - // as other records. + // as other records, which end elsewhere or differ. if request.after.is_empty() && self .store @@ -546,6 +540,12 @@ impl Streams { .to_string(), )); } + if page.cursor != request.to { + return Err(mismatch(format!( + "the replayed read ends at {:?}, not at {:?}", + page.cursor, request.to + ))); + } return Err(mismatch( "the replayed records differ from the records the read got".to_string(), )); @@ -1829,6 +1829,31 @@ mod tests { assert!(error.message.starts_with("REC-01"), "{error}"); } + #[tokio::test] + async fn a_replay_read_from_a_trimmed_start_is_rec_01() { + let setup = setup(quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup.streams.append(append("p", 2, "2")).await.unwrap(); + let request = read_after("", &[], Duration::ZERO); + let page = setup.streams.read(request.clone()).await.unwrap(); + assert_eq!(page.start_cursor, ""); + // Later records stand where the range began, so the replay finds as many as it wants. + setup.store.truncate(&chain("run-1"), "out", 0); + for sequence in 3..=4 { + setup + .streams + .append(append("p", sequence, &sequence.to_string())) + .await + .unwrap(); + } + let error = setup + .streams + .read_range(replay_of(&request, &page)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::ReplayRangeLost, "{error}"); + } + #[tokio::test] async fn a_floor_keeps_what_a_positions_only_reader_read_until_it_is_released() { let setup = setup(quick()); diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index 8296c4a4e0..4462f3914f 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -1264,3 +1264,42 @@ async fn a_range_names_a_chain_of_the_workflow_and_a_count_a_read_can_answer() { assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); } } + +/// A reader's first range starts at the stream's start, so it has no cursor to find gone. Once +/// its hold lapsed and retention trimmed the range, later records stand where it began. +#[cfg(feature = "redis")] +#[tokio::test] +async fn a_range_from_a_trimmed_start_is_lost_on_redis() { + let Ok(url) = std::env::var("STREAMS_REDIS_URL") else { + eprintln!("set STREAMS_REDIS_URL to run the Redis cases"); + return; + }; + let mut options = temporalio_streams::RedisStoreOptions::new(url); + options.key_prefix = workflow(); + options.retention = Duration::from_millis(300); + options.floor_lease = Some(Duration::from_millis(100)); + let store = Arc::new( + temporalio_streams::RedisStore::connect(options) + .await + .unwrap(), + ); + let calls = service(store, NexusOptions::default()); + let wf = workflow(); + let answer = held_read(&calls, &wf).await; + assert_eq!(answer["replay"]["after_token"], ""); + tokio::time::sleep(Duration::from_millis(600)).await; + write(&calls, &wf, 3, &["three"]).await; + write(&calls, &wf, 4, &["four"]).await; + let refused = read_range( + &calls, + replay_input(stream(&wf, TOPIC), &answer, Value::Null), + ) + .await + .unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "StreamReplayRangeLostError"), + "{}", + refused.message + ); +} From ea8c782ed11312f85696c14cd3efadfa122abd40 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 20:28:48 -0700 Subject: [PATCH 09/11] Said that AttachCaller announces only the newest record in the changelog. Core no longer arms an empty stream, since every append notifies. --- crates/sdk-core/CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 6aa59bc299..0904a9bbbb 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -140,8 +140,8 @@ relevant information. URL or request id, and a token that names no stream are `BAD_REQUEST`. An attach for the latest run describes the Workflow fresh, so it never lands on a chain that already ended. * Experimental: with `notify_on_append`, `AttachCaller` tells the stream's notifier about the - newest record after the attach, or arms an empty stream, so a caller that only waits for - progress hears of the records already there and of the first append. The API protos carry the + newest record after the attach, so a caller that only waits for progress hears of the records + already there. An empty stream needs nothing, since every append notifies. The API protos carry the `NexusHandler` progress fields and `nexus/notification-service.wit`. * Experimental: a Workflow publishes stream records with `WorkflowOutputStreamCommit` in its activation completion. The Worker appends them to its `stream_store` before it reports the From cf25c5766d46b563bc603319d1ffd54bc6557ab5 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 06:31:02 -0700 Subject: [PATCH 10/11] Checked a Nexus read's floor holder and refused one at `$end`. A holder must be //, so callers can't share or release another's hold. A positions-only read at `$end` failed its replay, which looked up the record there. --- crates/streams/src/nexus/mod.rs | 26 +++++++++++++++++++++++++ crates/streams/tests/nexus_service.rs | 28 +++++++++++++++++++++++++-- 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index ebcd839c89..8c5e2f71d9 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -298,6 +298,9 @@ impl NexusCalls { } let after = input.after_token.unwrap_or_default(); let floor_holder = input.floor_holder.unwrap_or_default(); + if let Err(failure) = check_holder(&floor_holder, &after) { + return failed(*failure); + } if input.latest_only.unwrap_or(false) { if !after.is_empty() { return failed(failure::input_mistake( @@ -865,6 +868,29 @@ fn read_failure(error: StreamError, state: Vec, positions: bool) -> NexusSte /// The stream address a call names. An owner kind this release can't open is refused before any /// codec work. +/// Checks a caller's floor holder. A Nexus caller holds as `//`, so callers in other namespaces can't share or release a holder. A positions-only read +/// can't start at `$end`, since its replay would look up the record there, which the live read +/// did not. The caller reads `latest_only` first and starts at the cursor it answers. +fn check_holder(floor_holder: &str, after: &str) -> Result<(), Box> { + if floor_holder.is_empty() { + return Ok(()); + } + let parts: Vec<&str> = floor_holder.splitn(3, '/').collect(); + if parts.len() != 3 || parts.iter().any(|part| part.is_empty()) { + return Err(Box::new(failure::input_mistake( + "a floor holder is //", + ))); + } + if after == crate::END { + return Err(Box::new(failure::input_mistake( + "a read that holds a floor starts at a cursor, not at $end; read latest_only first \ + and start at the cursor it answers", + ))); + } + Ok(()) +} + fn address(namespace: &str, stream: &StreamRef) -> Result> { if stream.kind != "workflow" { return Err(Box::new(failure::of_stream(&StreamError::unsupported( diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index cec39f4df8..3743f77c78 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -1051,15 +1051,39 @@ async fn a_record_the_reader_cannot_parse_has_an_empty_range_at_its_cursor() { assert!(replayed.records.is_empty()); } +#[tokio::test] +async fn a_floor_holder_names_the_caller_and_starts_at_a_cursor() { + let calls = service(memory(), NexusOptions::default()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + for holder in ["reader-run/1", "ns//1", "archive"] { + let mut input = read_input(&wf, ""); + input["floor_holder"] = json!(holder); + let refused = read(&calls, input).await.unwrap_err(); + assert_eq!( + kind_of(&refused), + ("BAD_REQUEST", false, "ValueError"), + "{holder}" + ); + } + let refused = read(&calls, holding(&wf, "$end")).await.unwrap_err(); + assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); + assert!( + refused.message.contains("latest_only"), + "{}", + refused.message + ); +} + #[tokio::test] async fn a_floor_holder_is_at_most_512_bytes_and_never_beside_latest_only() { let calls = service(memory(), NexusOptions::default()); let wf = workflow(); write(&calls, &wf, 1, &["one"]).await; let mut input = read_input(&wf, ""); - input["floor_holder"] = json!("h".repeat(512)); + input["floor_holder"] = json!(format!("ns/run/{}", "h".repeat(505))); read(&calls, input.clone()).await.unwrap(); - input["floor_holder"] = json!("h".repeat(513)); + input["floor_holder"] = json!(format!("ns/run/{}", "h".repeat(506))); let refused = read(&calls, input).await.unwrap_err(); assert_eq!(kind_of(&refused), ("BAD_REQUEST", false, "ValueError")); assert!(refused.message.contains("512"), "{}", refused.message); From 9ed12d35b15b3fbbcc16009e0ef1f5157798ae6d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 06:34:24 -0700 Subject: [PATCH 11/11] Checked the floor holder of a Nexus replay read too. --- crates/streams/src/nexus/mod.rs | 9 +++++++-- crates/streams/tests/nexus_service.rs | 25 +++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/crates/streams/src/nexus/mod.rs b/crates/streams/src/nexus/mod.rs index fbe1b24c39..d31e996e6d 100644 --- a/crates/streams/src/nexus/mod.rs +++ b/crates/streams/src/nexus/mod.rs @@ -462,6 +462,11 @@ impl NexusCalls { { return failed(failure::of_stream(&error)); } + if let Some(holder) = &input.floor_holder + && let Err(failure) = check_holder(holder, &input.after_token) + { + return failed(*failure); + } let request = ReadRangeRequest { stream: Some(address), chain_first_run_id: input.chain_first_run_id, @@ -471,8 +476,8 @@ impl NexusCalls { state, records_digest: input.records_digest.0, floor_holder: input.floor_holder.unwrap_or_default(), - // A read that names a floor holder never starts at the end, since only latest_only - // does, and it refuses a holder. + // A read that names a floor holder never starts at `$end`, since the read refuses it. + // The caller reads latest_only first and starts at its cursor. start_from_end: false, }; let mut response = match self.streams.read_range(request).await { diff --git a/crates/streams/tests/nexus_service.rs b/crates/streams/tests/nexus_service.rs index 1bd5a32018..837e0ebf59 100644 --- a/crates/streams/tests/nexus_service.rs +++ b/crates/streams/tests/nexus_service.rs @@ -988,6 +988,31 @@ fn base64_of(bytes: &[u8]) -> String { base64::engine::general_purpose::STANDARD.encode(bytes) } +#[tokio::test] +async fn a_holding_read_from_the_end_starts_at_latest_only_and_replays() { + let (streams, calls) = layered(memory()); + let wf = workflow(); + write(&calls, &wf, 1, &["one"]).await; + let mut latest = read_input(&wf, ""); + latest.as_object_mut().unwrap().remove("after_token"); + latest["latest_only"] = json!(true); + let end = read(&calls, latest).await.unwrap(); + append(&calls, append_input(&wf, "p", 2, 1, &["two"])) + .await + .unwrap(); + let live = read(&calls, holding(&wf, token(&end))).await.unwrap(); + assert_eq!(values(&live), ["SUPERSEDED p 1->2", "p@2 two"]); + let replayed = streams + .read_range(replay_of(&wf, &live, &[])) + .await + .unwrap(); + assert_eq!( + replayed.records.len(), + 2, + "the replay sees what the live read saw" + ); +} + #[tokio::test] async fn a_range_replays_with_the_state_the_read_sent() { let (streams, calls) = layered(memory());