From 4dcbfabda434bfac4af010c2eca24716991b23d8 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 22:57:30 -0700 Subject: [PATCH 1/2] 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 ea8c782ed11312f85696c14cd3efadfa122abd40 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 20:28:48 -0700 Subject: [PATCH 2/2] 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