From 56fd0655e48d4d7883469fb82601cbeaa0115c37 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Mon, 26 Oct 2026 22:22:53 -0700 Subject: [PATCH 01/13] Added writers for Activities and clients. A writer keeps one Producer epoch and one `sequence` order, so a retry after a Store failure writes a batch only once. The handles of one Producer epoch on one Stream share one writer. The writer suite runs on every Store. --- crates/streams/src/lib.rs | 1 + crates/streams/src/runtime.rs | 52 +- crates/streams/src/testing/mod.rs | 6 + crates/streams/src/testing/writer.rs | 683 +++++++++++++++++++++++++++ crates/streams/src/writer.rs | 397 ++++++++++++++++ 5 files changed, 1137 insertions(+), 2 deletions(-) create mode 100644 crates/streams/src/testing/writer.rs create mode 100644 crates/streams/src/writer.rs diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 2667f683b..07be4f74f 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -20,6 +20,7 @@ mod service; pub mod store; #[cfg(any(test, feature = "test-support"))] pub mod testing; +mod writer; pub use error::{Reason, StreamError, StreamResult}; pub use runtime::{MEMORY_RUNTIME_NAME, MemoryStreamRuntime, StoreFactory, StreamRuntime}; diff --git a/crates/streams/src/runtime.rs b/crates/streams/src/runtime.rs index 8b76d9c06..39070f0f6 100644 --- a/crates/streams/src/runtime.rs +++ b/crates/streams/src/runtime.rs @@ -5,11 +5,14 @@ use crate::{ memory::MemoryStore, metrics::Metrics, proto::{ - AppendRecordsRequest, CloseStreamRequest, DescribeStreamRequest, MemoryStreamConfig, - ReadRecordsRequest, + AppendRecordsRequest, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, + CloseWritersRequest, CloseWritersResponse, DescribeStreamRequest, FreeWriterRequest, + FreeWriterResponse, MemoryStreamConfig, OpenWriterRequest, OpenWriterResponse, + ReadRecordsRequest, WriteRecordsRequest, WriteRecordsResponse, }, service::StreamService, store::{DEFAULT_RETENTION, DEFAULT_SAFETY_TTL, Store}, + writer::Writers, }; use prost::Message; use std::{sync::Arc, time::Duration}; @@ -44,6 +47,7 @@ pub(crate) struct Shared { metrics: Metrics, factory: Arc, service: OnceCell>, + writers: Writers, } impl Shared { @@ -97,6 +101,7 @@ impl StreamRuntime { metrics: Metrics::new(&meter), factory, service: OnceCell::new(), + writers: Writers::default(), }); Arc::new(Self { shared }) } @@ -121,9 +126,21 @@ impl StreamRuntime { &self.shared.backend } + /// Releases one handle of a writer with no wait, as the `FreeWriter` method does. A bridge + /// calls it from a finalizer, which has no async runtime to await on. + pub fn free_writer(&self, writer_id: u64) -> StreamResult<()> { + self.shared.writers.free(writer_id) + } + /// Runs one method on encoded input and returns the encoded output. The bridge raises an error /// with [StreamError::code] and [StreamError::details]. pub async fn call(&self, method: &str, input: &[u8]) -> StreamResult> { + // `FreeWriter` needs no Store, so it works when the Store is down too. + if method == "FreeWriter" { + let request = decode::(input)?; + self.free_writer(request.writer_id)?; + return Ok(FreeWriterResponse {}.encode_to_vec()); + } let service = self.shared.service().await?; let output = match method { "CreateStream" => service.create_stream(decode(input)?).await?.encode_to_vec(), @@ -143,6 +160,37 @@ impl StreamRuntime { .describe_stream(decode::(input)?) .await? .encode_to_vec(), + "OpenWriter" => { + let request = decode::(input)?; + let writer_id = self.shared.writers.open(service, request).await?; + OpenWriterResponse { writer_id }.encode_to_vec() + } + "WriteRecords" => { + let request = decode::(input)?; + let (first_offset, last_offset) = self + .shared + .writers + .write(request.writer_id, request.records) + .await?; + WriteRecordsResponse { + first_offset, + last_offset, + } + .encode_to_vec() + } + "CloseWriters" => { + decode::(input)?; + let failed = self.shared.writers.close_all().await; + CloseWritersResponse { + failed: failed as i32, + } + .encode_to_vec() + } + "CloseWriter" => { + let request = decode::(input)?; + let last_offset = self.shared.writers.close(request.writer_id).await?; + CloseWriterResponse { last_offset }.encode_to_vec() + } _ => { return Err(StreamError::new( Reason::Unsupported, diff --git a/crates/streams/src/testing/mod.rs b/crates/streams/src/testing/mod.rs index 77ede4730..3b4fdef8f 100644 --- a/crates/streams/src/testing/mod.rs +++ b/crates/streams/src/testing/mod.rs @@ -5,6 +5,7 @@ mod support; pub mod service; pub mod store; +pub mod writer; pub use support::Needs; @@ -152,4 +153,9 @@ mod tests { async fn the_service_suite_passes_on_the_memory_store() { run(vec![Arc::new(MemoryBackend)], service::CASES).await; } + + #[tokio::test] + async fn the_writer_suite_passes_on_the_memory_store() { + run(vec![Arc::new(MemoryBackend)], writer::CASES).await; + } } diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs new file mode 100644 index 000000000..61359412b --- /dev/null +++ b/crates/streams/src/testing/writer.rs @@ -0,0 +1,683 @@ +//! The writers of Activities and clients, through the runtime's byte call, as lang sends them. +//! +//! Every Store runs the suite through its own runtime. + +use super::{ + case, + support::{Needs, stores_for}, +}; +use crate::{ + MemoryStreamRuntime, StreamError, StreamResult, StreamRuntime, + epoch::withdraws, + ids::{ + activity_producer_id, client_producer_id, local_activity_producer_id, + named_client_producer_id, + }, + proto::{ + AppendRecord, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, + CreateStreamRequest, CreateStreamResponse, OpenWriterRequest, OpenWriterResponse, Producer, + ProducerEpoch, ReadRecordsRequest, ReadRecordsResponse, StreamErrorDetails, + StreamProducerKind, StreamRecord, StreamRef, WorkflowTaskEpoch, WriteRecordsRequest, + WriteRecordsResponse, stream_record::Item, + }, + store::{ + Appended, Batch, Capabilities, Closed, Created, Entry, Owner, Store, StreamState, + Withdrawal, + }, +}; +use prost::Message; +use prost_types::Timestamp; +use std::{ + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; +use temporalio_common::protos::temporal::api::common::v1::Payload; +use tonic::Code; + +static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + +struct Case { + ext: Arc, + namespace: String, + stream: String, +} + +async fn cases() -> Vec { + let mut cases = vec![]; + for ext in super::support::stream_runtimes() { + let n = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let case = Case { + ext, + namespace: format!("ns-{now}-{n}"), + stream: format!("s-{now}-{n}"), + }; + let created: CreateStreamResponse = case + .call( + "CreateStream", + CreateStreamRequest { + namespace: case.namespace.clone(), + stream_id: case.stream.clone(), + ..Default::default() + }, + ) + .await + .unwrap(); + assert!(created.created); + cases.push(case); + } + cases +} + +fn client() -> Producer { + Producer { + kind: StreamProducerKind::Client as i32, + id: client_producer_id("me", "p"), + epoch: Some(ProducerEpoch { + attempt: 1, + ..Default::default() + }), + run_id: String::new(), + } +} + +fn activity(attempt: i32) -> Producer { + Producer { + kind: StreamProducerKind::Activity as i32, + id: activity_producer_id("r1", "a", "s"), + epoch: Some(ProducerEpoch { + attempt_scheduled_time: Some(Timestamp { + seconds: 10 * attempt as i64, + nanos: 0, + }), + attempt, + ..Default::default() + }), + run_id: "r1".to_string(), + } +} + +fn records(texts: &[&str]) -> Vec { + texts + .iter() + .map(|text| AppendRecord { + body: Some(Payload { + data: text.as_bytes().to_vec(), + ..Default::default() + }), + }) + .collect() +} + +fn reason(error: &StreamError) -> String { + let details = StreamErrorDetails::decode(error.details().as_slice()).unwrap(); + assert_eq!(details.reason(), error.reason.proto()); + error.reason.token().to_string() +} + +fn millis(ms: u64) -> Option { + Some(prost_types::Duration { + seconds: (ms / 1000) as i64, + nanos: ((ms % 1000) * 1_000_000) as i32, + }) +} + +impl Case { + async fn call( + &self, + method: &str, + request: Req, + ) -> Result { + let bytes = self.ext.call(method, &request.encode_to_vec()).await?; + Ok(Resp::decode(bytes.as_slice()).unwrap()) + } + + async fn open(&self, producer: Producer, change: impl Fn(&mut OpenWriterRequest)) -> u64 { + let mut request = OpenWriterRequest { + namespace: self.namespace.clone(), + stream: Some(StreamRef { + id: self.stream.clone(), + ..Default::default() + }), + producer: Some(producer), + ..Default::default() + }; + change(&mut request); + let opened: OpenWriterResponse = self.call("OpenWriter", request).await.unwrap(); + opened.writer_id + } + + /// An awaited publish. The Records are in the Store when it returns. + async fn write(&self, writer_id: u64, texts: &[&str]) -> Result<(i64, i64), StreamError> { + let written: WriteRecordsResponse = self + .call( + "WriteRecords", + WriteRecordsRequest { + writer_id, + records: records(texts), + }, + ) + .await?; + Ok((written.first_offset, written.last_offset)) + } + + async fn read(&self, wait_ms: u64) -> ReadRecordsResponse { + self.call( + "ReadRecords", + ReadRecordsRequest { + namespace: self.namespace.clone(), + stream: Some(StreamRef { + id: self.stream.clone(), + ..Default::default() + }), + wait: millis(wait_ms), + ..Default::default() + }, + ) + .await + .unwrap() + } + + async fn visible(&self) -> Vec { + let mut kept: Vec = vec![]; + for record in self.read(0).await.records { + match &record.item { + Some(Item::Superseded(s)) => kept.retain(|old| !withdraws(s, old)), + Some(Item::Data(_)) => kept.push(record), + _ => {} + } + } + kept.into_iter() + .filter_map(|record| match record.item { + Some(Item::Data(data)) => Some(String::from_utf8(data.body.unwrap().data).unwrap()), + _ => None, + }) + .collect() + } +} + +async fn a_write_is_in_the_store_when_it_answers_its_offsets() { + for case in cases().await { + let writer = case.open(client(), |_| {}).await; + let (first, last) = case.write(writer, &["a", "b"]).await.unwrap(); + assert!(0 < first && first < last); + let read = case.read(0).await; + let offsets: Vec = read.records.iter().map(|r| r.offset).collect(); + assert_eq!(offsets, [first, last]); + let (one, same) = case.write(writer, &["c"]).await.unwrap(); + assert_eq!(one, same); + assert!(one > last); + assert_eq!(case.write(writer, &[]).await.unwrap(), (0, 0)); + assert_eq!(case.visible().await, ["a", "b", "c"]); + let closed: CloseWriterResponse = case + .call("CloseWriter", CloseWriterRequest { writer_id: writer }) + .await + .unwrap(); + assert_eq!(closed.last_offset, one); + } +} + +async fn writes_split_at_the_batch_size_and_keep_one_sequence() { + for case in cases().await { + let writer = case.open(client(), |r| r.max_batch_size = 2).await; + let (first, last) = case.write(writer, &["1", "2", "3"]).await.unwrap(); + case.write(writer, &["4", "5"]).await.unwrap(); + let read = case.read(0).await; + assert_eq!(case.visible().await, ["1", "2", "3", "4", "5"]); + assert_eq!(read.records[0].offset, first); + assert_eq!(read.records[2].offset, last); + // Opening the same Producer epoch again returns the same writer, so it keeps the sequence. + assert_eq!(case.open(client(), |_| {}).await, writer); + } +} + +async fn a_failed_write_fails_every_later_write_and_the_close() { + for case in cases().await { + let writer = case.open(client(), |_| {}).await; + let close = CloseStreamRequest { + namespace: case.namespace.clone(), + stream: Some(StreamRef { + id: case.stream.clone(), + ..Default::default() + }), + producer: Some(client()), + ..Default::default() + }; + let _: crate::proto::CloseStreamResponse = case.call("CloseStream", close).await.unwrap(); + let error = case.write(writer, &["late"]).await.unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::FailedPrecondition, "stream_closed") + ); + assert_eq!( + reason(&case.write(writer, &["x"]).await.unwrap_err()), + "stream_closed" + ); + let error = case + .call::<_, CloseWriterResponse>("CloseWriter", CloseWriterRequest { writer_id: writer }) + .await + .unwrap_err(); + assert_eq!(reason(&error), "stream_closed"); + } +} + +async fn an_activity_retry_withdraws_the_earlier_attempt_even_with_nothing_to_write() { + for case in cases().await { + let first = case.open(activity(1), |_| {}).await; + case.write(first, &["a1"]).await.unwrap(); + case.open(activity(2), |_| {}).await; + assert!(case.visible().await.is_empty()); + } +} + +/// An Activity retry opens its writer on a closed or expired Stream, so a retry that only reads +/// the Stream still runs. Its first write fails with the real reason. +async fn an_activity_retry_opens_on_a_closed_or_expired_stream() { + for case in cases().await { + let first = case.open(activity(1), |_| {}).await; + case.write(first, &["a1"]).await.unwrap(); + let close = CloseStreamRequest { + namespace: case.namespace.clone(), + stream: Some(StreamRef { + id: case.stream.clone(), + ..Default::default() + }), + producer: Some(client()), + ..Default::default() + }; + let _: crate::proto::CloseStreamResponse = case.call("CloseStream", close).await.unwrap(); + let retry = case.open(activity(2), |_| {}).await; + let error = case.write(retry, &["a2"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_closed"); + + let short = format!("{}-short", case.stream); + let _: CreateStreamResponse = case + .call( + "CreateStream", + CreateStreamRequest { + namespace: case.namespace.clone(), + stream_id: short.clone(), + retention: millis(300), + ..Default::default() + }, + ) + .await + .unwrap(); + let on_short = |request: &mut OpenWriterRequest| { + request.stream = Some(StreamRef { + id: short.clone(), + ..Default::default() + }); + }; + let first = case.open(activity(1), on_short).await; + case.write(first, &["a1"]).await.unwrap(); + tokio::time::sleep(Duration::from_millis(500)).await; + let retry = case.open(activity(2), on_short).await; + let error = case.write(retry, &["a2"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_expired"); + } +} + +async fn a_closed_writer_is_gone() { + for case in cases().await { + let writer = case.open(client(), |_| {}).await; + case.write(writer, &["last"]).await.unwrap(); + let closed: CloseWriterResponse = case + .call("CloseWriter", CloseWriterRequest { writer_id: writer }) + .await + .unwrap(); + assert!(closed.last_offset > 0); + let error = case.write(writer, &["x"]).await.unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::NotFound, "stream_unknown_writer") + ); + } +} + +async fn a_write_splits_big_records_by_bytes_and_refuses_an_oversize_one() { + for case in cases().await { + let writer = case.open(client(), |_| {}).await; + let big = "x".repeat(1_400_000); + case.write(writer, &[&big, &big, &big, "end"]) + .await + .unwrap(); + assert_eq!(case.visible().await.len(), 4); + let huge = "y".repeat(1_700_000); + let error = case.write(writer, &[&huge]).await.unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::InvalidArgument, "stream_too_large") + ); + // A refusal before the Store leaves the writer usable. + case.write(writer, &["ok"]).await.unwrap(); + } +} + +async fn a_reopen_of_a_closed_writer_goes_on_with_its_sequence() { + for case in cases().await { + let writer = case.open(client(), |_| {}).await; + case.write(writer, &["1"]).await.unwrap(); + let _: CloseWriterResponse = case + .call("CloseWriter", CloseWriterRequest { writer_id: writer }) + .await + .unwrap(); + let again = case.open(client(), |_| {}).await; + assert_ne!(again, writer); + case.write(again, &["2"]).await.unwrap(); + assert_eq!(case.visible().await, ["1", "2"]); + } +} + +async fn the_handles_of_one_producer_share_a_writer_until_the_last_one_lets_go() { + for case in cases().await { + let a = case.open(client(), |_| {}).await; + let b = case.open(client(), |_| {}).await; + assert_eq!(a, b); + case.write(a, &["1"]).await.unwrap(); + let _: CloseWriterResponse = case + .call("CloseWriter", CloseWriterRequest { writer_id: a }) + .await + .unwrap(); + // The other handle still writes, on the same sequence. + case.write(b, &["2"]).await.unwrap(); + case.ext.free_writer(b).unwrap(); + let error = case.write(b, &["3"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_unknown_writer"); + assert_eq!(case.visible().await, ["1", "2"]); + } +} + +async fn a_named_client_with_a_higher_epoch_fences_its_earlier_self() { + for case in cases().await { + let named = |attempt| Producer { + kind: StreamProducerKind::Client as i32, + id: named_client_producer_id("ingest-1"), + epoch: Some(ProducerEpoch { + attempt, + ..Default::default() + }), + run_id: String::new(), + }; + let old = case.open(named(1), |_| {}).await; + case.write(old, &["old"]).await.unwrap(); + let new = case.open(named(2), |_| {}).await; + case.write(new, &["new"]).await.unwrap(); + let error = case.write(old, &["late"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_fenced"); + // The earlier epoch's Records are real messages, so the new epoch withdraws none. + assert_eq!(case.visible().await, ["old", "new"]); + } +} + +async fn closing_every_writer_drops_them_all() { + for case in cases().await { + let a = case.open(client(), |_| {}).await; + let b = case.open(activity(1), |_| {}).await; + case.write(a, &["a"]).await.unwrap(); + case.write(b, &["b"]).await.unwrap(); + let closed: crate::proto::CloseWritersResponse = case + .call("CloseWriters", crate::proto::CloseWritersRequest {}) + .await + .unwrap(); + assert_eq!(closed.failed, 0); + assert!(case.write(a, &["x"]).await.is_err()); + assert!(case.write(b, &["x"]).await.is_err()); + } +} + +/// A Store over another one, whose appends fail. It can write an append and lose its answer, as on +/// a dropped connection, and it refuses every append while it is down. +struct Flaky { + inner: Arc, + lose_next: AtomicBool, + down: AtomicBool, +} + +impl Flaky { + /// A Store that is up, over `inner`. + fn new(inner: Arc) -> Self { + Self { + inner, + lose_next: AtomicBool::new(false), + down: AtomicBool::new(false), + } + } +} + +#[async_trait::async_trait] +impl Store for Flaky { + fn capabilities(&self) -> Capabilities { + self.inner.capabilities() + } + + fn default_retention(&self) -> Duration { + self.inner.default_retention() + } + + async fn create( + &self, + namespace: &str, + stream: &StreamRef, + retention: Option, + owner: Owner, + withdrawal: &Withdrawal<'_>, + ) -> StreamResult { + self.inner + .create(namespace, stream, retention, owner, withdrawal) + .await + } + + async fn append( + &self, + namespace: &str, + stream: &StreamRef, + batch: &Batch<'_>, + ) -> StreamResult { + if self.down.load(Ordering::SeqCst) { + return Err(StreamError::storage("the Store is down")); + } + let appended = self.inner.append(namespace, stream, batch).await?; + if self.lose_next.swap(false, Ordering::SeqCst) { + return Err(StreamError::storage("the answer was lost")); + } + Ok(appended) + } + + async fn read( + &self, + namespace: &str, + stream_id: &str, + from: i64, + count: usize, + ) -> StreamResult<(StreamState, Vec)> { + self.inner.read(namespace, stream_id, from, count).await + } + + async fn wait_after( + &self, + namespace: &str, + stream_id: &str, + after: i64, + deadline: tokio::time::Instant, + ) -> StreamResult { + self.inner + .wait_after(namespace, stream_id, after, deadline) + .await + } + + async fn close( + &self, + namespace: &str, + stream: &StreamRef, + producer: &Producer, + record: &[u8], + ) -> StreamResult { + self.inner.close(namespace, stream, producer, record).await + } + + async fn read_back( + &self, + namespace: &str, + stream_id: &str, + before: i64, + count: usize, + ) -> StreamResult> { + self.inner + .read_back(namespace, stream_id, before, count) + .await + } +} + +/// A case on each backend under test, on a [Flaky] Store over the backend's Store, with its Stream +/// created. +async fn flaky_cases(lose_next: bool) -> Vec<(Arc, Case)> { + let mut cases = vec![]; + for target in stores_for(Needs::CALLS).await { + let store = Arc::new(Flaky { + lose_next: AtomicBool::new(lose_next), + ..Flaky::new(target.store) + }); + let case = Case { + ext: MemoryStreamRuntime::with_store(store.clone(), None), + namespace: target.namespace, + stream: target.stream, + }; + let _: CreateStreamResponse = case + .call( + "CreateStream", + CreateStreamRequest { + namespace: case.namespace.clone(), + stream_id: case.stream.clone(), + ..Default::default() + }, + ) + .await + .unwrap(); + cases.push((store, case)); + } + cases +} + +async fn a_write_whose_answer_is_lost_goes_in_once() { + for (_, case) in flaky_cases(true).await { + let writer = case.open(client(), |_| {}).await; + let (first, last) = case.write(writer, &["a", "b"]).await.unwrap(); + let read = case.read(0).await; + assert_eq!(case.visible().await, ["a", "b"]); + assert_eq!( + (read.records[0].offset, read.records[1].offset), + (first, last) + ); + case.write(writer, &["c"]).await.unwrap(); + assert_eq!(case.visible().await, ["a", "b", "c"]); + } +} + +/// A write that fails in a Store outage fails alone, and the next write succeeds once the Store is +/// back. +async fn a_writer_writes_again_after_a_store_outage() { + for (store, case) in flaky_cases(false).await { + let writer = case + .open(client(), |r| r.max_retry_duration = millis(100)) + .await; + case.write(writer, &["a"]).await.unwrap(); + store.down.store(true, Ordering::SeqCst); + let error = case.write(writer, &["lost"]).await.unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::Unavailable, "stream_storage") + ); + store.down.store(false, Ordering::SeqCst); + case.write(writer, &["b"]).await.unwrap(); + assert_eq!(case.visible().await, ["a", "b"]); + let closed: CloseWriterResponse = case + .call("CloseWriter", CloseWriterRequest { writer_id: writer }) + .await + .unwrap(); + assert!(closed.last_offset > 0); + } +} + +async fn a_write_that_waits_behind_a_close_is_refused() { + for case in cases().await { + let writer = case.open(client(), |_| {}).await; + case.write(writer, &["a"]).await.unwrap(); + let close = case.call::<_, CloseWriterResponse>( + "CloseWriter", + CloseWriterRequest { writer_id: writer }, + ); + let write = async { + tokio::task::yield_now().await; + case.write(writer, &["late"]).await + }; + let (closed, written) = tokio::join!(close, write); + closed.unwrap(); + assert_eq!(reason(&written.unwrap_err()), "stream_unknown_writer"); + assert_eq!(case.visible().await, ["a"]); + } +} + +async fn a_local_activity_writer_withdraws_the_records_of_a_failed_task() { + for case in cases().await { + let local_activity = |call: &str, started: i64| Producer { + kind: StreamProducerKind::LocalActivity as i32, + id: local_activity_producer_id("r1", call), + epoch: Some(ProducerEpoch { + task_started_event_id: started, + task_attempt: 1, + task_started_time: Some(Timestamp { + seconds: started, + nanos: 0, + }), + attempt: 1, + ..Default::default() + }), + run_id: "r1".to_string(), + }; + let facts = |completed: i64, failed: Vec| { + move |request: &mut OpenWriterRequest| { + request.previous_completed_task = Some(WorkflowTaskEpoch { + started_event_id: completed, + ..Default::default() + }); + request.failed_tasks = failed.clone(); + } + }; + // Task [3] fails after its Local Activity writes, and task [4] completes with no write. + let orphan = case.open(local_activity("1", 3), facts(2, vec![])).await; + case.write(orphan, &["orphan"]).await.unwrap(); + let next = case.open(local_activity("2", 5), facts(4, vec![3])).await; + case.write(next, &["next"]).await.unwrap(); + assert_eq!(case.visible().await, ["next"]); + // Task [5] completed while call 2 ran, and task [6], where the call resolved, failed. + // Call 2 runs again in task [8] and writes nothing. Its open withdraws its earlier task. + case.open(local_activity("2", 8), facts(5, vec![6])).await; + assert!(case.visible().await.is_empty()); + } +} + +/// The cases of the suite, in file order. +pub const CASES: &[super::Case] = &[ + case!(a_write_is_in_the_store_when_it_answers_its_offsets), + case!(writes_split_at_the_batch_size_and_keep_one_sequence), + case!(a_failed_write_fails_every_later_write_and_the_close), + case!(an_activity_retry_withdraws_the_earlier_attempt_even_with_nothing_to_write), + case!(an_activity_retry_opens_on_a_closed_or_expired_stream), + case!(a_closed_writer_is_gone), + case!(a_write_splits_big_records_by_bytes_and_refuses_an_oversize_one), + case!(a_reopen_of_a_closed_writer_goes_on_with_its_sequence), + case!(the_handles_of_one_producer_share_a_writer_until_the_last_one_lets_go), + case!(a_named_client_with_a_higher_epoch_fences_its_earlier_self), + case!(closing_every_writer_drops_them_all), + case!(a_write_whose_answer_is_lost_goes_in_once), + case!(a_writer_writes_again_after_a_store_outage), + case!(a_write_that_waits_behind_a_close_is_refused), + case!(a_local_activity_writer_withdraws_the_records_of_a_failed_task), +]; diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs new file mode 100644 index 000000000..d0af10584 --- /dev/null +++ b/crates/streams/src/writer.rs @@ -0,0 +1,397 @@ +//! Writers for Activities and clients. `write` appends its Records at once and returns their +//! offsets, as an awaited publish needs. A writer keeps one Producer epoch and one `sequence` +//! order, so that the Store writes a retried batch only once. Lang can buffer Records on its own +//! and write each batch with one call. +//! +//! The handles of one Producer epoch on one Stream share one writer. Each open adds a handle and +//! each close or free releases one. The last release drops the writer. + +use crate::{ + Reason, StreamError, StreamResult, + proto::{ + AppendRecord, AppendRecordsRequest, OpenWriterRequest, Producer, StreamProducerKind, + StreamRef, WorkflowTaskEpoch, + }, + service::StreamService, + store::{MAX_BATCH_BYTES, MAX_RECORD_BYTES}, +}; +use parking_lot::Mutex; +use prost::Message; +use std::{ + collections::HashMap, + sync::{ + Arc, + atomic::{AtomicI64, AtomicU64, Ordering}, + }, + time::Duration, +}; +use tokio::time::Instant; + +const DEFAULT_BATCH: usize = 1000; +const DEFAULT_RETRY: Duration = Duration::from_secs(10); +const FIRST_BACKOFF: Duration = Duration::from_millis(50); + +/// The last `sequence` of any batch in this process. Every writer numbers its batches from it, so +/// that a writer that opens again after its close continues above its earlier batches. +static LAST_SEQUENCE: AtomicI64 = AtomicI64::new(0); + +fn next_sequence() -> i64 { + LAST_SEQUENCE.fetch_add(1, Ordering::Relaxed) + 1 +} + +#[derive(Default)] +struct State { + last_offset: i64, + /// A write that the Store refused, such as on a closed Stream or a newer epoch. Every later + /// call fails with it. A Store failure fails only its own write. + failed: Option, + closed: bool, +} + +struct Writer { + namespace: String, + stream: StreamRef, + producer: Producer, + previous_completed_task: Option, + failed_tasks: Vec, + max_batch: usize, + max_retry: Duration, + service: Arc, + state: Mutex, + /// Allows one write at a time, so that the batches of one epoch reach the Store in order. + writing: tokio::sync::Mutex<()>, +} + +fn duration(value: Option) -> Option { + let value = value?; + let duration = Duration::new(value.seconds.max(0) as u64, value.nanos.max(0) as u32); + (!duration.is_zero()).then_some(duration) +} + +impl Writer { + /// The number of `records` in the next batch. The Store refuses a batch over its limit, so a + /// batch ends at the count limit or the byte limit, whichever comes first. + fn batch_len<'a>(&self, records: impl Iterator) -> usize { + let mut count = 0; + let mut bytes = 0; + for record in records.take(self.max_batch) { + let size = stored_size(&self.producer, record); + if count > 0 && bytes + size > MAX_BATCH_BYTES { + break; + } + bytes += size; + count += 1; + } + count + } + + /// Appends `records` now and returns the offsets of the first and the last one. Records past + /// the Store's batch limit go in more batches, in order. + async fn write(&self, mut records: Vec) -> StreamResult<(i64, i64)> { + let _writing = self.writing.lock().await; + // A close that started while this write waited has ended the writer. + if self.state.lock().closed { + return Err(StreamError::new( + Reason::UnknownWriter, + "The writer closed while the write waited", + )); + } + let (mut first, mut last) = (0, 0); + while !records.is_empty() { + let count = self.batch_len(records.iter()); + let rest = records.split_off(count); + if let Some(error) = &self.state.lock().failed { + return Err(error.clone()); + } + let sequence = next_sequence(); + let (batch_first, batch_last) = match self.send(records, sequence).await { + Ok(offsets) => offsets, + // The next write takes a higher `sequence`, so the Store refuses a late copy of + // this batch after it. + Err(error) if error.reason == Reason::Storage => return Err(error), + Err(error) => { + self.state.lock().failed = Some(error.clone()); + return Err(error); + } + }; + if first == 0 { + first = batch_first; + } + last = batch_last; + records = rest; + } + let mut state = self.state.lock(); + state.last_offset = state.last_offset.max(last); + Ok((first, last)) + } + + /// Appends one batch and returns its first and last offset. After a Store failure it sends the + /// batch again with the same `sequence`, so the dedupe key stops a second write. + async fn send(&self, records: Vec, sequence: i64) -> StreamResult<(i64, i64)> { + let request = AppendRecordsRequest { + namespace: self.namespace.clone(), + stream: Some(self.stream.clone()), + producer: Some(self.producer.clone()), + sequence, + records, + previous_completed_task: self.previous_completed_task, + failed_tasks: self.failed_tasks.clone(), + ..Default::default() + }; + let until = Instant::now() + self.max_retry; + let mut backoff = FIRST_BACKOFF; + loop { + match self.service.append_records(request.clone()).await { + Ok(response) => return Ok((response.first_offset, response.last_offset)), + Err(error) + if error.reason == Reason::Storage && Instant::now() + backoff < until => + { + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(Duration::from_secs(1)); + } + Err(error) => return Err(error), + } + } + } +} + +/// The size of a Record once stored, with a margin for the envelope's own fields. +fn stored_size(producer: &Producer, record: &AppendRecord) -> usize { + record.encoded_len() + producer.encoded_len() + 32 +} + +/// Refuses a Record over the Store's Record limit, before it reaches the Store. +fn check_sizes(producer: &Producer, records: &[AppendRecord]) -> StreamResult<()> { + if let Some(big) = records + .iter() + .map(|record| stored_size(producer, record)) + .find(|size| *size > MAX_RECORD_BYTES) + { + return Err(StreamError::new( + Reason::TooLarge, + format!("A Record has about {big} bytes, more than {MAX_RECORD_BYTES}"), + )); + } + Ok(()) +} + +/// The namespace, Stream id, Producer id and epoch key of a writer. +type WriterKey = (String, String, String, String); + +/// One open writer and the handles that share it. +struct Open { + writer: Arc, + key: WriterKey, + handles: usize, +} + +#[derive(Default)] +struct Table { + open: HashMap, + /// The writer of each key, so a second open shares its sequence instead of reusing it. + by_producer: HashMap, +} + +impl Table { + /// Releases one handle of a writer, or every handle with `all`. Returns the writer when it has + /// no handle left, so that the caller ends it. + fn release(&mut self, id: u64, all: bool) -> StreamResult>> { + let open = self + .open + .get_mut(&id) + .ok_or_else(|| StreamError::new(Reason::UnknownWriter, format!("No writer {id}")))?; + open.handles -= 1; + if open.handles > 0 && !all { + return Ok(None); + } + let open = self.open.remove(&id).expect("found above"); + self.by_producer.remove(&open.key); + Ok(Some(open.writer)) + } +} + +/// The open writers of one runtime. +#[derive(Default)] +pub(crate) struct Writers { + next: AtomicU64, + table: Mutex, +} + +impl Writers { + /// Opens a writer, or adds a handle to the one that this Producer epoch has on the Stream. A + /// writer of a Local Activity, or of an Activity attempt above 1, first sends an empty batch. + /// So the new epoch withdraws the earlier one's Records even when it publishes nothing. On a + /// closed or expired Stream the open still succeeds, and the first write fails with the reason. + pub(crate) async fn open( + &self, + service: &Arc, + request: OpenWriterRequest, + ) -> StreamResult { + let stream = request + .stream + .ok_or_else(|| StreamError::invalid("The writer has no Stream"))?; + crate::ids::check_stream_id(&stream.id)?; + let producer = request + .producer + .ok_or_else(|| StreamError::invalid("The writer has no Producer"))?; + crate::ids::check_producer(&producer)?; + if producer.kind == StreamProducerKind::Workflow as i32 { + return Err(StreamError::invalid("Workflow code doesn't open a writer")); + } + let epoch = producer.epoch.unwrap_or_default(); + let key = ( + request.namespace.clone(), + stream.id.clone(), + producer.id.clone(), + crate::epoch::epoch_key(&epoch), + ); + if let Some(id) = self.share(&key) { + return Ok(id); + } + let writer = Arc::new(Writer { + namespace: request.namespace, + stream, + producer, + previous_completed_task: request.previous_completed_task, + failed_tasks: request.failed_tasks, + max_batch: match request.max_batch_size { + n if n > 0 => n as usize, + _ => DEFAULT_BATCH, + }, + max_retry: duration(request.max_retry_duration).unwrap_or(DEFAULT_RETRY), + service: service.clone(), + state: Mutex::default(), + writing: tokio::sync::Mutex::new(()), + }); + // A Local Activity that runs again in a later task has attempt 1 again, so its writer + // always sends the empty batch. + let retry = match StreamProducerKind::try_from(writer.producer.kind).unwrap_or_default() { + StreamProducerKind::LocalActivity => true, + StreamProducerKind::Activity | StreamProducerKind::StandaloneActivity => { + epoch.attempt > 1 + } + _ => false, + }; + if retry { + match writer.send(vec![], next_sequence()).await { + // A Stream with no key has nothing to withdraw yet. A closed or expired one accepts + // no more Records, and a retry that only reads it must still run. + Ok(_) => {} + Err(error) + if matches!( + error.reason, + Reason::NotFound | Reason::Closed | Reason::Expired + ) => {} + Err(error) => return Err(error), + } + } + let mut table = self.table.lock(); + if let Some(id) = table.by_producer.get(&key).copied() { + table.open.get_mut(&id).expect("indexed").handles += 1; + return Ok(id); + } + let id = self.next.fetch_add(1, Ordering::Relaxed) + 1; + table.by_producer.insert(key.clone(), id); + table.open.insert( + id, + Open { + writer, + key, + handles: 1, + }, + ); + Ok(id) + } + + /// Adds a handle to the open writer of `key`, if there is one. + fn share(&self, key: &WriterKey) -> Option { + let mut table = self.table.lock(); + let id = table.by_producer.get(key).copied()?; + table.open.get_mut(&id).expect("indexed").handles += 1; + Some(id) + } + + fn get(&self, id: u64) -> StreamResult> { + self.table + .lock() + .open + .get(&id) + .map(|open| open.writer.clone()) + .ok_or_else(|| StreamError::new(Reason::UnknownWriter, format!("No writer {id}"))) + } + + /// Appends Records at once and returns the first and the last offset. A retry after a Store + /// failure resends the same batch, so that the Store writes it once. + pub(crate) async fn write( + &self, + id: u64, + records: Vec, + ) -> StreamResult<(i64, i64)> { + let writer = self.get(id)?; + { + let state = writer.state.lock(); + if let Some(error) = &state.failed { + return Err(error.clone()); + } + if state.closed { + return Err(StreamError::new( + Reason::UnknownWriter, + format!("Writer {id} is closed"), + )); + } + } + check_sizes(&writer.producer, &records)?; + writer.write(records).await + } + + /// Closes every open writer, whatever its handles. Returns how many had failed. + pub(crate) async fn close_all(&self) -> usize { + let ids: Vec = self.table.lock().open.keys().copied().collect(); + let mut failed = 0; + for id in ids { + let released = self.table.lock().release(id, true); + let Ok(Some(writer)) = released else { + continue; + }; + if end(&writer).await.is_err() { + failed += 1; + } + } + failed + } + + /// Releases one handle. The last one waits for a write in flight, then drops the writer. + /// Returns its last offset, or the error it failed with. + pub(crate) async fn close(&self, id: u64) -> StreamResult { + let writer = self.get(id)?; + if self.table.lock().release(id, false)?.is_none() { + let state = writer.state.lock(); + return match &state.failed { + Some(error) => Err(error.clone()), + None => Ok(state.last_offset), + }; + } + end(&writer).await + } + + /// Releases one handle with no wait, for a handle that lang drops. A write in flight still + /// completes, and a write that waits behind it is refused once no handle is left. + pub(crate) fn free(&self, id: u64) -> StreamResult<()> { + if let Some(writer) = self.table.lock().release(id, false)? { + writer.state.lock().closed = true; + } + Ok(()) + } +} + +/// Ends a writer that no handle uses. It waits for a write in flight, then returns its last offset +/// or the error it failed with. +async fn end(writer: &Writer) -> StreamResult { + writer.state.lock().closed = true; + let _writing = writer.writing.lock().await; + let state = writer.state.lock(); + match &state.failed { + Some(error) => Err(error.clone()), + None => Ok(state.last_offset), + } +} From 1085709de68a26faa58f3b40336433e372e54342 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 1 Nov 2026 03:01:53 -0700 Subject: [PATCH 02/13] Renamed StreamRef to StreamHandle in the writer and its suite. --- crates/streams/src/testing/writer.rs | 26 +++++++++++++------------- crates/streams/src/writer.rs | 12 ++++++------ 2 files changed, 19 insertions(+), 19 deletions(-) diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs index 86c55edb2..662d7114e 100644 --- a/crates/streams/src/testing/writer.rs +++ b/crates/streams/src/testing/writer.rs @@ -16,8 +16,8 @@ use crate::{ proto::{ AppendRecord, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, CreateStreamRequest, CreateStreamResponse, OpenWriterRequest, OpenWriterResponse, Producer, - ProducerEpoch, ReadRecordsRequest, ReadRecordsResponse, StreamErrorDetails, - StreamProducerKind, StreamRecord, StreamRef, WorkflowTaskEpoch, WriteRecordsRequest, + ProducerEpoch, ReadRecordsRequest, ReadRecordsResponse, StreamErrorDetails, StreamHandle, + StreamProducerKind, StreamRecord, WorkflowTaskEpoch, WriteRecordsRequest, WriteRecordsResponse, stream_record::Item, }, store::{ @@ -141,7 +141,7 @@ impl Case { async fn open(&self, producer: Producer, change: impl Fn(&mut OpenWriterRequest)) -> u64 { let mut request = OpenWriterRequest { namespace: self.namespace.clone(), - stream: Some(StreamRef { + handle: Some(StreamHandle { id: self.stream.clone(), ..Default::default() }), @@ -172,7 +172,7 @@ impl Case { "ReadRecords", ReadRecordsRequest { namespace: self.namespace.clone(), - stream: Some(StreamRef { + handle: Some(StreamHandle { id: self.stream.clone(), ..Default::default() }), @@ -242,7 +242,7 @@ async fn a_failed_write_fails_every_later_write_and_the_close() { let writer = case.open(client(), |_| {}).await; let close = CloseStreamRequest { namespace: case.namespace.clone(), - stream: Some(StreamRef { + handle: Some(StreamHandle { id: case.stream.clone(), ..Default::default() }), @@ -284,7 +284,7 @@ async fn an_activity_retry_opens_on_a_closed_or_expired_stream() { case.write(first, &["a1"]).await.unwrap(); let close = CloseStreamRequest { namespace: case.namespace.clone(), - stream: Some(StreamRef { + handle: Some(StreamHandle { id: case.stream.clone(), ..Default::default() }), @@ -310,7 +310,7 @@ async fn an_activity_retry_opens_on_a_closed_or_expired_stream() { .await .unwrap(); let on_short = |request: &mut OpenWriterRequest| { - request.stream = Some(StreamRef { + request.handle = Some(StreamHandle { id: short.clone(), ..Default::default() }); @@ -460,26 +460,26 @@ impl Store for Flaky { async fn create( &self, namespace: &str, - stream: &StreamRef, + handle: &StreamHandle, retention: Option, owner: Owner, withdrawal: &Withdrawal<'_>, ) -> StreamResult { self.inner - .create(namespace, stream, retention, owner, withdrawal) + .create(namespace, handle, retention, owner, withdrawal) .await } async fn append( &self, namespace: &str, - stream: &StreamRef, + handle: &StreamHandle, batch: &Batch<'_>, ) -> StreamResult { if self.down.load(Ordering::SeqCst) { return Err(StreamError::storage("the Store is down")); } - let appended = self.inner.append(namespace, stream, batch).await?; + let appended = self.inner.append(namespace, handle, batch).await?; if self.lose_next.swap(false, Ordering::SeqCst) { return Err(StreamError::storage("the answer was lost")); } @@ -511,11 +511,11 @@ impl Store for Flaky { async fn close( &self, namespace: &str, - stream: &StreamRef, + handle: &StreamHandle, producer: &Producer, record: &[u8], ) -> StreamResult { - self.inner.close(namespace, stream, producer, record).await + self.inner.close(namespace, handle, producer, record).await } async fn read_back( diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index d0af10584..3fe1950ef 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -9,8 +9,8 @@ use crate::{ Reason, StreamError, StreamResult, proto::{ - AppendRecord, AppendRecordsRequest, OpenWriterRequest, Producer, StreamProducerKind, - StreamRef, WorkflowTaskEpoch, + AppendRecord, AppendRecordsRequest, OpenWriterRequest, Producer, StreamHandle, + StreamProducerKind, WorkflowTaskEpoch, }, service::StreamService, store::{MAX_BATCH_BYTES, MAX_RECORD_BYTES}, @@ -50,7 +50,7 @@ struct State { struct Writer { namespace: String, - stream: StreamRef, + handle: StreamHandle, producer: Producer, previous_completed_task: Option, failed_tasks: Vec, @@ -130,7 +130,7 @@ impl Writer { async fn send(&self, records: Vec, sequence: i64) -> StreamResult<(i64, i64)> { let request = AppendRecordsRequest { namespace: self.namespace.clone(), - stream: Some(self.stream.clone()), + handle: Some(self.handle.clone()), producer: Some(self.producer.clone()), sequence, records, @@ -228,7 +228,7 @@ impl Writers { request: OpenWriterRequest, ) -> StreamResult { let stream = request - .stream + .handle .ok_or_else(|| StreamError::invalid("The writer has no Stream"))?; crate::ids::check_stream_id(&stream.id)?; let producer = request @@ -250,7 +250,7 @@ impl Writers { } let writer = Arc::new(Writer { namespace: request.namespace, - stream, + handle: stream, producer, previous_completed_task: request.previous_completed_task, failed_tasks: request.failed_tasks, From 7c23329f6438fc32649e3f8e9f4b02d1dedba669 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 1 Nov 2026 10:39:14 -0700 Subject: [PATCH 03/13] Called the writer's open count refs, not handles. --- crates/streams/src/runtime.rs | 2 +- crates/streams/src/testing/writer.rs | 6 ++--- crates/streams/src/writer.rs | 36 ++++++++++++++-------------- 3 files changed, 22 insertions(+), 22 deletions(-) diff --git a/crates/streams/src/runtime.rs b/crates/streams/src/runtime.rs index 420d87739..33436d992 100644 --- a/crates/streams/src/runtime.rs +++ b/crates/streams/src/runtime.rs @@ -121,7 +121,7 @@ impl StreamRuntime { self.shared.name } - /// Releases one handle of a writer with no wait, as the `FreeWriter` method does. A bridge + /// Releases one reference to a writer with no wait, as the `FreeWriter` method does. A bridge /// calls it from a finalizer, which has no async runtime to await on. pub fn free_writer(&self, writer_id: u64) -> StreamResult<()> { self.shared.writers.free(writer_id) diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs index 662d7114e..9da396b47 100644 --- a/crates/streams/src/testing/writer.rs +++ b/crates/streams/src/testing/writer.rs @@ -375,7 +375,7 @@ async fn a_reopen_of_a_closed_writer_goes_on_with_its_sequence() { } } -async fn the_handles_of_one_producer_share_a_writer_until_the_last_one_lets_go() { +async fn the_opens_of_one_producer_share_a_writer_until_the_last_one_lets_go() { for case in cases().await { let a = case.open(client(), |_| {}).await; let b = case.open(client(), |_| {}).await; @@ -385,7 +385,7 @@ async fn the_handles_of_one_producer_share_a_writer_until_the_last_one_lets_go() .call("CloseWriter", CloseWriterRequest { writer_id: a }) .await .unwrap(); - // The other handle still writes, on the same sequence. + // The other reference still writes, on the same sequence. case.write(b, &["2"]).await.unwrap(); case.ext.free_writer(b).unwrap(); let error = case.write(b, &["3"]).await.unwrap_err(); @@ -669,7 +669,7 @@ pub const CASES: &[super::Case] = &[ case!(a_closed_writer_is_gone), case!(a_write_splits_big_records_by_bytes_and_refuses_an_oversize_one), case!(a_reopen_of_a_closed_writer_goes_on_with_its_sequence), - case!(the_handles_of_one_producer_share_a_writer_until_the_last_one_lets_go), + case!(the_opens_of_one_producer_share_a_writer_until_the_last_one_lets_go), case!(a_named_client_with_a_higher_epoch_fences_its_earlier_self), case!(closing_every_writer_drops_them_all), case!(a_write_whose_answer_is_lost_goes_in_once), diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index 3fe1950ef..4b9638b63 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -3,7 +3,7 @@ //! order, so that the Store writes a retried batch only once. Lang can buffer Records on its own //! and write each batch with one call. //! -//! The handles of one Producer epoch on one Stream share one writer. Each open adds a handle and +//! The opens of one Producer epoch on one Stream share one writer. Each open adds a reference, and //! each close or free releases one. The last release drops the writer. use crate::{ @@ -178,11 +178,11 @@ fn check_sizes(producer: &Producer, records: &[AppendRecord]) -> StreamResult<() /// The namespace, Stream id, Producer id and epoch key of a writer. type WriterKey = (String, String, String, String); -/// One open writer and the handles that share it. +/// One open writer and the references to it. struct Open { writer: Arc, key: WriterKey, - handles: usize, + refs: usize, } #[derive(Default)] @@ -193,15 +193,15 @@ struct Table { } impl Table { - /// Releases one handle of a writer, or every handle with `all`. Returns the writer when it has - /// no handle left, so that the caller ends it. + /// Releases one reference to a writer, or every reference with `all`. Returns the writer when + /// it has no reference left, so that the caller ends it. fn release(&mut self, id: u64, all: bool) -> StreamResult>> { let open = self .open .get_mut(&id) .ok_or_else(|| StreamError::new(Reason::UnknownWriter, format!("No writer {id}")))?; - open.handles -= 1; - if open.handles > 0 && !all { + open.refs -= 1; + if open.refs > 0 && !all { return Ok(None); } let open = self.open.remove(&id).expect("found above"); @@ -218,7 +218,7 @@ pub(crate) struct Writers { } impl Writers { - /// Opens a writer, or adds a handle to the one that this Producer epoch has on the Stream. A + /// Opens a writer, or adds a reference to the one that this Producer epoch has on the Stream. A /// writer of a Local Activity, or of an Activity attempt above 1, first sends an empty batch. /// So the new epoch withdraws the earlier one's Records even when it publishes nothing. On a /// closed or expired Stream the open still succeeds, and the first write fails with the reason. @@ -287,7 +287,7 @@ impl Writers { } let mut table = self.table.lock(); if let Some(id) = table.by_producer.get(&key).copied() { - table.open.get_mut(&id).expect("indexed").handles += 1; + table.open.get_mut(&id).expect("indexed").refs += 1; return Ok(id); } let id = self.next.fetch_add(1, Ordering::Relaxed) + 1; @@ -297,17 +297,17 @@ impl Writers { Open { writer, key, - handles: 1, + refs: 1, }, ); Ok(id) } - /// Adds a handle to the open writer of `key`, if there is one. + /// Adds a reference to the open writer of `key`, if there is one. fn share(&self, key: &WriterKey) -> Option { let mut table = self.table.lock(); let id = table.by_producer.get(key).copied()?; - table.open.get_mut(&id).expect("indexed").handles += 1; + table.open.get_mut(&id).expect("indexed").refs += 1; Some(id) } @@ -344,7 +344,7 @@ impl Writers { writer.write(records).await } - /// Closes every open writer, whatever its handles. Returns how many had failed. + /// Closes every open writer, whatever its references. Returns how many had failed. pub(crate) async fn close_all(&self) -> usize { let ids: Vec = self.table.lock().open.keys().copied().collect(); let mut failed = 0; @@ -360,7 +360,7 @@ impl Writers { failed } - /// Releases one handle. The last one waits for a write in flight, then drops the writer. + /// Releases one reference. The last one waits for a write in flight, then drops the writer. /// Returns its last offset, or the error it failed with. pub(crate) async fn close(&self, id: u64) -> StreamResult { let writer = self.get(id)?; @@ -374,8 +374,8 @@ impl Writers { end(&writer).await } - /// Releases one handle with no wait, for a handle that lang drops. A write in flight still - /// completes, and a write that waits behind it is refused once no handle is left. + /// Releases one reference with no wait, for a reference that lang drops. A write in flight + /// still completes, and a write that waits behind it is refused once no reference is left. pub(crate) fn free(&self, id: u64) -> StreamResult<()> { if let Some(writer) = self.table.lock().release(id, false)? { writer.state.lock().closed = true; @@ -384,8 +384,8 @@ impl Writers { } } -/// Ends a writer that no handle uses. It waits for a write in flight, then returns its last offset -/// or the error it failed with. +/// Ends a writer that no reference uses. It waits for a write in flight, then returns its last +/// offset or the error it failed with. async fn end(writer: &Writer) -> StreamResult { writer.state.lock().closed = true; let _writing = writer.writing.lock().await; From d496f5998bcd0c3202eae60d78fab00ad2a9eabb Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 4 Nov 2026 01:53:38 -0700 Subject: [PATCH 04/13] Capped a writer's batch at the Store's Record limit. The Store now refuses an append of more than 1000 Records, so a larger max_batch_size would fail every full batch. --- .../protos/temporal/sdk/streams/v1/runtime.proto | 2 +- crates/streams/src/testing/writer.rs | 16 ++++++++++++++-- crates/streams/src/writer.rs | 8 ++++---- 3 files changed, 19 insertions(+), 7 deletions(-) diff --git a/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto index 415f58653..be56d3d70 100644 --- a/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto +++ b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto @@ -33,7 +33,7 @@ message OpenWriterRequest { string namespace = 1; temporal.api.stream.v1.StreamHandle handle = 2; temporal.api.stream.v1.Producer producer = 3; - // A write sends at most this many Records in one batch. Zero is 1000. + // A write sends at most this many Records in one batch. Zero, or a value above 1000, is 1000. int32 max_batch_size = 5; // How long a write retries a Store failure. Empty is 10 s. google.protobuf.Duration max_retry_duration = 6; diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs index 9da396b47..7cef61701 100644 --- a/crates/streams/src/testing/writer.rs +++ b/crates/streams/src/testing/writer.rs @@ -21,8 +21,8 @@ use crate::{ WriteRecordsResponse, stream_record::Item, }, store::{ - Appended, Batch, Capabilities, Closed, Created, Entry, Owner, Store, StreamState, - Withdrawal, + Appended, Batch, Capabilities, Closed, Created, Entry, MAX_BATCH_RECORDS, Owner, Store, + StreamState, Withdrawal, }, }; use prost::Message; @@ -237,6 +237,17 @@ async fn writes_split_at_the_batch_size_and_keep_one_sequence() { } } +async fn a_batch_size_over_the_store_limit_writes_in_more_batches() { + for case in cases().await { + let writer = case.open(client(), |r| r.max_batch_size = 5000).await; + let texts: Vec = (0..=MAX_BATCH_RECORDS).map(|i| i.to_string()).collect(); + let texts: Vec<&str> = texts.iter().map(String::as_str).collect(); + // The Store refuses one batch of this many, so the write goes in two. + let (first, last) = case.write(writer, &texts).await.unwrap(); + assert!(first < last); + } +} + async fn a_failed_write_fails_every_later_write_and_the_close() { for case in cases().await { let writer = case.open(client(), |_| {}).await; @@ -663,6 +674,7 @@ async fn a_local_activity_writer_withdraws_the_records_of_a_failed_task() { pub const CASES: &[super::Case] = &[ case!(a_write_is_in_the_store_when_it_answers_its_offsets), case!(writes_split_at_the_batch_size_and_keep_one_sequence), + case!(a_batch_size_over_the_store_limit_writes_in_more_batches), case!(a_failed_write_fails_every_later_write_and_the_close), case!(an_activity_retry_withdraws_the_earlier_attempt_even_with_nothing_to_write), case!(an_activity_retry_opens_on_a_closed_or_expired_stream), diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index 4b9638b63..6393d5601 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -13,7 +13,7 @@ use crate::{ StreamProducerKind, WorkflowTaskEpoch, }, service::StreamService, - store::{MAX_BATCH_BYTES, MAX_RECORD_BYTES}, + store::{MAX_BATCH_BYTES, MAX_BATCH_RECORDS, MAX_RECORD_BYTES}, }; use parking_lot::Mutex; use prost::Message; @@ -27,7 +27,6 @@ use std::{ }; use tokio::time::Instant; -const DEFAULT_BATCH: usize = 1000; const DEFAULT_RETRY: Duration = Duration::from_secs(10); const FIRST_BACKOFF: Duration = Duration::from_millis(50); @@ -254,9 +253,10 @@ impl Writers { producer, previous_completed_task: request.previous_completed_task, failed_tasks: request.failed_tasks, + // The Store refuses a batch over its count limit, so a larger size can't go in one batch. max_batch: match request.max_batch_size { - n if n > 0 => n as usize, - _ => DEFAULT_BATCH, + n if n > 0 => (n as usize).min(MAX_BATCH_RECORDS), + _ => MAX_BATCH_RECORDS, }, max_retry: duration(request.max_retry_duration).unwrap_or(DEFAULT_RETRY), service: service.clone(), From 358c45363577664566bd69b7f420c2a268ffcd0d Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 4 Nov 2026 18:08:47 -0700 Subject: [PATCH 05/13] Took the writer's retry rule from the Producer's epoch case. The case also gives an Activity's attempt, so the rule reads one field. The writer suite builds its Producers and creates from the new shapes. --- crates/streams/src/testing/writer.rs | 79 +++++++++++++++------------- crates/streams/src/writer.rs | 13 ++--- 2 files changed, 48 insertions(+), 44 deletions(-) diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs index 7cef61701..6fd5e7276 100644 --- a/crates/streams/src/testing/writer.rs +++ b/crates/streams/src/testing/writer.rs @@ -14,15 +14,15 @@ use crate::{ named_client_producer_id, }, proto::{ - AppendRecord, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, - CreateStreamRequest, CreateStreamResponse, OpenWriterRequest, OpenWriterResponse, Producer, - ProducerEpoch, ReadRecordsRequest, ReadRecordsResponse, StreamErrorDetails, StreamHandle, - StreamProducerKind, StreamRecord, WorkflowTaskEpoch, WriteRecordsRequest, - WriteRecordsResponse, stream_record::Item, + ActivityEpoch, AppendRecord, ClientEpoch, CloseStreamRequest, CloseWriterRequest, + CloseWriterResponse, CreateStreamRequest, CreateStreamResponse, LocalActivityEpoch, + OpenWriterRequest, OpenWriterResponse, Producer, ProducerEpoch, ReadRecordsRequest, + ReadRecordsResponse, StreamErrorDetails, StreamHandle, StreamRecord, WorkflowTaskEpoch, + WriteRecordsRequest, WriteRecordsResponse, producer_epoch::Kind, stream_record::Item, }, store::{ - Appended, Batch, Capabilities, Closed, Created, Entry, MAX_BATCH_RECORDS, Owner, Store, - StreamState, Withdrawal, + Appended, Batch, Capabilities, Closed, Create, Created, Entry, MAX_BATCH_RECORDS, Store, + StreamState, }, }; use prost::Message; @@ -63,7 +63,10 @@ async fn cases() -> Vec { "CreateStream", CreateStreamRequest { namespace: case.namespace.clone(), - stream_id: case.stream.clone(), + stream_handle: Some(StreamHandle { + id: case.stream.clone(), + ..Default::default() + }), ..Default::default() }, ) @@ -77,11 +80,9 @@ async fn cases() -> Vec { fn client() -> Producer { Producer { - kind: StreamProducerKind::Client as i32, id: client_producer_id("me", "p"), epoch: Some(ProducerEpoch { - attempt: 1, - ..Default::default() + kind: Some(Kind::Client(ClientEpoch { attempt: 1 })), }), run_id: String::new(), } @@ -89,15 +90,15 @@ fn client() -> Producer { fn activity(attempt: i32) -> Producer { Producer { - kind: StreamProducerKind::Activity as i32, id: activity_producer_id("r1", "a", "s"), epoch: Some(ProducerEpoch { - attempt_scheduled_time: Some(Timestamp { - seconds: 10 * attempt as i64, - nanos: 0, - }), - attempt, - ..Default::default() + kind: Some(Kind::Activity(ActivityEpoch { + attempt_scheduled_time: Some(Timestamp { + seconds: 10 * attempt as i64, + nanos: 0, + }), + attempt, + })), }), run_id: "r1".to_string(), } @@ -313,7 +314,10 @@ async fn an_activity_retry_opens_on_a_closed_or_expired_stream() { "CreateStream", CreateStreamRequest { namespace: case.namespace.clone(), - stream_id: short.clone(), + stream_handle: Some(StreamHandle { + id: short.clone(), + ..Default::default() + }), retention: millis(300), ..Default::default() }, @@ -408,11 +412,9 @@ async fn the_opens_of_one_producer_share_a_writer_until_the_last_one_lets_go() { async fn a_named_client_with_a_higher_epoch_fences_its_earlier_self() { for case in cases().await { let named = |attempt| Producer { - kind: StreamProducerKind::Client as i32, id: named_client_producer_id("ingest-1"), epoch: Some(ProducerEpoch { - attempt, - ..Default::default() + kind: Some(Kind::Client(ClientEpoch { attempt })), }), run_id: String::new(), }; @@ -472,13 +474,9 @@ impl Store for Flaky { &self, namespace: &str, handle: &StreamHandle, - retention: Option, - owner: Owner, - withdrawal: &Withdrawal<'_>, + create: &Create<'_>, ) -> StreamResult { - self.inner - .create(namespace, handle, retention, owner, withdrawal) - .await + self.inner.create(namespace, handle, create).await } async fn append( @@ -561,7 +559,10 @@ async fn flaky_cases(lose_next: bool) -> Vec<(Arc, Case)> { "CreateStream", CreateStreamRequest { namespace: case.namespace.clone(), - stream_id: case.stream.clone(), + stream_handle: Some(StreamHandle { + id: case.stream.clone(), + ..Default::default() + }), ..Default::default() }, ) @@ -634,17 +635,19 @@ async fn a_write_that_waits_behind_a_close_is_refused() { async fn a_local_activity_writer_withdraws_the_records_of_a_failed_task() { for case in cases().await { let local_activity = |call: &str, started: i64| Producer { - kind: StreamProducerKind::LocalActivity as i32, id: local_activity_producer_id("r1", call), epoch: Some(ProducerEpoch { - task_started_event_id: started, - task_attempt: 1, - task_started_time: Some(Timestamp { - seconds: started, - nanos: 0, - }), - attempt: 1, - ..Default::default() + kind: Some(Kind::LocalActivity(LocalActivityEpoch { + task: Some(WorkflowTaskEpoch { + started_event_id: started, + attempt: 1, + started_time: Some(Timestamp { + seconds: started, + nanos: 0, + }), + }), + attempt: 1, + })), }), run_id: "r1".to_string(), }; diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index 4b752f117..72229d30c 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -8,9 +8,10 @@ use crate::{ Reason, StreamError, StreamResult, + epoch::ProducerKind, proto::{ AppendRecord, AppendRecordsRequest, OpenWriterRequest, Producer, StreamHandle, - StreamProducerKind, WorkflowTaskEpoch, + WorkflowTaskEpoch, producer_epoch::Kind, }, service::StreamService, store::{MAX_BATCH_BYTES, MAX_BATCH_RECORDS, MAX_RECORD_BYTES}, @@ -234,7 +235,7 @@ impl Writers { .producer .ok_or_else(|| StreamError::invalid("The writer has no Producer"))?; crate::ids::check_producer(&producer)?; - if producer.kind == StreamProducerKind::Workflow as i32 { + if ProducerKind::try_from(&producer)? == ProducerKind::Workflow { return Err(StreamError::invalid("Workflow code doesn't open a writer")); } let epoch = producer.epoch.unwrap_or_default(); @@ -265,10 +266,10 @@ impl Writers { }); // A Local Activity that runs again in a later task has attempt 1 again, so its writer // always sends the empty batch. - let retry = match StreamProducerKind::try_from(writer.producer.kind).unwrap_or_default() { - StreamProducerKind::LocalActivity => true, - StreamProducerKind::Activity | StreamProducerKind::StandaloneActivity => { - epoch.attempt > 1 + let retry = match epoch.kind { + Some(Kind::LocalActivity(_)) => true, + Some(Kind::Activity(activity) | Kind::StandaloneActivity(activity)) => { + activity.attempt > 1 } _ => false, }; From 06284abf00faff676deed50944ceca40a2a6e487 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 4 Nov 2026 18:10:19 -0700 Subject: [PATCH 06/13] Shortened a writer comment to the line limit. --- crates/streams/src/writer.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index 72229d30c..b3be3467d 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -254,7 +254,7 @@ impl Writers { producer, previous_completed_task: request.previous_completed_task, failed_tasks: request.failed_tasks, - // The Store refuses a batch over its count limit, so a larger size can't go in one batch. + // The Store refuses a batch over its Record limit, so no batch can be larger. max_batch: match request.max_batch_size { n if n > 0 => (n as usize).min(MAX_BATCH_RECORDS), _ => MAX_BATCH_RECORDS, From 804eb13554a4d032c386a639a2b7456b7ae5c816 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 4 Nov 2026 22:44:13 -0700 Subject: [PATCH 07/13] Retried a writer's batch only when the Store is unavailable. Corrupt Store data reads the same on a retry, so the write fails at once and the writer fails. The writer suite checks it on a Store that refuses one append as corrupt. --- crates/streams/src/testing/writer.rs | 35 +++++++++++++++++++++++----- crates/streams/src/writer.rs | 5 ++-- 2 files changed, 32 insertions(+), 8 deletions(-) diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs index 6fd5e7276..8fa60049f 100644 --- a/crates/streams/src/testing/writer.rs +++ b/crates/streams/src/testing/writer.rs @@ -446,11 +446,13 @@ async fn closing_every_writer_drops_them_all() { } /// A Store over another one, whose appends fail. It can write an append and lose its answer, as on -/// a dropped connection, and it refuses every append while it is down. +/// a dropped connection, and it refuses every append while it is down. It can also refuse the next +/// append as corrupt Store data. struct Flaky { inner: Arc, lose_next: AtomicBool, down: AtomicBool, + corrupt_next: AtomicBool, } impl Flaky { @@ -460,6 +462,7 @@ impl Flaky { inner, lose_next: AtomicBool::new(false), down: AtomicBool::new(false), + corrupt_next: AtomicBool::new(false), } } } @@ -486,11 +489,14 @@ impl Store for Flaky { batch: &Batch<'_>, ) -> StreamResult { if self.down.load(Ordering::SeqCst) { - return Err(StreamError::storage("the Store is down")); + return Err(StreamError::storage_unavailable("the Store is down")); + } + if self.corrupt_next.swap(false, Ordering::SeqCst) { + return Err(StreamError::storage_corrupt("a stored field is bad")); } let appended = self.inner.append(namespace, handle, batch).await?; if self.lose_next.swap(false, Ordering::SeqCst) { - return Err(StreamError::storage("the answer was lost")); + return Err(StreamError::storage_unavailable("the answer was lost")); } Ok(appended) } @@ -522,9 +528,8 @@ impl Store for Flaky { namespace: &str, handle: &StreamHandle, producer: &Producer, - record: &[u8], ) -> StreamResult { - self.inner.close(namespace, handle, producer, record).await + self.inner.close(namespace, handle, producer).await } async fn read_back( @@ -600,7 +605,7 @@ async fn a_writer_writes_again_after_a_store_outage() { let error = case.write(writer, &["lost"]).await.unwrap_err(); assert_eq!( (error.code(), reason(&error).as_str()), - (Code::Unavailable, "stream_storage") + (Code::Unavailable, "stream_storage_unavailable") ); store.down.store(false, Ordering::SeqCst); case.write(writer, &["b"]).await.unwrap(); @@ -613,6 +618,23 @@ async fn a_writer_writes_again_after_a_store_outage() { } } +/// A write that meets corrupt Store data fails at once, since a retry reads the same data. The +/// writer then fails every later write. +async fn a_write_on_corrupt_store_data_fails_with_no_retry() { + for (store, case) in flaky_cases(false).await { + let writer = case.open(client(), |_| {}).await; + store.corrupt_next.store(true, Ordering::SeqCst); + let error = case.write(writer, &["a"]).await.unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::DataLoss, "stream_storage_corrupt") + ); + assert!(case.visible().await.is_empty()); + let again = case.write(writer, &["b"]).await.unwrap_err(); + assert_eq!(reason(&again), "stream_storage_corrupt"); + } +} + async fn a_write_that_waits_behind_a_close_is_refused() { for case in cases().await { let writer = case.open(client(), |_| {}).await; @@ -689,6 +711,7 @@ pub const CASES: &[super::Case] = &[ case!(closing_every_writer_drops_them_all), case!(a_write_whose_answer_is_lost_goes_in_once), case!(a_writer_writes_again_after_a_store_outage), + case!(a_write_on_corrupt_store_data_fails_with_no_retry), case!(a_write_that_waits_behind_a_close_is_refused), case!(a_local_activity_writer_withdraws_the_records_of_a_failed_task), ]; diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index b3be3467d..dce5b2718 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -108,7 +108,7 @@ impl Writer { Ok(offsets) => offsets, // The next write takes a higher `sequence`, so the Store refuses a late copy of // this batch after it. - Err(error) if error.reason == Reason::Storage => return Err(error), + Err(error) if error.reason == Reason::StorageUnavailable => return Err(error), Err(error) => { self.state.lock().failed = Some(error.clone()); return Err(error); @@ -144,7 +144,8 @@ impl Writer { match self.service.append_records(request.clone()).await { Ok(response) => return Ok((response.first_offset, response.last_offset)), Err(error) - if error.reason == Reason::Storage && Instant::now() + backoff < until => + if error.reason == Reason::StorageUnavailable + && Instant::now() + backoff < until => { tokio::time::sleep(backoff).await; backoff = (backoff * 2).min(Duration::from_secs(1)); From 387eaa55e4632cdac5c426cfbecd0ce1359a66fa Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 4 Nov 2026 22:51:02 -0700 Subject: [PATCH 08/13] Called the writer's and its suite's Stream a ref. An open carries its StreamRef in ref, and the writer keeps it as stream_ref. --- crates/streams/src/testing/writer.rs | 30 ++++++++++++++-------------- crates/streams/src/writer.rs | 10 +++++----- 2 files changed, 20 insertions(+), 20 deletions(-) diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs index 8fa60049f..176f0781f 100644 --- a/crates/streams/src/testing/writer.rs +++ b/crates/streams/src/testing/writer.rs @@ -17,7 +17,7 @@ use crate::{ ActivityEpoch, AppendRecord, ClientEpoch, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, CreateStreamRequest, CreateStreamResponse, LocalActivityEpoch, OpenWriterRequest, OpenWriterResponse, Producer, ProducerEpoch, ReadRecordsRequest, - ReadRecordsResponse, StreamErrorDetails, StreamHandle, StreamRecord, WorkflowTaskEpoch, + ReadRecordsResponse, StreamErrorDetails, StreamRecord, StreamRef, WorkflowTaskEpoch, WriteRecordsRequest, WriteRecordsResponse, producer_epoch::Kind, stream_record::Item, }, store::{ @@ -63,7 +63,7 @@ async fn cases() -> Vec { "CreateStream", CreateStreamRequest { namespace: case.namespace.clone(), - stream_handle: Some(StreamHandle { + stream_ref: Some(StreamRef { id: case.stream.clone(), ..Default::default() }), @@ -142,7 +142,7 @@ impl Case { async fn open(&self, producer: Producer, change: impl Fn(&mut OpenWriterRequest)) -> u64 { let mut request = OpenWriterRequest { namespace: self.namespace.clone(), - handle: Some(StreamHandle { + r#ref: Some(StreamRef { id: self.stream.clone(), ..Default::default() }), @@ -173,7 +173,7 @@ impl Case { "ReadRecords", ReadRecordsRequest { namespace: self.namespace.clone(), - handle: Some(StreamHandle { + r#ref: Some(StreamRef { id: self.stream.clone(), ..Default::default() }), @@ -254,7 +254,7 @@ async fn a_failed_write_fails_every_later_write_and_the_close() { let writer = case.open(client(), |_| {}).await; let close = CloseStreamRequest { namespace: case.namespace.clone(), - handle: Some(StreamHandle { + r#ref: Some(StreamRef { id: case.stream.clone(), ..Default::default() }), @@ -296,7 +296,7 @@ async fn an_activity_retry_opens_on_a_closed_or_expired_stream() { case.write(first, &["a1"]).await.unwrap(); let close = CloseStreamRequest { namespace: case.namespace.clone(), - handle: Some(StreamHandle { + r#ref: Some(StreamRef { id: case.stream.clone(), ..Default::default() }), @@ -314,7 +314,7 @@ async fn an_activity_retry_opens_on_a_closed_or_expired_stream() { "CreateStream", CreateStreamRequest { namespace: case.namespace.clone(), - stream_handle: Some(StreamHandle { + stream_ref: Some(StreamRef { id: short.clone(), ..Default::default() }), @@ -325,7 +325,7 @@ async fn an_activity_retry_opens_on_a_closed_or_expired_stream() { .await .unwrap(); let on_short = |request: &mut OpenWriterRequest| { - request.handle = Some(StreamHandle { + request.r#ref = Some(StreamRef { id: short.clone(), ..Default::default() }); @@ -476,16 +476,16 @@ impl Store for Flaky { async fn create( &self, namespace: &str, - handle: &StreamHandle, + stream_ref: &StreamRef, create: &Create<'_>, ) -> StreamResult { - self.inner.create(namespace, handle, create).await + self.inner.create(namespace, stream_ref, create).await } async fn append( &self, namespace: &str, - handle: &StreamHandle, + stream_ref: &StreamRef, batch: &Batch<'_>, ) -> StreamResult { if self.down.load(Ordering::SeqCst) { @@ -494,7 +494,7 @@ impl Store for Flaky { if self.corrupt_next.swap(false, Ordering::SeqCst) { return Err(StreamError::storage_corrupt("a stored field is bad")); } - let appended = self.inner.append(namespace, handle, batch).await?; + let appended = self.inner.append(namespace, stream_ref, batch).await?; if self.lose_next.swap(false, Ordering::SeqCst) { return Err(StreamError::storage_unavailable("the answer was lost")); } @@ -526,10 +526,10 @@ impl Store for Flaky { async fn close( &self, namespace: &str, - handle: &StreamHandle, + stream_ref: &StreamRef, producer: &Producer, ) -> StreamResult { - self.inner.close(namespace, handle, producer).await + self.inner.close(namespace, stream_ref, producer).await } async fn read_back( @@ -564,7 +564,7 @@ async fn flaky_cases(lose_next: bool) -> Vec<(Arc, Case)> { "CreateStream", CreateStreamRequest { namespace: case.namespace.clone(), - stream_handle: Some(StreamHandle { + stream_ref: Some(StreamRef { id: case.stream.clone(), ..Default::default() }), diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index dce5b2718..52493245f 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -10,7 +10,7 @@ use crate::{ Reason, StreamError, StreamResult, epoch::ProducerKind, proto::{ - AppendRecord, AppendRecordsRequest, OpenWriterRequest, Producer, StreamHandle, + AppendRecord, AppendRecordsRequest, OpenWriterRequest, Producer, StreamRef, WorkflowTaskEpoch, producer_epoch::Kind, }, service::StreamService, @@ -50,7 +50,7 @@ struct State { struct Writer { namespace: String, - handle: StreamHandle, + stream_ref: StreamRef, producer: Producer, previous_completed_task: Option, failed_tasks: Vec, @@ -130,7 +130,7 @@ impl Writer { async fn send(&self, records: Vec, sequence: i64) -> StreamResult<(i64, i64)> { let request = AppendRecordsRequest { namespace: self.namespace.clone(), - handle: Some(self.handle.clone()), + r#ref: Some(self.stream_ref.clone()), producer: Some(self.producer.clone()), sequence, records, @@ -229,7 +229,7 @@ impl Writers { request: OpenWriterRequest, ) -> StreamResult { let stream = request - .handle + .r#ref .ok_or_else(|| StreamError::invalid("The writer has no Stream"))?; crate::ids::check_stream_id(&stream.id)?; let producer = request @@ -251,7 +251,7 @@ impl Writers { } let writer = Arc::new(Writer { namespace: request.namespace, - handle: stream, + stream_ref: stream, producer, previous_completed_task: request.previous_completed_task, failed_tasks: request.failed_tasks, From d4bca4160d610caea6c52c972cf2e5a0af976849 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Thu, 5 Nov 2026 10:49:21 -0700 Subject: [PATCH 09/13] Sent an open's StreamWriter with each of the writer's batches. OpenWriterRequest carries the Producer and the task facts in one StreamWriter, and each batch sends it as AppendRecords does. --- crates/streams/src/testing/writer.rs | 24 +++++++++++++++++------- crates/streams/src/writer.rs | 19 +++++++++---------- 2 files changed, 26 insertions(+), 17 deletions(-) diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs index 176f0781f..eeb84eb75 100644 --- a/crates/streams/src/testing/writer.rs +++ b/crates/streams/src/testing/writer.rs @@ -17,8 +17,9 @@ use crate::{ ActivityEpoch, AppendRecord, ClientEpoch, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, CreateStreamRequest, CreateStreamResponse, LocalActivityEpoch, OpenWriterRequest, OpenWriterResponse, Producer, ProducerEpoch, ReadRecordsRequest, - ReadRecordsResponse, StreamErrorDetails, StreamRecord, StreamRef, WorkflowTaskEpoch, - WriteRecordsRequest, WriteRecordsResponse, producer_epoch::Kind, stream_record::Item, + ReadRecordsResponse, StreamErrorDetails, StreamRecord, StreamRef, StreamWriter, + WorkflowTaskEpoch, WriteRecordsRequest, WriteRecordsResponse, producer_epoch::Kind, + stream_record::Item, }, store::{ Appended, Batch, Capabilities, Closed, Create, Created, Entry, MAX_BATCH_RECORDS, Store, @@ -78,6 +79,14 @@ async fn cases() -> Vec { cases } +/// The writer of a write by `producer`, with no withdrawal facts. +fn writer_for(producer: Producer) -> Option { + Some(StreamWriter { + producer: Some(producer), + ..Default::default() + }) +} + fn client() -> Producer { Producer { id: client_producer_id("me", "p"), @@ -146,7 +155,7 @@ impl Case { id: self.stream.clone(), ..Default::default() }), - producer: Some(producer), + writer: writer_for(producer), ..Default::default() }; change(&mut request); @@ -258,7 +267,7 @@ async fn a_failed_write_fails_every_later_write_and_the_close() { id: case.stream.clone(), ..Default::default() }), - producer: Some(client()), + writer: writer_for(client()), ..Default::default() }; let _: crate::proto::CloseStreamResponse = case.call("CloseStream", close).await.unwrap(); @@ -300,7 +309,7 @@ async fn an_activity_retry_opens_on_a_closed_or_expired_stream() { id: case.stream.clone(), ..Default::default() }), - producer: Some(client()), + writer: writer_for(client()), ..Default::default() }; let _: crate::proto::CloseStreamResponse = case.call("CloseStream", close).await.unwrap(); @@ -675,11 +684,12 @@ async fn a_local_activity_writer_withdraws_the_records_of_a_failed_task() { }; let facts = |completed: i64, failed: Vec| { move |request: &mut OpenWriterRequest| { - request.previous_completed_task = Some(WorkflowTaskEpoch { + let writer = request.writer.get_or_insert_with(Default::default); + writer.previous_completed_task = Some(WorkflowTaskEpoch { started_event_id: completed, ..Default::default() }); - request.failed_tasks = failed.clone(); + writer.failed_tasks = failed.clone(); } }; // Task [3] fails after its Local Activity writes, and task [4] completes with no write. diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index 52493245f..635f9d44b 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -10,8 +10,8 @@ use crate::{ Reason, StreamError, StreamResult, epoch::ProducerKind, proto::{ - AppendRecord, AppendRecordsRequest, OpenWriterRequest, Producer, StreamRef, - WorkflowTaskEpoch, producer_epoch::Kind, + AppendRecord, AppendRecordsRequest, OpenWriterRequest, Producer, StreamRef, StreamWriter, + producer_epoch::Kind, }, service::StreamService, store::{MAX_BATCH_BYTES, MAX_BATCH_RECORDS, MAX_RECORD_BYTES}, @@ -52,8 +52,8 @@ struct Writer { namespace: String, stream_ref: StreamRef, producer: Producer, - previous_completed_task: Option, - failed_tasks: Vec, + /// The open's writer, with its Producer and withdrawal facts, which each batch sends. + stream_writer: StreamWriter, max_batch: usize, max_retry: Duration, service: Arc, @@ -131,11 +131,9 @@ impl Writer { let request = AppendRecordsRequest { namespace: self.namespace.clone(), r#ref: Some(self.stream_ref.clone()), - producer: Some(self.producer.clone()), + writer: Some(self.stream_writer.clone()), sequence, records, - previous_completed_task: self.previous_completed_task, - failed_tasks: self.failed_tasks.clone(), ..Default::default() }; let until = Instant::now() + self.max_retry; @@ -232,8 +230,10 @@ impl Writers { .r#ref .ok_or_else(|| StreamError::invalid("The writer has no Stream"))?; crate::ids::check_stream_id(&stream.id)?; - let producer = request + let stream_writer = request.writer.unwrap_or_default(); + let producer = stream_writer .producer + .clone() .ok_or_else(|| StreamError::invalid("The writer has no Producer"))?; crate::ids::check_producer(&producer)?; if ProducerKind::try_from(&producer)? == ProducerKind::Workflow { @@ -253,8 +253,7 @@ impl Writers { namespace: request.namespace, stream_ref: stream, producer, - previous_completed_task: request.previous_completed_task, - failed_tasks: request.failed_tasks, + stream_writer, // The Store refuses a batch over its Record limit, so no batch can be larger. max_batch: match request.max_batch_size { n if n > 0 => (n as usize).min(MAX_BATCH_RECORDS), From 36b5383597dbff1e67d7d77d044fe0ed6bc520dc Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Thu, 5 Nov 2026 10:55:04 -0700 Subject: [PATCH 10/13] Refused a writer whose StreamWriter carries a withdraw_run. Only a retry run's first append withdraws a run, and a writer sends many batches. The writer suite checks the refusal. --- crates/streams/src/testing/writer.rs | 30 ++++++++++++++++++++++++++++ crates/streams/src/writer.rs | 6 +++++- 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs index eeb84eb75..862fed106 100644 --- a/crates/streams/src/testing/writer.rs +++ b/crates/streams/src/testing/writer.rs @@ -644,6 +644,35 @@ async fn a_write_on_corrupt_store_data_fails_with_no_retry() { } } +/// An open of a writer with a run to withdraw fails with `stream_invalid` and writes nothing, since +/// only a retry run's first append withdraws a run and a writer sends many batches. +async fn a_writer_with_a_run_to_withdraw_is_refused() { + for case in cases().await { + let request = OpenWriterRequest { + namespace: case.namespace.clone(), + r#ref: Some(StreamRef { + id: case.stream.clone(), + ..Default::default() + }), + writer: Some(StreamWriter { + producer: Some(client()), + withdraw_run: "r1".to_string(), + ..Default::default() + }), + ..Default::default() + }; + let error = case + .call::<_, OpenWriterResponse>("OpenWriter", request) + .await + .unwrap_err(); + assert_eq!( + (error.code(), reason(&error).as_str()), + (Code::InvalidArgument, "stream_invalid") + ); + assert!(case.visible().await.is_empty()); + } +} + async fn a_write_that_waits_behind_a_close_is_refused() { for case in cases().await { let writer = case.open(client(), |_| {}).await; @@ -722,6 +751,7 @@ pub const CASES: &[super::Case] = &[ case!(a_write_whose_answer_is_lost_goes_in_once), case!(a_writer_writes_again_after_a_store_outage), case!(a_write_on_corrupt_store_data_fails_with_no_retry), + case!(a_writer_with_a_run_to_withdraw_is_refused), case!(a_write_that_waits_behind_a_close_is_refused), case!(a_local_activity_writer_withdraws_the_records_of_a_failed_task), ]; diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index 635f9d44b..307ad4fbc 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -52,7 +52,7 @@ struct Writer { namespace: String, stream_ref: StreamRef, producer: Producer, - /// The open's writer, with its Producer and withdrawal facts, which each batch sends. + /// The open's writer, with its Producer and task facts, which each batch sends. stream_writer: StreamWriter, max_batch: usize, max_retry: Duration, @@ -231,6 +231,10 @@ impl Writers { .ok_or_else(|| StreamError::invalid("The writer has no Stream"))?; crate::ids::check_stream_id(&stream.id)?; let stream_writer = request.writer.unwrap_or_default(); + // Only a retry run's first append withdraws a run, and a writer sends many batches. + if !stream_writer.withdraw_run.is_empty() { + return Err(StreamError::invalid("A writer doesn't withdraw a run")); + } let producer = stream_writer .producer .clone() From 872c90fcc4782380c661de2e763484f9f205fc8c Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Thu, 5 Nov 2026 15:38:09 -0700 Subject: [PATCH 11/13] Checked that a bad writer request is invalid in the runtime test. --- crates/streams/src/runtime.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/crates/streams/src/runtime.rs b/crates/streams/src/runtime.rs index aec3ac054..9f21f687b 100644 --- a/crates/streams/src/runtime.rs +++ b/crates/streams/src/runtime.rs @@ -248,12 +248,17 @@ pub(crate) fn decode(input: &[u8]) -> StreamResult { mod tests { use super::*; - const METHODS: [&str; 5] = [ + const METHODS: [&str; 10] = [ "CreateStream", "AppendRecords", "CloseStream", "ReadRecords", "DescribeStream", + "OpenWriter", + "WriteRecords", + "CloseWriters", + "CloseWriter", + "FreeWriter", ]; fn runtime() -> Arc { From a3225534e1418d608a6b8d3b32674c80fbb415df Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 6 Nov 2026 17:13:44 -0700 Subject: [PATCH 12/13] Put the Producer kind and the writer's withdrawal into variants. The review asked to stop representing invalid states. The Producer kind is now a oneof with run_id only in the run-based variants, so a client can't carry a run. A write's task facts and withdraw_run move into RunWithdrawal, which only a run writer sets. --- .../temporal/api/stream/v1/message.proto | 102 +++++++++++------- .../streamservice/v1/request_response.proto | 2 +- .../temporal/sdk/core/streams/streams.proto | 7 +- 3 files changed, 68 insertions(+), 43 deletions(-) diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto index 63bca5f6c..7385028a1 100644 --- a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto @@ -29,27 +29,6 @@ message StreamRef { string backend = 3; } -// Experimental. Might change in the future. -// The ordered attempt of a Producer. A higher epoch supersedes the Records of a lower epoch of the -// same Producer. The case is the kind of the Producer. It decides how the Store orders the epochs, -// and which earlier Records a higher epoch withdraws. -message ProducerEpoch { - oneof kind { - // A Workflow run, which writes in its Workflow Task. A higher epoch withdraws the Records - // of the run's Workflow Tasks that never completed, also those of their Local Activities. - WorkflowTaskEpoch workflow = 1; - // A Local Activity. A higher epoch withdraws the Producer's lower epochs. - LocalActivityEpoch local_activity = 2; - // An Activity of a Workflow. A higher epoch withdraws the Producer's lower epochs. - ActivityEpoch activity = 3; - // A Standalone Activity. A higher epoch withdraws the Producer's lower epochs. - ActivityEpoch standalone_activity = 4; - // A client. A higher epoch only fences the lower ones, since a restarted client's earlier - // Records are real messages, not the Records of a failed attempt. - ClientEpoch client = 5; - } -} - // Experimental. Might change in the future. // One Workflow Task attempt of a run. It orders by started_event_id, then attempt, then // started_time. @@ -86,32 +65,78 @@ message ClientEpoch { } // Experimental. Might change in the future. -// The code that wrote a batch. Its id stays the same across attempts, and its epoch orders them. +// A Workflow run Producer. It writes in its Workflow Task. A higher epoch withdraws the Records of +// the run's Workflow Tasks that never completed, with those of their Local Activities. +message WorkflowProducer { + // The first execution run id of the run chain. + string run_id = 1; + WorkflowTaskEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// A Local Activity Producer. A higher epoch withdraws the Producer's lower epochs. +message LocalActivityProducer { + string run_id = 1; + LocalActivityEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// An Activity Producer, for an Activity of a Workflow or a Standalone Activity. A higher epoch +// withdraws the Producer's lower epochs. +message ActivityProducer { + string run_id = 1; + ActivityEpoch epoch = 2; +} + +// Experimental. Might change in the future. +// A client Producer. A higher epoch only fences the lower ones, since a restarted client's earlier +// Records are real messages, not the Records of a failed attempt. A client has no run. +message ClientProducer { + ClientEpoch epoch = 1; +} + +// Experimental. Might change in the future. +// The code that wrote a batch. Its id stays the same across attempts, and the kind's case orders +// its epochs and picks which earlier Records a higher epoch withdraws. A run-based kind carries its +// run, so a client can't. message Producer { // Stable across attempts, such as `wf:` for the Workflow Producer of a run. string id = 1; - // Required. Its case gives the kind of the Producer. - ProducerEpoch epoch = 2; - // The run of the Producer, so nobody parses the id. Empty for a client. - string run_id = 3; + // Required. The case is the kind of the Producer. + oneof kind { + WorkflowProducer workflow = 2; + LocalActivityProducer local_activity = 3; + ActivityProducer activity = 4; + ActivityProducer standalone_activity = 5; + ClientProducer client = 6; + } } // Experimental. Might change in the future. // The Producer of a write, and which earlier Records the write withdraws. An append, a close and // a writer carry the same fields, so the Store orders and withdraws each write the same way. message StreamWriter { - // Required on a write. Its epoch's case picks the withdrawal rule. + // Required on a write. Its Producer's kind picks the withdrawal rule. Producer producer = 1; - // Workflow and Local Activity only. The run's last completed Workflow Task, from a Local - // Activity's start. Its started_event_id is 0 before the first. The write withdraws the run's - // Local Activity Records whose Workflow Task never completed. - WorkflowTaskEpoch previous_completed_task = 2; - // Workflow and Local Activity only. The started event ids of the run's failed or timed out - // Workflow Tasks. The write withdraws the run's Local Activity Records of these tasks. - repeated int64 failed_tasks = 3; - // Retry run only. The previous run, on the retry run's first write. Every Record whose - // Producer carries it is withdrawn. - string withdraw_run = 4; + // Set only by a Workflow or Local Activity write, since only a run withdraws orphaned Local + // Activity Records or a previous run. Unset for an Activity or client write. + RunWithdrawal withdrawal = 2; +} + +// Experimental. Might change in the future. +// What a Workflow or Local Activity write withdraws from its run. One write can carry both the task +// facts and a run to withdraw, so these are fields of one message and not a oneof. +message RunWithdrawal { + // The run's last completed Workflow Task, from a Local Activity's start. Its started_event_id + // is 0 before the first. The write withdraws the run's Local Activity Records whose Workflow + // Task never completed. + WorkflowTaskEpoch previous_completed_task = 1; + // The started event ids of the run's failed or timed out Workflow Tasks. The write withdraws + // the run's Local Activity Records of these tasks. + repeated int64 failed_tasks = 2; + // Retry run only. The previous run, on the retry run's first write. Every Record whose Producer + // carries it is withdrawn. + string withdraw_run = 3; } // Experimental. Might change in the future. @@ -125,8 +150,7 @@ message AppendRecord { // batch of a higher epoch, or for a retry run's first write, and a read builds a Superseded item // from them. A reader drops the earlier Records that the facts match. message StreamSuperseded { - // The writer of the batch that withdraws. Its Producer's epoch is empty when only - // withdraw_run applies. + // The writer of the batch that withdraws, with its withdrawal facts. StreamWriter writer = 1; } diff --git a/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto b/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto index cdd61456e..463f91194 100644 --- a/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto +++ b/crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto @@ -75,7 +75,7 @@ message AppendRecordsRequest { int64 sequence = 4; repeated temporal.api.stream.v1.AppendRecord records = 5; // Only read by the server, on a Workflow Producer's batch. The batch withdraws no Workflow - // Task and keeps no task facts, but the writer's withdraw_run still applies. + // Task and keeps no task facts, but the writer's withdrawal withdraw_run still applies. bool defer_withdrawal = 6; // Only read by the server. The server checks the Producer against the Activity's current // attempt, so nobody else writes as that Activity. diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index 5e97b2ac7..f9fb65566 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -33,9 +33,10 @@ message MemoryStreamConfig { message OpenWriterRequest { string namespace = 1; temporal.api.stream.v1.StreamRef ref = 2; - // Each batch sends it as AppendRecords does. For a Local Activity it has the run's task facts, - // so the writes withdraw the run's orphaned Local Activity Records. Its withdraw_run must be - // empty, since only a retry run's first append withdraws a run, and a writer sends many. + // Each batch sends it as AppendRecords does. For a Local Activity its withdrawal has the run's + // task facts, so the writes withdraw the run's orphaned Local Activity Records. The withdrawal's + // withdraw_run must be empty, since only a retry run's first append withdraws a run, and a + // writer sends many. temporal.api.stream.v1.StreamWriter writer = 3; // A write sends at most this many Records in one batch. Zero, or a value above 1000, is 1000. int32 max_batch_size = 4; From c86fc12a5345b3dfca7638a826832e5a79d19d0c Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 6 Nov 2026 17:59:01 -0700 Subject: [PATCH 13/13] Read the Producer kind and the writer's withdrawal through the oneofs. The protos moved these into variants. The crate reads them through helpers on Producer and StreamWriter, so the stored format, the epoch keys and the test vectors stay the same. --- crates/streams/src/epoch.rs | 127 ++++++++++++------- crates/streams/src/ids.rs | 50 +++++--- crates/streams/src/memory.rs | 14 ++- crates/streams/src/record.rs | 11 +- crates/streams/src/service.rs | 65 +++++----- crates/streams/src/testing/service.rs | 170 +++++++++++++++----------- crates/streams/src/testing/store.rs | 46 +++---- crates/streams/src/testing/support.rs | 23 ++-- crates/streams/src/testing/writer.rs | 57 ++++----- crates/streams/src/writer.rs | 27 ++-- crates/streams/tests/capabilities.rs | 23 ++-- crates/streams/tests/scans.rs | 13 +- crates/streams/tests/vectors.rs | 97 +++++++++------ 13 files changed, 414 insertions(+), 309 deletions(-) diff --git a/crates/streams/src/epoch.rs b/crates/streams/src/epoch.rs index 697d66ef6..15b1ca6a2 100644 --- a/crates/streams/src/epoch.rs +++ b/crates/streams/src/epoch.rs @@ -7,8 +7,8 @@ use crate::{ StreamError, StreamResult, proto::{ - Producer, ProducerEpoch, StreamRecord, StreamSuperseded, WorkflowTaskEpoch, - producer_epoch::Kind, stream_record::Item, + Producer, StreamRecord, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, producer::Kind, + stream_record::Item, }, }; use prost_types::Timestamp; @@ -42,22 +42,52 @@ impl From<&Kind> for ProducerKind { } } -/// Refuses a Producer with no epoch, or with an epoch that has no case. +/// Refuses a Producer with no kind case. impl TryFrom<&Producer> for ProducerKind { type Error = StreamError; fn try_from(producer: &Producer) -> StreamResult { - let epoch = producer - .epoch - .as_ref() - .ok_or_else(|| StreamError::invalid("The Producer has no epoch"))?; - kind(epoch).ok_or_else(|| StreamError::invalid("The Producer's epoch has no kind")) + kind_of(producer).ok_or_else(|| StreamError::invalid("The Producer has no kind")) } } -/// The kind of an epoch's case, `None` when the epoch has no case. -fn kind(epoch: &ProducerEpoch) -> Option { - epoch.kind.as_ref().map(ProducerKind::from) +/// The kind case of a Producer, `None` when it has none. +pub fn kind_of(producer: &Producer) -> Option { + producer.kind.as_ref().map(ProducerKind::from) +} + +/// The run of a Producer. Empty for a client, which has no run. +pub fn producer_run_id(producer: &Producer) -> &str { + match &producer.kind { + Some(Kind::Workflow(w)) => &w.run_id, + Some(Kind::LocalActivity(l)) => &l.run_id, + Some(Kind::Activity(a) | Kind::StandaloneActivity(a)) => &a.run_id, + Some(Kind::Client(_)) | None => "", + } +} + +/// The run that a writer withdraws, empty for none. +pub fn writer_withdraw_run(writer: &StreamWriter) -> &str { + writer + .withdrawal + .as_ref() + .map_or("", |w| w.withdraw_run.as_str()) +} + +/// The started event ids of the run's failed Workflow Tasks that a writer reports. +pub fn writer_failed_tasks(writer: &StreamWriter) -> &[i64] { + writer + .withdrawal + .as_ref() + .map_or(&[], |w| w.failed_tasks.as_slice()) +} + +/// The run's last completed Workflow Task that a writer reports, `None` for none. +pub fn writer_previous_completed_task(writer: &StreamWriter) -> Option<&WorkflowTaskEpoch> { + writer + .withdrawal + .as_ref() + .and_then(|w| w.previous_completed_task.as_ref()) } /// The length of a [task_key]. @@ -92,20 +122,24 @@ pub fn task_key(task: &WorkflowTaskEpoch) -> [u8; TASK_KEY_LEN] { } /// The key of an epoch. It is the [task_of] key, then the attempt's scheduled seconds and nanos -/// and the attempt, in 8, 4 and 4 bytes. A field that the case doesn't have is zero, and an epoch +/// and the attempt, in 8, 4 and 4 bytes. A field that the case doesn't have is zero, and a Producer /// with no case is all zeros. -pub fn epoch_key(epoch: &ProducerEpoch) -> [u8; EPOCH_KEY_LEN] { - let (scheduled, attempt) = match &epoch.kind { - Some(Kind::LocalActivity(local)) => (None, local.attempt), +pub fn epoch_key(kind: Option<&Kind>) -> [u8; EPOCH_KEY_LEN] { + let (scheduled, attempt) = match kind { + Some(Kind::LocalActivity(local)) => (None, local.epoch.map_or(0, |e| e.attempt)), Some(Kind::Activity(activity) | Kind::StandaloneActivity(activity)) => { - (activity.attempt_scheduled_time.as_ref(), activity.attempt) + let epoch = activity.epoch; + ( + epoch.and_then(|e| e.attempt_scheduled_time), + epoch.map_or(0, |e| e.attempt), + ) } - Some(Kind::Client(client)) => (None, client.attempt), + Some(Kind::Client(client)) => (None, client.epoch.map_or(0, |e| e.attempt)), Some(Kind::Workflow(_)) | None => (None, 0), }; - let (seconds, nanos) = time_fields(scheduled); + let (seconds, nanos) = time_fields(scheduled.as_ref()); let mut key = [0; EPOCH_KEY_LEN]; - key[..TASK_KEY_LEN].copy_from_slice(&task_key(&task_of(epoch))); + key[..TASK_KEY_LEN].copy_from_slice(&task_key(&task_of(kind))); key[24..32].copy_from_slice(&be64(seconds)); key[32..36].copy_from_slice(&be32(nanos)); key[36..].copy_from_slice(&be32(attempt)); @@ -114,16 +148,16 @@ pub fn epoch_key(epoch: &ProducerEpoch) -> [u8; EPOCH_KEY_LEN] { /// The Workflow Task of an epoch. It is a Workflow's own task, or the task that dispatched a Local /// Activity, and zero for the other kinds. -pub fn task_of(epoch: &ProducerEpoch) -> WorkflowTaskEpoch { - match &epoch.kind { - Some(Kind::Workflow(task)) => *task, - Some(Kind::LocalActivity(local)) => local.task.unwrap_or_default(), +pub fn task_of(kind: Option<&Kind>) -> WorkflowTaskEpoch { + match kind { + Some(Kind::Workflow(w)) => w.epoch.unwrap_or_default(), + Some(Kind::LocalActivity(local)) => local.epoch.and_then(|e| e.task).unwrap_or_default(), _ => WorkflowTaskEpoch::default(), } } /// Orders two epochs field by field. -pub fn compare_epochs(a: &ProducerEpoch, b: &ProducerEpoch) -> Ordering { +pub fn compare_epochs(a: Option<&Kind>, b: Option<&Kind>) -> Ordering { epoch_key(a).cmp(&epoch_key(b)) } @@ -150,48 +184,52 @@ pub fn withdraws(superseded: &StreamSuperseded, record: &StreamRecord) -> bool { let Some(producer) = data.producer.as_ref() else { return false; }; - if !writer.withdraw_run.is_empty() && producer.run_id == writer.withdraw_run { + let withdraw_run = writer_withdraw_run(writer); + if !withdraw_run.is_empty() && producer_run_id(producer) == withdraw_run { return true; } let Some(withdrawing) = writer.producer.as_ref() else { return false; }; - let (Some(epoch), Some(old)) = (withdrawing.epoch.as_ref(), producer.epoch.as_ref()) else { + let (new, old) = (withdrawing.kind.as_ref(), producer.kind.as_ref()); + let (Some(new_kind), Some(old_kind)) = (new, old) else { return false; }; - match kind(epoch) { - None | Some(ProducerKind::Client) => return false, - Some(ProducerKind::Workflow) => {} + match ProducerKind::from(new_kind) { + ProducerKind::Client => return false, + ProducerKind::Workflow => {} same => { - return kind(old) == same + return ProducerKind::from(old_kind) == same && producer.id == withdrawing.id - && compare_epochs(old, epoch) == Ordering::Less; + && compare_epochs(old, new) == Ordering::Less; } } if !of_run(producer, withdrawing) { return false; } - let (task, old) = (task_of(epoch), task_of(old)); - let started = old.started_event_id; - if writer.failed_tasks.contains(&started) { + let (task, old_task) = (task_of(new), task_of(old)); + let started = old_task.started_event_id; + if writer_failed_tasks(writer).contains(&started) { return true; } - let completed = writer.previous_completed_task.as_ref(); + let completed = writer_previous_completed_task(writer); let after = completed.map_or(0, |c| c.started_event_id); if after < started && started < task.started_event_id { return true; } let lower_at = |at: &WorkflowTaskEpoch| { - started == at.started_event_id && compare_tasks(&old, at) == Ordering::Less + started == at.started_event_id && compare_tasks(&old_task, at) == Ordering::Less }; lower_at(&task) || completed.is_some_and(lower_at) } /// Whether a Producer is the Workflow run that withdraws, or one of its Local Activities. fn of_run(producer: &Producer, withdrawing: &Producer) -> bool { - match producer.epoch.as_ref().and_then(kind) { + match kind_of(producer) { Some(ProducerKind::Workflow) => producer.id == withdrawing.id, - Some(ProducerKind::LocalActivity) => producer.run_id == withdrawing.run_id, + Some(ProducerKind::LocalActivity) => { + producer_run_id(producer) == producer_run_id(withdrawing) + } _ => false, } } @@ -199,17 +237,14 @@ fn of_run(producer: &Producer, withdrawing: &Producer) -> bool { #[cfg(test)] mod tests { use super::*; - use crate::proto::ClientEpoch; + use crate::proto::{ClientEpoch, ClientProducer, producer::Kind}; // Without the clamp, the bytes of a negative field would outrank every real epoch. #[test] fn a_negative_field_compares_as_zero() { - let negative = ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt: -1 })), - }; - assert_eq!( - compare_epochs(&negative, &ProducerEpoch::default()), - Ordering::Equal - ); + let negative = Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt: -1 }), + }); + assert_eq!(compare_epochs(Some(&negative), None), Ordering::Equal); } } diff --git a/crates/streams/src/ids.rs b/crates/streams/src/ids.rs index d50cf47f3..45eb1e77b 100644 --- a/crates/streams/src/ids.rs +++ b/crates/streams/src/ids.rs @@ -11,7 +11,11 @@ //! script. Each part is percent-encoded, so a `:` or a brace in an id can't make two Streams share //! a key. -use crate::{StreamError, StreamResult, epoch::ProducerKind, proto::Producer}; +use crate::{ + StreamError, StreamResult, + epoch::{ProducerKind, producer_run_id}, + proto::Producer, +}; /// The longest Stream id, in bytes. pub const MAX_STREAM_ID_BYTES: usize = 4096; @@ -151,13 +155,14 @@ pub fn check_producer(producer: &Producer) -> StreamResult<()> { kind, ProducerKind::Workflow | ProducerKind::LocalActivity | ProducerKind::Activity ); - if runs && decode_part(parts[0]).as_deref() != Some(producer.run_id.as_str()) { + if runs && decode_part(parts[0]).as_deref() != Some(producer_run_id(producer)) { return Err(StreamError::invalid(format!( "The Producer id {:?} names another run than its run id {:?}", - producer.id, producer.run_id + producer.id, + producer_run_id(producer) ))); } - if kind == ProducerKind::Client && !producer.run_id.is_empty() { + if kind == ProducerKind::Client && !producer_run_id(producer).is_empty() { return Err(StreamError::invalid("A client Producer has no run id")); } Ok(()) @@ -207,20 +212,34 @@ impl StreamKeys { #[cfg(test)] mod tests { use super::*; - use crate::proto::{ProducerEpoch, producer_epoch::Kind}; + use crate::proto::{ + ActivityProducer, ClientProducer, LocalActivityProducer, WorkflowProducer, producer::Kind, + }; fn producer(kind: ProducerKind, id: &str, run_id: &str) -> Producer { + let run_id = run_id.to_string(); let kind = match kind { - ProducerKind::Workflow => Kind::Workflow(Default::default()), - ProducerKind::LocalActivity => Kind::LocalActivity(Default::default()), - ProducerKind::Activity => Kind::Activity(Default::default()), - ProducerKind::StandaloneActivity => Kind::StandaloneActivity(Default::default()), - ProducerKind::Client => Kind::Client(Default::default()), + ProducerKind::Workflow => Kind::Workflow(WorkflowProducer { + run_id, + ..Default::default() + }), + ProducerKind::LocalActivity => Kind::LocalActivity(LocalActivityProducer { + run_id, + ..Default::default() + }), + ProducerKind::Activity => Kind::Activity(ActivityProducer { + run_id, + ..Default::default() + }), + ProducerKind::StandaloneActivity => Kind::StandaloneActivity(ActivityProducer { + run_id, + ..Default::default() + }), + ProducerKind::Client => Kind::Client(ClientProducer::default()), }; Producer { id: id.to_string(), - epoch: Some(ProducerEpoch { kind: Some(kind) }), - run_id: run_id.to_string(), + kind: Some(kind), } } @@ -267,14 +286,9 @@ mod tests { producer(ProducerKind::Workflow, &workflow_producer_id(run), "other"), producer(ProducerKind::Activity, "act:run%3A1:a", run), Producer { - epoch: None, - ..producer(ProducerKind::Workflow, "wf:x", "x") - }, - Producer { - epoch: Some(ProducerEpoch::default()), + kind: None, ..producer(ProducerKind::Workflow, "wf:x", "x") }, - producer(ProducerKind::Client, "client:a:b", "run"), producer(ProducerKind::Client, "client:a|b", ""), producer(ProducerKind::Client, "client:run%3a1", ""), ]; diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index 589334a59..de076e762 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -5,7 +5,9 @@ use crate::{ Reason, StreamError, StreamResult, digest::DIGEST_LEN, - epoch::{EPOCH_KEY_LEN, ProducerKind, TASK_KEY_LEN, epoch_key, task_key, task_of}, + epoch::{ + EPOCH_KEY_LEN, ProducerKind, TASK_KEY_LEN, epoch_key, producer_run_id, task_key, task_of, + }, ids::{StreamKeys, encode_part}, offset::{entry_id_of_offset, offset_of_entry_id}, proto::{Producer, StreamRef, WorkflowTaskEpoch}, @@ -693,8 +695,8 @@ impl Store for MemoryStore { let stream_id = stream_ref.id.as_str(); let producer = batch.producer; let kind = ProducerKind::try_from(producer)?; - let epoch = epoch_key(&producer.epoch.unwrap_or_default()); - let run = encode_run(&producer.run_id); + let epoch = epoch_key(producer.kind.as_ref()); + let run = encode_run(producer_run_id(producer)); let withdrawal = &batch.withdrawal; let withdrawn_run = encode_run(withdrawal.withdraw_run); let (safety, tombstone) = (millis(self.safety_ttl), millis(TOMBSTONE)); @@ -702,7 +704,7 @@ impl Store for MemoryStore { let member = format!("{}|{}", kind as i32, producer.id); // The Workflow Task of a Local Activity of a run, which the run's task facts can withdraw. let task = (kind == ProducerKind::LocalActivity && !run.is_empty()) - .then(|| task_key(&task_of(&producer.epoch.unwrap_or_default()))); + .then(|| task_key(&task_of(producer.kind.as_ref()))); let facts = withdrawal.tasks.filter(|_| task.is_some()); let completed = facts.and_then(|facts| facts.completed).map(task_key); let open_field = format!("oa|{run}"); @@ -944,8 +946,8 @@ impl Store for MemoryStore { ) -> StreamResult { let stream_id = stream_ref.id.as_str(); let kind = ProducerKind::try_from(producer)?; - let epoch = epoch_key(&producer.epoch.unwrap_or_default()); - let run = encode_run(&producer.run_id); + let epoch = epoch_key(producer.kind.as_ref()); + let run = encode_run(producer_run_id(producer)); let field = format!("p|{}|{}", kind as i32, producer.id); let tombstone = millis(TOMBSTONE); let chain = owner_chain(stream_ref); diff --git a/crates/streams/src/record.rs b/crates/streams/src/record.rs index 85d32e1f4..2277adf9b 100644 --- a/crates/streams/src/record.rs +++ b/crates/streams/src/record.rs @@ -3,8 +3,8 @@ use crate::{ Reason, StreamError, StreamResult, proto::{ - AppendRecord, Producer, StreamData, StreamRecord, StreamSuperseded, StreamWriter, - stream_record::Item, + AppendRecord, Producer, RunWithdrawal, StreamData, StreamRecord, StreamSuperseded, + StreamWriter, stream_record::Item, }, store::{MAX_BATCH_BYTES, MAX_BATCH_RECORDS, MAX_RECORD_BYTES}, }; @@ -65,11 +65,14 @@ pub(crate) fn data_records( /// The facts a `Superseded` item of this batch gives the Reader. pub(crate) fn superseded(producer: &Producer, withdraw_run: &str) -> Vec { + let withdrawal = (!withdraw_run.is_empty()).then(|| RunWithdrawal { + withdraw_run: withdraw_run.to_string(), + ..Default::default() + }); StreamSuperseded { writer: Some(StreamWriter { producer: Some(producer.clone()), - withdraw_run: withdraw_run.to_string(), - ..Default::default() + withdrawal, }), } .encode_to_vec() diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index b617e00e4..b074cbe53 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -4,15 +4,18 @@ use crate::{ Reason, StreamError, StreamResult, digest::batch_digest, - epoch::{ProducerKind, task_of, withdraws}, + epoch::{ + ProducerKind, producer_run_id, task_of, withdraws, writer_failed_tasks, + writer_previous_completed_task, writer_withdraw_run, + }, ids::{check_producer, check_stream_id}, metrics::Metrics, proto::{ AppendRecordsRequest, AppendRecordsResponse, CloseStreamRequest, CloseStreamResponse, CreateStreamRequest, CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse, - Producer, ProducerEpoch, ReadRecordsRequest, ReadRecordsResponse, StreamIdConflictPolicy, + Producer, ReadRecordsRequest, ReadRecordsResponse, RunWithdrawal, StreamIdConflictPolicy, StreamReadStart, StreamRecord, StreamRef, StreamStatus, StreamSuperseded, StreamWriter, - WorkflowTaskEpoch, producer_epoch::Kind, stream_record::Item, + WorkflowProducer, WorkflowTaskEpoch, producer::Kind, stream_record::Item, }, record::{data_records, superseded}, store::{ @@ -108,7 +111,7 @@ fn stream_id(stream_ref: Option<&StreamRef>) -> StreamResult<&str> { /// so a Reader withdraws only that run's Records. fn run_superseded(owner: Option<&Producer>, withdraw_run: &str) -> Vec { let producer = Producer { - epoch: None, + kind: None, ..owner.cloned().unwrap_or_default() }; superseded(&producer, withdraw_run) @@ -121,20 +124,23 @@ fn orphan_superseded( completed: Option<&WorkflowTaskEpoch>, failed_tasks: &[i64], ) -> Vec { - let task = producer.epoch.as_ref().map(task_of).unwrap_or_default(); + let run_id = producer_run_id(producer); + let task = task_of(producer.kind.as_ref()); let workflow = Producer { - id: crate::ids::workflow_producer_id(&producer.run_id), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Workflow(task)), - }), - run_id: producer.run_id.clone(), + id: crate::ids::workflow_producer_id(run_id), + kind: Some(Kind::Workflow(WorkflowProducer { + run_id: run_id.to_string(), + epoch: Some(task), + })), }; StreamSuperseded { writer: Some(StreamWriter { producer: Some(workflow), - previous_completed_task: completed.copied(), - failed_tasks: failed_tasks.to_vec(), - ..Default::default() + withdrawal: Some(RunWithdrawal { + previous_completed_task: completed.copied(), + failed_tasks: failed_tasks.to_vec(), + withdraw_run: String::new(), + }), }), } .encode_to_vec() @@ -151,9 +157,7 @@ fn outside_producer(producer: Option<&Producer>) -> StreamResult<&Producer> { )); } // Without its task, a re-dispatch in a later task has the same epoch as the first dispatch. - let task = producer - .epoch - .map_or(0, |epoch| task_of(&epoch).started_event_id); + let task = task_of(producer.kind.as_ref()).started_event_id; if kind == ProducerKind::LocalActivity && task <= 0 { return Err(StreamError::invalid( "A Local Activity Producer's epoch must name the Workflow Task that dispatched it", @@ -165,8 +169,8 @@ fn outside_producer(producer: Option<&Producer>) -> StreamResult<&Producer> { /// Refuses task facts that a Store would drop or misread. Only a Local Activity has its run's task /// facts, and a failed task is a started event id, which is never zero or negative. fn check_task_facts(kind: ProducerKind, writer: &StreamWriter) -> StreamResult<()> { - let completed = writer.previous_completed_task.as_ref(); - if completed.is_none() && writer.failed_tasks.is_empty() { + let completed = writer_previous_completed_task(writer); + if completed.is_none() && writer_failed_tasks(writer).is_empty() { return Ok(()); } if kind != ProducerKind::LocalActivity { @@ -180,7 +184,7 @@ fn check_task_facts(kind: ProducerKind, writer: &StreamWriter) -> StreamResult<( )); } if completed.is_some_and(|c| c.started_event_id < 0) - || writer.failed_tasks.iter().any(|&id| id <= 0) + || writer_failed_tasks(writer).iter().any(|&id| id <= 0) { return Err(StreamError::invalid( "A task's started event id must be positive", @@ -243,7 +247,7 @@ impl StreamService { "A create with withdraw_run needs the owner run", )); }; - if request.withdraw_run == owner.run_id { + if request.withdraw_run == producer_run_id(owner) { return Err(StreamError::invalid("A run doesn't withdraw itself")); } self.store.capabilities().require(Capabilities::RUN_FENCE)?; @@ -302,10 +306,11 @@ impl StreamService { .capabilities() .require(Capabilities::NEXT_SEQUENCE)?; } - if !writer.withdraw_run.is_empty() { + let withdraw_run = writer_withdraw_run(&writer); + if !withdraw_run.is_empty() { // Only a retry run's Local Activity withdraws its previous run, as only the owner's // create does. Any other Producer could withdraw and fence an unrelated run. - if kind != ProducerKind::LocalActivity || writer.withdraw_run == producer.run_id { + if kind != ProducerKind::LocalActivity || withdraw_run == producer_run_id(producer) { return Err(StreamError::invalid( "Only a Local Activity of a later run withdraws a run", )); @@ -315,26 +320,24 @@ impl StreamService { check_task_facts(kind, &writer)?; let digest = batch_digest(&request.records); let records = data_records(producer, request.records)?; - let bytes = superseded(producer, &writer.withdraw_run); + let bytes = superseded(producer, withdraw_run); // A Local Activity sends its run's task facts, so that the batch withdraws the run's // orphaned Local Activity Records. These are the Records of the tasks that never completed. - let completed = writer - .previous_completed_task - .filter(|c| c.started_event_id > 0); + let completed = writer_previous_completed_task(&writer).filter(|c| c.started_event_id > 0); let orphans = (kind == ProducerKind::LocalActivity - && writer.previous_completed_task.is_some()) - .then(|| orphan_superseded(producer, completed.as_ref(), &writer.failed_tasks)); + && writer_previous_completed_task(&writer).is_some()) + .then(|| orphan_superseded(producer, completed, writer_failed_tasks(&writer))); let batch = Batch { producer, sequence: request.sequence, records: &records, digest, withdrawal: Withdrawal { - withdraw_run: &writer.withdraw_run, + withdraw_run, superseded: &bytes, tasks: orphans.as_ref().map(|orphans| TaskFacts { - completed: completed.as_ref(), - failed_tasks: &writer.failed_tasks, + completed, + failed_tasks: writer_failed_tasks(&writer), superseded: orphans, }), }, diff --git a/crates/streams/src/testing/service.rs b/crates/streams/src/testing/service.rs index 11191fe87..7cc54d9fe 100644 --- a/crates/streams/src/testing/service.rs +++ b/crates/streams/src/testing/service.rs @@ -6,15 +6,16 @@ use super::{case, support::Needs}; use crate::{ StreamError, StreamRuntime, - epoch::withdraws, + epoch::{withdraws, writer_withdraw_run}, ids::{activity_producer_id, client_producer_id, named_client_producer_id}, proto::{ - ActivityEpoch, AppendRecord, AppendRecordsRequest, AppendRecordsResponse, ClientEpoch, - CloseStreamRequest, CloseStreamResponse, CreateStreamRequest, CreateStreamResponse, - DescribeStreamRequest, DescribeStreamResponse, LocalActivityEpoch, Producer, ProducerEpoch, - ReadRecordsRequest, ReadRecordsResponse, StreamErrorDetails, StreamIdConflictPolicy, - StreamReadStart, StreamRecord, StreamRef, StreamStatus, StreamSuperseded, StreamWriter, - WorkflowTaskEpoch, producer_epoch::Kind, stream_record::Item, + ActivityEpoch, ActivityProducer, AppendRecord, AppendRecordsRequest, AppendRecordsResponse, + ClientEpoch, ClientProducer, CloseStreamRequest, CloseStreamResponse, CreateStreamRequest, + CreateStreamResponse, DescribeStreamRequest, DescribeStreamResponse, LocalActivityEpoch, + LocalActivityProducer, Producer, ReadRecordsRequest, ReadRecordsResponse, RunWithdrawal, + StreamErrorDetails, StreamIdConflictPolicy, StreamReadStart, StreamRecord, StreamRef, + StreamStatus, StreamSuperseded, StreamWriter, WorkflowProducer, WorkflowTaskEpoch, + producer::Kind, stream_record::Item, }, }; use prost::Message; @@ -88,10 +89,9 @@ fn writer_for(producer: Producer) -> Option { fn client(attempt: i32) -> Producer { Producer { id: client_producer_id("me", "p"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt })), - }), - run_id: String::new(), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt }), + })), } } @@ -108,16 +108,16 @@ fn named(name: &str, attempt: i32) -> Producer { fn activity(attempt: i32, scheduled: i64) -> Producer { Producer { id: activity_producer_id("r1", "a", "s"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Activity(ActivityEpoch { + kind: Some(Kind::Activity(ActivityProducer { + run_id: "r1".to_string(), + epoch: Some(ActivityEpoch { attempt_scheduled_time: Some(Timestamp { seconds: scheduled, nanos: 0, }), attempt, - })), - }), - run_id: "r1".to_string(), + }), + })), } } @@ -125,10 +125,10 @@ fn activity(attempt: i32, scheduled: i64) -> Producer { fn workflow(run: &str) -> Producer { Producer { id: crate::ids::workflow_producer_id(run), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Workflow(WorkflowTaskEpoch::default())), - }), - run_id: run.to_string(), + kind: Some(Kind::Workflow(WorkflowProducer { + run_id: run.to_string(), + epoch: Some(WorkflowTaskEpoch::default()), + })), } } @@ -388,13 +388,13 @@ async fn a_workflow_or_a_client_closes_a_stream_with_no_owner() { let run = format!("{}-run", case.stream); let activity = Producer { id: activity_producer_id(&run, "a", "s"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Activity(ActivityEpoch { + kind: Some(Kind::Activity(ActivityProducer { + run_id: run.clone(), + epoch: Some(ActivityEpoch { attempt: 1, ..Default::default() - })), - }), - run_id: run.clone(), + }), + })), }; let error = case.close(activity).await.unwrap_err(); assert_eq!( @@ -432,13 +432,13 @@ async fn a_create_checks_its_owner_and_its_ref() { }; let local_activity = Producer { id: crate::ids::local_activity_producer_id(&run, "call-1"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::LocalActivity(LocalActivityEpoch { + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: run.clone(), + epoch: Some(LocalActivityEpoch { task: Some(task), attempt: 1, - })), - }), - ..workflow.clone() + }), + })), }; let chain = |workflow_id: &str| StreamRef { id: case.stream.clone(), @@ -500,11 +500,12 @@ async fn a_create_checks_its_owner_and_its_ref() { async fn a_producer_must_match_its_kind_and_not_be_a_workflow() { for case in cases_for(Needs::CALLS).await { case.create(None).await.unwrap(); - let activity = ProducerEpoch { - kind: Some(Kind::Activity(ActivityEpoch::default())), - }; - for epoch in [Some(activity), None, Some(ProducerEpoch::default())] { - let wrong = Producer { epoch, ..client(1) }; + let activity = Kind::Activity(ActivityProducer { + run_id: String::new(), + epoch: Some(ActivityEpoch::default()), + }); + for kind in [Some(activity), None] { + let wrong = Producer { kind, ..client(1) }; let error = case.append(wrong, 1, vec![record("x")]).await.unwrap_err(); assert_eq!( (error.code(), reason(&error).as_str()), @@ -515,13 +516,13 @@ async fn a_producer_must_match_its_kind_and_not_be_a_workflow() { assert_eq!(error.code(), Code::InvalidArgument); let no_task = Producer { id: crate::ids::local_activity_producer_id("r1", "call-1"), - run_id: "r1".to_string(), - epoch: Some(ProducerEpoch { - kind: Some(Kind::LocalActivity(LocalActivityEpoch { + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: "r1".to_string(), + epoch: Some(LocalActivityEpoch { task: None, attempt: 1, - })), - }), + }), + })), }; let error = case .append(no_task, 1, vec![record("x")]) @@ -1044,8 +1045,10 @@ async fn a_retry_run_s_local_activity_calls_withdraw_the_previous_run() { r#ref: case.stream_ref(), writer: Some(StreamWriter { producer: Some(local_activity(run)), - withdraw_run: withdraw_run.to_string(), - ..Default::default() + withdrawal: Some(RunWithdrawal { + withdraw_run: withdraw_run.to_string(), + ..Default::default() + }), }), sequence: 1, records: vec![record(text)], @@ -1080,11 +1083,12 @@ async fn a_retry_run_s_local_activity_calls_withdraw_the_previous_run() { assert_eq!(texts(&read), ["old", "", "new"]); let superseded = superseded_of(&read.records[1]); let writer = superseded.writer.as_ref().unwrap(); - assert_eq!(writer.withdraw_run, previous); - // The item's writer has its Producer, which the WIT requires. + assert_eq!(writer_withdraw_run(writer), previous); + // The item's writer has its Producer, which the WIT requires. A create's item has no kind, + // so a Reader can't read it as an epoch, and the run stays in the id. let producer = writer.producer.as_ref().unwrap(); assert_eq!(producer.id, crate::ids::workflow_producer_id(&retry)); - assert_eq!(producer.run_id, retry); + assert!(producer.kind.is_none()); let error = call::<_, CreateStreamResponse>( &case.ext, "CreateStream", @@ -1124,8 +1128,10 @@ async fn only_a_later_run_s_local_activity_withdraws_a_run_on_append() { r#ref: case.stream_ref(), writer: Some(StreamWriter { producer: Some(producer), - withdraw_run: run.clone(), - ..Default::default() + withdrawal: Some(RunWithdrawal { + withdraw_run: run.clone(), + ..Default::default() + }), }), sequence: 1, records: vec![record("other")], @@ -1153,13 +1159,13 @@ fn user_local_activity(run: &str, started: i64, seconds: i64) -> Producer { }; Producer { id: crate::ids::local_activity_producer_id(run, "call-1"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::LocalActivity(LocalActivityEpoch { + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: run.to_string(), + epoch: Some(LocalActivityEpoch { task: Some(task), attempt: 1, - })), - }), - run_id: run.to_string(), + }), + })), } } @@ -1242,11 +1248,13 @@ fn local_activity_write( r#ref: case.stream_ref(), writer: Some(StreamWriter { producer: Some(producer), - previous_completed_task: Some(WorkflowTaskEpoch { - started_event_id: completed, + withdrawal: Some(RunWithdrawal { + previous_completed_task: Some(WorkflowTaskEpoch { + started_event_id: completed, + ..Default::default() + }), ..Default::default() }), - ..Default::default() }), sequence: 1, records: vec![record(text)], @@ -1254,11 +1262,18 @@ fn local_activity_write( } } -/// The writer of a write, to change its Producer or its task facts. +/// The writer of a write, to change its Producer. fn writer_mut(request: &mut AppendRecordsRequest) -> &mut StreamWriter { request.writer.get_or_insert_with(Default::default) } +/// The withdrawal facts of a write, to change its task facts. +fn withdrawal_mut(request: &mut AppendRecordsRequest) -> &mut RunWithdrawal { + writer_mut(request) + .withdrawal + .get_or_insert_with(Default::default) +} + /// The close Activity closes a Workflow's Stream as the run's Workflow Producer, which the service /// refuses on an append. async fn the_close_activity_closes_a_workflow_stream_as_its_workflow_producer() { @@ -1374,7 +1389,7 @@ async fn a_failed_task_before_the_last_completed_one_is_withdrawn_by_its_id() { .unwrap(); // Task [3] failed and task [4] completed with no write here. Task [5] knows both. let mut next = local_activity_write(&case, &run, "call-2", 5, 4, "next"); - writer_mut(&mut next).failed_tasks = vec![3]; + withdrawal_mut(&mut next).failed_tasks = vec![3]; let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", next).await.unwrap(); let read = case.read(ReadRecordsRequest::default()).await.unwrap(); assert_eq!(texts(&read), ["orphan", "", "next"]); @@ -1399,7 +1414,7 @@ async fn a_first_write_of_a_task_that_never_completed_is_fenced() { .unwrap(); // Task [3] runs call-2 and fails before call-2 writes. Task [4] knows that [3] failed. let mut retry = local_activity_write(&case, &run, "call-3", 4, 2, "retry"); - writer_mut(&mut retry).failed_tasks = vec![3]; + withdrawal_mut(&mut retry).failed_tasks = vec![3]; let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", retry).await.unwrap(); // Call-1 of the completed task [2] still runs and writes with its older facts. let more = AppendRecordsRequest { @@ -1428,11 +1443,11 @@ async fn a_write_with_bad_task_facts_fails() { case.create(None).await.unwrap(); let facts = |failed: Vec| { let mut request = local_activity_write(&case, &run, "call-1", 4, 2, "bad"); - writer_mut(&mut request).failed_tasks = failed; + withdrawal_mut(&mut request).failed_tasks = failed; request }; let mut no_completed = facts(vec![3]); - writer_mut(&mut no_completed).previous_completed_task = None; + withdrawal_mut(&mut no_completed).previous_completed_task = None; let mut by_client = facts(vec![]); writer_mut(&mut by_client).producer = Some(client(1)); let mut by_activity = facts(vec![3]); @@ -1461,10 +1476,10 @@ async fn a_write_with_no_task_facts_is_withdrawn_by_a_later_write() { let run = format!("{}-run", case.stream); case.create(None).await.unwrap(); let mut orphan = local_activity_write(&case, &run, "call-1", 3, 2, "orphan"); - writer_mut(&mut orphan).previous_completed_task = None; + withdrawal_mut(&mut orphan).previous_completed_task = None; let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", orphan).await.unwrap(); let mut next = local_activity_write(&case, &run, "call-2", 5, 4, "next"); - writer_mut(&mut next).failed_tasks = vec![3]; + withdrawal_mut(&mut next).failed_tasks = vec![3]; let _: AppendRecordsResponse = call(&case.ext, "AppendRecords", next).await.unwrap(); let read = case.read(ReadRecordsRequest::default()).await.unwrap(); assert_eq!(texts(&read), ["orphan", "", "next"]); @@ -1488,7 +1503,7 @@ async fn a_batch_writes_an_orphan_item_and_an_epoch_item() { .await .unwrap(); let mut rerun = local_activity_write(&case, &run, "call-1", 5, 4, "rerun"); - writer_mut(&mut rerun).failed_tasks = vec![3]; + withdrawal_mut(&mut rerun).failed_tasks = vec![3]; let answer: AppendRecordsResponse = call(&case.ext, "AppendRecords", rerun).await.unwrap(); let read = case.read(ReadRecordsRequest::default()).await.unwrap(); assert_eq!( @@ -1499,9 +1514,16 @@ async fn a_batch_writes_an_orphan_item_and_an_epoch_item() { superseded_of(&read.records[1]), superseded_of(&read.records[2]), ); - let kind = |item: &StreamSuperseded| { - let producer = item.writer.as_ref()?.producer.as_ref()?; - producer.epoch?.kind + let kind = |item: &StreamSuperseded| -> Option { + Some( + item.writer + .as_ref()? + .producer + .as_ref()? + .kind + .as_ref()? + .clone(), + ) }; assert!(matches!(kind(orphans), Some(Kind::Workflow(_)))); assert!(matches!(kind(epoch), Some(Kind::LocalActivity(_)))); @@ -1569,21 +1591,23 @@ fn task_write( ) -> AppendRecordsRequest { let producer = Producer { id: crate::ids::local_activity_producer_id(run, call), - epoch: Some(ProducerEpoch { - kind: Some(Kind::LocalActivity(LocalActivityEpoch { + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: run.to_string(), + epoch: Some(LocalActivityEpoch { task: Some(task), attempt: 1, - })), - }), - run_id: run.to_string(), + }), + })), }; AppendRecordsRequest { namespace: case.namespace.clone(), r#ref: case.stream_ref(), writer: Some(StreamWriter { producer: Some(producer), - previous_completed_task: completed, - ..Default::default() + withdrawal: Some(RunWithdrawal { + previous_completed_task: completed, + ..Default::default() + }), }), sequence: 1, records: vec![record(text)], diff --git a/crates/streams/src/testing/store.rs b/crates/streams/src/testing/store.rs index 5715a42f2..9221007d4 100644 --- a/crates/streams/src/testing/store.rs +++ b/crates/streams/src/testing/store.rs @@ -18,8 +18,8 @@ use crate::{ memory::{EntryId, ProducerField}, offset::{entry_id_of_offset, offset_of_entry_id}, proto::{ - ActivityEpoch, ClientEpoch, Producer, ProducerEpoch, StreamRef, WorkflowTaskEpoch, - producer_epoch::Kind, + ActivityEpoch, ActivityProducer, ClientEpoch, ClientProducer, Producer, StreamRef, + WorkflowProducer, WorkflowTaskEpoch, producer::Kind, }, store::{Batch, ConflictPolicy, Create, CreateOutcome, Created, Owner, TaskFacts, Withdrawal}, }; @@ -30,26 +30,25 @@ use std::time::Duration; fn activity(run: &str, attempt: i32, scheduled: i64) -> Producer { Producer { id: activity_producer_id(run, "a", "s"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Activity(ActivityEpoch { + kind: Some(Kind::Activity(ActivityProducer { + run_id: run.to_string(), + epoch: Some(ActivityEpoch { attempt_scheduled_time: Some(Timestamp { seconds: scheduled, nanos: 0, }), attempt, - })), - }), - run_id: run.to_string(), + }), + })), } } fn client(attempt: i32) -> Producer { Producer { id: client_producer_id("me", "p1"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt })), - }), - run_id: String::new(), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt }), + })), } } @@ -774,10 +773,10 @@ async fn a_stale_producer_cant_close_the_stream() { fn workflow(run: &str) -> Producer { Producer { id: workflow_producer_id(run), - run_id: run.to_string(), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Workflow(WorkflowTaskEpoch::default())), - }), + kind: Some(Kind::Workflow(WorkflowProducer { + run_id: run.to_string(), + epoch: Some(WorkflowTaskEpoch::default()), + })), } } @@ -800,13 +799,14 @@ async fn a_workflow_or_a_client_closes_a_stream_with_no_owner() { async fn a_producer_with_no_epoch_case_is_refused() { for target in stores_for(Needs::CALLS).await { target.create(Owner::Client).await; - for epoch in [None, Some(ProducerEpoch::default())] { - let producer = Producer { epoch, ..client(1) }; - let error = append(&target, &producer, 1, &["a"]).await.unwrap_err(); - assert_eq!(error.reason, Reason::Invalid); - let error = target.close(&producer).await.unwrap_err(); - assert_eq!(error.reason, Reason::Invalid); - } + let producer = Producer { + kind: None, + ..client(1) + }; + let error = append(&target, &producer, 1, &["a"]).await.unwrap_err(); + assert_eq!(error.reason, Reason::Invalid); + let error = target.close(&producer).await.unwrap_err(); + assert_eq!(error.reason, Reason::Invalid); assert!(target.texts().await.is_empty()); } } @@ -977,7 +977,7 @@ async fn a_producer_field_has_the_shared_layout() { target.create(Owner::Workflow).await; let kind = ProducerKind::Activity as i32; let field = |producer: &Producer, held: Option<&Producer>, offsets, texts: &[&str]| { - let epoch = |producer: &Producer| epoch_key(&producer.epoch.unwrap_or_default()); + let epoch = |producer: &Producer| epoch_key(producer.kind.as_ref()); ProducerField { fence: epoch(producer), held: held.map(epoch), diff --git a/crates/streams/src/testing/support.rs b/crates/streams/src/testing/support.rs index c0822051a..3e035a932 100644 --- a/crates/streams/src/testing/support.rs +++ b/crates/streams/src/testing/support.rs @@ -10,9 +10,9 @@ use crate::{ epoch::withdraws, ids::local_activity_producer_id, proto::{ - AppendRecord, LocalActivityEpoch, Producer, ProducerEpoch, StreamData, StreamRecord, - StreamRef, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, producer_epoch::Kind, - stream_record::Item, + AppendRecord, LocalActivityEpoch, LocalActivityProducer, Producer, RunWithdrawal, + StreamData, StreamRecord, StreamRef, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, + producer::Kind, stream_record::Item, }, store::{Appended, Batch, Closed, Create, Created, Entry, Owner, Store, Withdrawal}, }; @@ -132,13 +132,13 @@ impl Run { }; Producer { id: local_activity_producer_id(&self.id, call), - epoch: Some(ProducerEpoch { - kind: Some(Kind::LocalActivity(LocalActivityEpoch { + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: self.id.clone(), + epoch: Some(LocalActivityEpoch { task: Some(task), attempt, - })), - }), - run_id: self.id.clone(), + }), + })), } } } @@ -193,11 +193,14 @@ pub(crate) fn digest(texts: &[&str]) -> [u8; crate::digest::DIGEST_LEN] { } pub(crate) fn superseded(producer: &Producer, withdrawal: &Withdrawal<'_>) -> Vec { + let run = (!withdrawal.withdraw_run.is_empty()).then(|| RunWithdrawal { + withdraw_run: withdrawal.withdraw_run.to_string(), + ..Default::default() + }); StreamSuperseded { writer: Some(StreamWriter { producer: Some(producer.clone()), - withdraw_run: withdrawal.withdraw_run.to_string(), - ..Default::default() + withdrawal: run, }), } .encode_to_vec() diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs index 862fed106..c509f688d 100644 --- a/crates/streams/src/testing/writer.rs +++ b/crates/streams/src/testing/writer.rs @@ -14,12 +14,12 @@ use crate::{ named_client_producer_id, }, proto::{ - ActivityEpoch, AppendRecord, ClientEpoch, CloseStreamRequest, CloseWriterRequest, - CloseWriterResponse, CreateStreamRequest, CreateStreamResponse, LocalActivityEpoch, - OpenWriterRequest, OpenWriterResponse, Producer, ProducerEpoch, ReadRecordsRequest, - ReadRecordsResponse, StreamErrorDetails, StreamRecord, StreamRef, StreamWriter, - WorkflowTaskEpoch, WriteRecordsRequest, WriteRecordsResponse, producer_epoch::Kind, - stream_record::Item, + ActivityEpoch, ActivityProducer, AppendRecord, ClientEpoch, ClientProducer, + CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, CreateStreamRequest, + CreateStreamResponse, LocalActivityEpoch, LocalActivityProducer, OpenWriterRequest, + OpenWriterResponse, Producer, ReadRecordsRequest, ReadRecordsResponse, RunWithdrawal, + StreamErrorDetails, StreamRecord, StreamRef, StreamWriter, WorkflowTaskEpoch, + WriteRecordsRequest, WriteRecordsResponse, producer::Kind, stream_record::Item, }, store::{ Appended, Batch, Capabilities, Closed, Create, Created, Entry, MAX_BATCH_RECORDS, Store, @@ -90,26 +90,25 @@ fn writer_for(producer: Producer) -> Option { fn client() -> Producer { Producer { id: client_producer_id("me", "p"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt: 1 })), - }), - run_id: String::new(), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt: 1 }), + })), } } fn activity(attempt: i32) -> Producer { Producer { id: activity_producer_id("r1", "a", "s"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Activity(ActivityEpoch { + kind: Some(Kind::Activity(ActivityProducer { + run_id: "r1".to_string(), + epoch: Some(ActivityEpoch { attempt_scheduled_time: Some(Timestamp { seconds: 10 * attempt as i64, nanos: 0, }), attempt, - })), - }), - run_id: "r1".to_string(), + }), + })), } } @@ -422,10 +421,9 @@ async fn a_named_client_with_a_higher_epoch_fences_its_earlier_self() { for case in cases().await { let named = |attempt| Producer { id: named_client_producer_id("ingest-1"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt })), - }), - run_id: String::new(), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt }), + })), }; let old = case.open(named(1), |_| {}).await; case.write(old, &["old"]).await.unwrap(); @@ -656,8 +654,10 @@ async fn a_writer_with_a_run_to_withdraw_is_refused() { }), writer: Some(StreamWriter { producer: Some(client()), - withdraw_run: "r1".to_string(), - ..Default::default() + withdrawal: Some(RunWithdrawal { + withdraw_run: "r1".to_string(), + ..Default::default() + }), }), ..Default::default() }; @@ -696,8 +696,9 @@ async fn a_local_activity_writer_withdraws_the_records_of_a_failed_task() { for case in cases().await { let local_activity = |call: &str, started: i64| Producer { id: local_activity_producer_id("r1", call), - epoch: Some(ProducerEpoch { - kind: Some(Kind::LocalActivity(LocalActivityEpoch { + kind: Some(Kind::LocalActivity(LocalActivityProducer { + run_id: "r1".to_string(), + epoch: Some(LocalActivityEpoch { task: Some(WorkflowTaskEpoch { started_event_id: started, attempt: 1, @@ -707,18 +708,18 @@ async fn a_local_activity_writer_withdraws_the_records_of_a_failed_task() { }), }), attempt: 1, - })), - }), - run_id: "r1".to_string(), + }), + })), }; let facts = |completed: i64, failed: Vec| { move |request: &mut OpenWriterRequest| { let writer = request.writer.get_or_insert_with(Default::default); - writer.previous_completed_task = Some(WorkflowTaskEpoch { + let withdrawal = writer.withdrawal.get_or_insert_with(Default::default); + withdrawal.previous_completed_task = Some(WorkflowTaskEpoch { started_event_id: completed, ..Default::default() }); - writer.failed_tasks = failed.clone(); + withdrawal.failed_tasks = failed.clone(); } }; // Task [3] fails after its Local Activity writes, and task [4] completes with no write. diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index 307ad4fbc..88e31342f 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -8,10 +8,10 @@ use crate::{ Reason, StreamError, StreamResult, - epoch::ProducerKind, + epoch::{ProducerKind, writer_withdraw_run}, proto::{ AppendRecord, AppendRecordsRequest, OpenWriterRequest, Producer, StreamRef, StreamWriter, - producer_epoch::Kind, + producer::Kind, }, service::StreamService, store::{MAX_BATCH_BYTES, MAX_BATCH_RECORDS, MAX_RECORD_BYTES}, @@ -232,7 +232,7 @@ impl Writers { crate::ids::check_stream_id(&stream.id)?; let stream_writer = request.writer.unwrap_or_default(); // Only a retry run's first append withdraws a run, and a writer sends many batches. - if !stream_writer.withdraw_run.is_empty() { + if !writer_withdraw_run(&stream_writer).is_empty() { return Err(StreamError::invalid("A writer doesn't withdraw a run")); } let producer = stream_writer @@ -243,16 +243,24 @@ impl Writers { if ProducerKind::try_from(&producer)? == ProducerKind::Workflow { return Err(StreamError::invalid("Workflow code doesn't open a writer")); } - let epoch = producer.epoch.unwrap_or_default(); let key = ( request.namespace.clone(), stream.id.clone(), producer.id.clone(), - crate::epoch::epoch_key(&epoch), + crate::epoch::epoch_key(producer.kind.as_ref()), ); if let Some(id) = self.share(&key) { return Ok(id); } + // A Local Activity that runs again in a later task has attempt 1 again, so its writer + // always sends the empty batch. + let retry = match producer.kind.as_ref() { + Some(Kind::LocalActivity(_)) => true, + Some(Kind::Activity(activity) | Kind::StandaloneActivity(activity)) => { + activity.epoch.map_or(0, |e| e.attempt) > 1 + } + _ => false, + }; let writer = Arc::new(Writer { namespace: request.namespace, stream_ref: stream, @@ -268,15 +276,6 @@ impl Writers { state: Mutex::default(), writing: tokio::sync::Mutex::new(()), }); - // A Local Activity that runs again in a later task has attempt 1 again, so its writer - // always sends the empty batch. - let retry = match epoch.kind { - Some(Kind::LocalActivity(_)) => true, - Some(Kind::Activity(activity) | Kind::StandaloneActivity(activity)) => { - activity.attempt > 1 - } - _ => false, - }; if retry { match writer.send(vec![], next_sequence()).await { // A Stream with no key has nothing to withdraw yet. A closed or expired one accepts diff --git a/crates/streams/tests/capabilities.rs b/crates/streams/tests/capabilities.rs index 2b312f1da..34212ccc9 100644 --- a/crates/streams/tests/capabilities.rs +++ b/crates/streams/tests/capabilities.rs @@ -12,10 +12,10 @@ use temporalio_streams::{ ids::client_producer_id, memory::MemoryStore, proto::{ - AppendRecord, AppendRecordsRequest, ClientEpoch, CloseStreamRequest, CreateStreamRequest, - DescribeStreamRequest, DescribeStreamResponse, Producer, ProducerEpoch, StreamErrorDetails, - StreamErrorReason, StreamRef, StreamStatus, StreamWriter, WorkflowTaskEpoch, - producer_epoch::Kind, + AppendRecord, AppendRecordsRequest, ClientEpoch, ClientProducer, CloseStreamRequest, + CreateStreamRequest, DescribeStreamRequest, DescribeStreamResponse, Producer, + StreamErrorDetails, StreamErrorReason, StreamRef, StreamStatus, StreamWriter, + WorkflowProducer, WorkflowTaskEpoch, producer::Kind, }, store::{Appended, Batch, Capabilities, Closed, Create, Created, Entry, Store, StreamState}, }; @@ -120,10 +120,9 @@ fn case(without: Capabilities) -> Case { fn client() -> Producer { Producer { id: client_producer_id("me", "p"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt: 1 })), - }), - run_id: String::new(), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt: 1 }), + })), } } @@ -186,10 +185,10 @@ impl Case { withdraw_run: withdraw_run.to_string(), owner: Some(Producer { id: temporalio_streams::ids::workflow_producer_id(&self.run_id), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Workflow(WorkflowTaskEpoch::default())), - }), - run_id: self.run_id.clone(), + kind: Some(Kind::Workflow(WorkflowProducer { + run_id: self.run_id.clone(), + epoch: Some(WorkflowTaskEpoch::default()), + })), }), stream_ref: Some(StreamRef { id: self.stream.clone(), diff --git a/crates/streams/tests/scans.rs b/crates/streams/tests/scans.rs index bcc5ea044..e2e5f54c4 100644 --- a/crates/streams/tests/scans.rs +++ b/crates/streams/tests/scans.rs @@ -14,10 +14,10 @@ use temporalio_streams::{ ids::client_producer_id, memory::MemoryStore, proto::{ - AppendRecord, AppendRecordsRequest, AppendRecordsResponse, ClientEpoch, - CreateStreamRequest, CreateStreamResponse, Producer, ProducerEpoch, ReadRecordsRequest, + AppendRecord, AppendRecordsRequest, AppendRecordsResponse, ClientEpoch, ClientProducer, + CreateStreamRequest, CreateStreamResponse, Producer, ReadRecordsRequest, ReadRecordsResponse, StreamErrorDetails, StreamErrorReason, StreamRef, StreamWriter, - producer_epoch::Kind, + producer::Kind, }, store::{Appended, Batch, Capabilities, Create, Created, Entry, Store, StreamState}, }; @@ -129,10 +129,9 @@ async fn create(ext: &StreamRuntime) { fn client() -> Producer { Producer { id: client_producer_id("me", "p"), - epoch: Some(ProducerEpoch { - kind: Some(Kind::Client(ClientEpoch { attempt: 1 })), - }), - run_id: String::new(), + kind: Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt: 1 }), + })), } } diff --git a/crates/streams/tests/vectors.rs b/crates/streams/tests/vectors.rs index 853a4e96c..d824561a4 100644 --- a/crates/streams/tests/vectors.rs +++ b/crates/streams/tests/vectors.rs @@ -12,9 +12,10 @@ use temporalio_streams::{ ids::{self, StreamKeys}, offset::{entry_id_of_offset, offset_of_entry_id}, proto::{ - ActivityEpoch, AppendRecord, ClientEpoch, LocalActivityEpoch, Producer, ProducerEpoch, - StreamData, StreamRecord, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, - producer_epoch::Kind, stream_record::Item, + ActivityEpoch, ActivityProducer, AppendRecord, ClientEpoch, ClientProducer, + LocalActivityEpoch, LocalActivityProducer, Producer, RunWithdrawal, StreamData, + StreamRecord, StreamSuperseded, StreamWriter, WorkflowProducer, WorkflowTaskEpoch, + producer::Kind, stream_record::Item, }, }; @@ -39,31 +40,43 @@ fn number(value: &Value, field: &str) -> i64 { value[field].as_i64().unwrap_or(0) } -/// The epoch of a kind, from the flat fields of the vectors. The case takes the fields of its kind. -fn epoch(kind: &str, value: &Value) -> ProducerEpoch { +/// The Producer kind of a vector, from its flat fields and its run. The case takes the fields of +/// its kind, and a run-based kind carries the run. `empty` has no case. +fn kind_of(kind: &str, value: &Value, run_id: &str) -> Option { let task = WorkflowTaskEpoch { started_event_id: number(value, "task_started_event_id"), attempt: number(value, "task_attempt") as i32, started_time: time(&value["task_started_time"]), }; let attempt = number(value, "attempt") as i32; - let activity = ActivityEpoch { - attempt_scheduled_time: time(&value["attempt_scheduled_time"]), - attempt, - }; - let kind = match kind { - "workflow" => Some(Kind::Workflow(task)), - "local_activity" => Some(Kind::LocalActivity(LocalActivityEpoch { - task: Some(task), + let run_id = run_id.to_string(); + let activity = ActivityProducer { + run_id: run_id.clone(), + epoch: Some(ActivityEpoch { + attempt_scheduled_time: time(&value["attempt_scheduled_time"]), attempt, + }), + }; + match kind { + "workflow" => Some(Kind::Workflow(WorkflowProducer { + run_id, + epoch: Some(task), + })), + "local_activity" => Some(Kind::LocalActivity(LocalActivityProducer { + run_id, + epoch: Some(LocalActivityEpoch { + task: Some(task), + attempt, + }), })), "activity" => Some(Kind::Activity(activity)), "standalone_activity" => Some(Kind::StandaloneActivity(activity)), - "client" => Some(Kind::Client(ClientEpoch { attempt })), + "client" => Some(Kind::Client(ClientProducer { + epoch: Some(ClientEpoch { attempt }), + })), "empty" => None, other => panic!("unknown kind {other}"), - }; - ProducerEpoch { kind } + } } #[test] @@ -137,10 +150,10 @@ fn offsets_match() { #[test] fn epoch_keys_match() { for case in load("epochs.json").as_array().unwrap() { - let epoch = epoch(text(case, "kind"), &case["epoch"]); - assert_eq!(epoch_key(&epoch).to_vec(), hex(text(case, "key"))); + let kind = kind_of(text(case, "kind"), &case["epoch"], ""); + assert_eq!(epoch_key(kind.as_ref()).to_vec(), hex(text(case, "key"))); assert_eq!( - task_key(&task_of(&epoch)).to_vec(), + task_key(&task_of(kind.as_ref())).to_vec(), hex(text(case, "task_key")) ); } @@ -152,29 +165,34 @@ fn the_appendix_e_withdrawals_match() { assert!(cases.as_array().unwrap().len() >= 35); for case in cases.as_array().unwrap() { let s = &case["superseded"]; + // A create's item has no epoch, and a Reader must not read it as an empty one. let withdrawing = Producer { id: text(s, "producer_id").to_string(), - // A create's item has no epoch, and a Reader must not read it as an empty one. - epoch: s - .get("epoch") - .map(|value| epoch(text(s, "producer_kind"), value)), - run_id: s["run_id"].as_str().unwrap_or_default().to_string(), + kind: s.get("epoch").and_then(|value| { + kind_of( + text(s, "producer_kind"), + value, + s["run_id"].as_str().unwrap_or_default(), + ) + }), }; let superseded = StreamSuperseded { writer: Some(StreamWriter { producer: Some(withdrawing), - previous_completed_task: s.get("previous_completed_task").map(|c| { - WorkflowTaskEpoch { - started_event_id: number(c, "started_event_id"), - attempt: number(c, "attempt") as i32, - started_time: time(&c["started_time"]), - } + withdrawal: Some(RunWithdrawal { + previous_completed_task: s.get("previous_completed_task").map(|c| { + WorkflowTaskEpoch { + started_event_id: number(c, "started_event_id"), + attempt: number(c, "attempt") as i32, + started_time: time(&c["started_time"]), + } + }), + failed_tasks: s["failed_tasks"] + .as_array() + .map(|list| list.iter().map(|v| v.as_i64().unwrap()).collect()) + .unwrap_or_default(), + withdraw_run: s["withdraw_run"].as_str().unwrap_or_default().to_string(), }), - failed_tasks: s["failed_tasks"] - .as_array() - .map(|list| list.iter().map(|v| v.as_i64().unwrap()).collect()) - .unwrap_or_default(), - withdraw_run: s["withdraw_run"].as_str().unwrap_or_default().to_string(), }), }; let p = &case["record"]["producer"]; @@ -182,8 +200,13 @@ fn the_appendix_e_withdrawals_match() { item: Some(Item::Data(StreamData { producer: Some(Producer { id: text(p, "id").to_string(), - epoch: p.get("epoch").map(|value| epoch(text(p, "kind"), value)), - run_id: p["run_id"].as_str().unwrap_or_default().to_string(), + kind: p.get("epoch").and_then(|value| { + kind_of( + text(p, "kind"), + value, + p["run_id"].as_str().unwrap_or_default(), + ) + }), }), ..Default::default() })),