diff --git a/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto index e149af265..f7ece3ea0 100644 --- a/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto +++ b/crates/streams/protos/temporal/sdk/streams/v1/runtime.proto @@ -6,7 +6,9 @@ package temporal.sdk.streams.v1; import "google/protobuf/duration.proto"; import "google/protobuf/timestamp.proto"; +import "temporal/api/enums/v1/stream.proto"; import "temporal/api/stream/v1/message.proto"; +import "temporal/api/streamservice/v1/request_response.proto"; // The settings of the memory Stream runtime, for tests. Each field has the number of the same // field in the config of the other Stores, such as Redis, so a backend can send that config here @@ -160,3 +162,89 @@ message ScopedStreamIdRequest { message ScopedStreamIdResponse { string id = 1; } + +// Method `OpenActivityScope`, asynchronous. Opens the scope of one Activity attempt, which writes +// as the Producer of `DeriveProducer`, with one writer for each Stream. `CloseScope` ends it. +message OpenActivityScopeRequest { + string namespace = 1; + ActivityContext context = 2; + // The Streams in the attempt's arguments. From attempt 2, the open withdraws the earlier + // attempts' Records on each. A missing, closed or expired Stream doesn't fail it. + repeated temporal.api.stream.v1.StreamRef streams = 3; +} + +// Method `OpenClientScope`, synchronous. Opens the scope of the client Producer `client:` at +// `epoch`, or with no name `client::` at epoch 1, with a new key for each scope. +message OpenClientScopeRequest { + string namespace = 1; + string identity = 2; + string name = 3; + // 1 or more, with a name only. + int32 epoch = 4; +} + +message OpenScopeResponse { + uint64 scope_id = 1; + temporal.api.stream.v1.Producer producer = 2; +} + +// Method `Publish`, asynchronous. Appends Records as the scope's Producer and answers when the +// Store has them. The first publish to a Stream opens the scope's writer, as `OpenWriter` does. +message PublishRequest { + uint64 scope_id = 1; + temporal.api.stream.v1.StreamRef stream = 2; + // The bodies are in their stored form, after lang's converter and codec. + repeated temporal.api.stream.v1.AppendRecord records = 3; +} + +message PublishResponse { + // The offsets of the first and the last Record. Zero when the request has no Record. + int64 first_offset = 1; + int64 last_offset = 2; +} + +// Method `CloseStreamAs`, asynchronous, with `CloseStreamResponse`. Closes the scope's writer for +// the Stream, then the Stream, as the scope's Producer. An Activity scope fails with NOT_OWNER. +message CloseStreamAsRequest { + uint64 scope_id = 1; + // The runtime sets `namespace` and `producer` from the scope. Payloads are in their stored + // form. + temporal.api.streamservice.v1.CloseStreamRequest close = 2; +} + +// Method `CloseScope`, asynchronous. Closes every writer of the scope and frees it. A failed +// writer doesn't fail the call. The response counts it instead. +message CloseScopeRequest { + uint64 scope_id = 1; +} + +message CloseScopeResponse { + int32 failed_writers = 1; + // The first failure, if any. + StreamFailure failure = 2; +} + +// A failure that a response reports, as lang raises it from a failed call. +message StreamFailure { + temporal.api.enums.v1.StreamErrorReason reason = 1; + string message = 2; +} + +// Method `FreeScope`, synchronous. Frees the scope and releases its writers with no wait, such as +// from a finalizer. A write in flight still completes. +message FreeScopeRequest { + uint64 scope_id = 1; +} + +// The response of `FreeScope` and the other `Free` methods, which do nothing once a shutdown ended. +message FreeResponse { +} + +// Method `Shutdown`, asynchronous. Closes every scope and writer, for the backend's shutdown. Then +// only `Shutdown`, the static and the `Free` methods run. A dropped one runs again. +message ShutdownRequest { +} + +message ShutdownResponse { + int32 failed_writers = 1; +} diff --git a/crates/streams/src/error.rs b/crates/streams/src/error.rs index a149d9127..a778cd8c9 100644 --- a/crates/streams/src/error.rs +++ b/crates/streams/src/error.rs @@ -1,6 +1,6 @@ //! The error of every failed call, and its form for lang. -use crate::proto::{StreamErrorDetails, StreamErrorReason}; +use crate::proto::{StreamErrorDetails, StreamErrorReason, StreamFailure}; use prost::Message; use tonic::Code; @@ -38,9 +38,10 @@ pub enum Reason { Invalid, /// The Store failed or did not answer. Storage, - /// No writer has the id. + /// No writer has the id. A scope, reader or buffer id that the runtime doesn't have gives it + /// too. UnknownWriter, - /// The Store lacks a capability that the call needs. + /// The Store lacks a capability that the call needs, or the runtime shut down. Unsupported, } @@ -146,6 +147,14 @@ impl StreamError { } .encode_to_vec() } + + /// The error as a response reports it, when a failed write doesn't fail the call. + pub fn failure(&self) -> StreamFailure { + StreamFailure { + reason: self.reason.proto() as i32, + message: self.message.clone(), + } + } } #[cfg(test)] diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 3ecfa8e48..9a8e39faa 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -16,6 +16,7 @@ pub mod offset; pub mod protos; mod record; mod runtime; +mod scope; mod sdk; mod service; pub mod store; diff --git a/crates/streams/src/runtime.rs b/crates/streams/src/runtime.rs index 254f83714..33e299bcf 100644 --- a/crates/streams/src/runtime.rs +++ b/crates/streams/src/runtime.rs @@ -1,21 +1,29 @@ //! The Stream runtime, a plain object that lang's bridge builds per client and calls by name. use crate::{ - StreamError, StreamResult, + Reason, StreamError, StreamResult, memory::MemoryStore, metrics::Metrics, proto::{ - AppendRecordsRequest, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, - CloseWritersRequest, CloseWritersResponse, DescribeStreamRequest, FreeWriterRequest, - FreeWriterResponse, MemoryStreamConfig, OpenWriterRequest, OpenWriterResponse, - ReadRecordsRequest, WriteRecordsRequest, WriteRecordsResponse, + AppendRecordsRequest, CloseScopeRequest, CloseStreamRequest, CloseWriterRequest, + CloseWriterResponse, CloseWritersRequest, CloseWritersResponse, DescribeStreamRequest, + FreeResponse, FreeScopeRequest, FreeWriterRequest, FreeWriterResponse, MemoryStreamConfig, + OpenWriterRequest, OpenWriterResponse, ReadRecordsRequest, ShutdownRequest, + WriteRecordsRequest, WriteRecordsResponse, }, + scope::Scopes, service::StreamService, store::{DEFAULT_RETENTION, DEFAULT_SAFETY_TTL, Store}, - writer::Writers, + writer::{Holder, Writers}, }; use prost::Message; -use std::{sync::Arc, time::Duration}; +use std::{ + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::Duration, +}; use temporalio_common::telemetry::metrics::TemporalMeter; use tokio::sync::OnceCell; @@ -47,10 +55,30 @@ pub(crate) struct Shared { metrics: Metrics, factory: Arc, service: OnceCell>, - writers: Writers, + pub(crate) writers: Writers, + pub(crate) scopes: Scopes, + shut_down: AtomicBool, + /// Set when a `Shutdown` ends every scope and writer. Then the `Free` methods do nothing. + finished: AtomicBool, } impl Shared { + pub(crate) fn is_shut_down(&self) -> bool { + self.shut_down.load(Ordering::SeqCst) + } + + fn is_finished(&self) -> bool { + self.finished.load(Ordering::SeqCst) + } + + /// A free of something that a running shutdown already took has nothing left to do. + fn taken_by_shutdown(&self, freed: StreamResult<()>) -> StreamResult<()> { + match freed { + Err(error) if error.reason == Reason::UnknownWriter && self.is_shut_down() => Ok(()), + freed => freed, + } + } + /// The service, which connects its Store on its first use. pub(crate) async fn service(&self) -> StreamResult<&Arc> { self.service @@ -102,6 +130,9 @@ impl StreamRuntime { factory, service: OnceCell::new(), writers: Writers::default(), + scopes: Scopes::default(), + shut_down: AtomicBool::new(false), + finished: AtomicBool::new(false), }); Arc::new(Self { shared }) } @@ -129,14 +160,25 @@ impl StreamRuntime { /// 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) + if self.shared.is_finished() { + return Ok(()); + } + let freed = self.shared.writers.free(writer_id, Holder::Lang); + self.shared.taken_by_shutdown(freed) + } + + #[cfg(test)] + pub(crate) fn shared(&self) -> &Arc { + &self.shared } /// Runs any 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> { + self.check_open(method)?; // The methods of `call_sync` don't connect, so they work when the Store is down too. let service = || self.shared.service(); + let shared = &self.shared; let output = match method { "CreateStream" => service() .await? @@ -165,7 +207,15 @@ impl StreamRuntime { .encode_to_vec(), "OpenWriter" => { let request = decode::(input)?; - let writer_id = self.shared.writers.open(service().await?, request).await?; + let writers = &self.shared.writers; + let writer_id = writers + .open(service().await?, request, Holder::Lang) + .await?; + // A shutdown during the open has already closed every writer. + if shared.is_shut_down() { + let _ = shared.writers.free(writer_id, Holder::Lang); + return Err(shut_down()); + } OpenWriterResponse { writer_id }.encode_to_vec() } "WriteRecords" => { @@ -183,6 +233,7 @@ impl StreamRuntime { } "CloseWriters" => { decode::(input)?; + // The writers of scopes stay, since another Worker of the client may use them. let failed = self.shared.writers.close_all().await; CloseWritersResponse { failed: failed as i32, @@ -191,9 +242,31 @@ impl StreamRuntime { } "CloseWriter" => { let request = decode::(input)?; - let last_offset = self.shared.writers.close(request.writer_id).await?; + let last_offset = (self.shared.writers) + .close(request.writer_id, Holder::Lang) + .await?; CloseWriterResponse { last_offset }.encode_to_vec() } + "OpenActivityScope" => shared + .open_activity_scope(decode(input)?) + .await? + .encode_to_vec(), + "Publish" => shared.publish(decode(input)?).await?.encode_to_vec(), + "CloseStreamAs" => shared + .close_stream_as(decode(input)?) + .await? + .encode_to_vec(), + "CloseScope" => { + let request = decode::(input)?; + shared.close_scope(request.scope_id).await?.encode_to_vec() + } + "Shutdown" => { + decode::(input)?; + shared.shut_down.store(true, Ordering::SeqCst); + let closed = shared.close_everything().await; + shared.finished.store(true, Ordering::SeqCst); + closed.encode_to_vec() + } _ => return self.call_sync(method, input), }; Ok(output) @@ -202,15 +275,52 @@ impl StreamRuntime { /// Runs a method that never waits, with no async runtime entered, so a finalizer can call it. /// `runtime.proto` marks these methods as synchronous or static. pub fn call_sync(&self, method: &str, input: &[u8]) -> StreamResult> { + // A finalizer can run after the shutdown, which already freed everything. Before that, a + // free still works, so that lang can end a buffer whose pump is gone. + if method.starts_with("Free") && self.shared.is_finished() { + return Ok(vec![]); + } + self.check_open(method)?; + let output = self.run_sync(method, input); + match output { + Err(error) if method.starts_with("Free") => { + self.shared.taken_by_shutdown(Err(error)).map(|()| vec![]) + } + output => output, + } + } + + fn run_sync(&self, method: &str, input: &[u8]) -> StreamResult> { + let shared = &self.shared; match method { "FreeWriter" => { let request = decode::(input)?; self.free_writer(request.writer_id)?; Ok(FreeWriterResponse {}.encode_to_vec()) } + "OpenClientScope" => Ok(shared.open_client_scope(decode(input)?)?.encode_to_vec()), + "FreeScope" => { + let request = decode::(input)?; + shared.free_scope(request.scope_id)?; + Ok(FreeResponse {}.encode_to_vec()) + } _ => crate::call_static(method, input), } } + + /// Refuses every method after `Shutdown`, except `Shutdown`, the `Free` methods and the static + /// ones. + fn check_open(&self, method: &str) -> StreamResult<()> { + // A dropped `Shutdown` leaves work that the next one does. + if !self.shared.is_shut_down() + || method.starts_with("Free") + || crate::sdk::is_static(method) + || method == "Shutdown" + { + return Ok(()); + } + Err(shut_down()) + } } /// The name of the memory runtime, for a bridge that selects a runtime by name. @@ -265,6 +375,10 @@ fn duration(value: Option) -> Option { (!duration.is_zero()).then_some(duration) } +pub(crate) fn shut_down() -> StreamError { + StreamError::new(Reason::Unsupported, "The Stream runtime shut down") +} + pub(crate) fn decode(input: &[u8]) -> StreamResult { M::decode(input).map_err(|error| StreamError::invalid(format!("The request is bad: {error}"))) } diff --git a/crates/streams/src/scope.rs b/crates/streams/src/scope.rs new file mode 100644 index 000000000..6f188658a --- /dev/null +++ b/crates/streams/src/scope.rs @@ -0,0 +1,696 @@ +//! Producer scopes. A scope is the Producer of one Activity attempt or one client, and the writers +//! that write as it, one for each Stream. Lang keeps only the scope id. + +use crate::{ + Reason, StreamError, StreamResult, derive_producer, ids, + proto::{ + CloseScopeResponse, CloseStreamAsRequest, CloseStreamResponse, DeriveProducerRequest, + OpenActivityScopeRequest, OpenClientScopeRequest, OpenScopeResponse, OpenWriterRequest, + Producer, ProducerEpoch, ProducerPurpose, PublishRequest, PublishResponse, + ShutdownResponse, StreamProducerKind, StreamRef, WorkflowTaskEpoch, + }, + runtime::Shared, + writer::Holder, +}; +use parking_lot::Mutex; +use std::{ + collections::{HashMap, HashSet}, + hash::{BuildHasher, Hasher}, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, +}; + +pub(crate) struct Scope { + namespace: String, + producer: Producer, + previous_completed_task: Option, + failed_tasks: Vec, + /// An Activity attempt's scope, which doesn't close a Stream. + activity: bool, + writers: Mutex, + /// Allows one open at a time for each Stream, so that two first publishes share one writer. + opening: Mutex>>>, +} + +#[derive(Default)] +struct ScopeWriters { + by_stream: HashMap, + /// Set by the close or free of the scope. A writer that opens after it is released at once. + ended: bool, +} + +impl Scope { + fn new( + namespace: String, + producer: Producer, + previous_completed_task: Option, + failed_tasks: Vec, + activity: bool, + ) -> Self { + Self { + namespace, + producer, + previous_completed_task, + failed_tasks, + activity, + writers: Mutex::default(), + opening: Mutex::default(), + } + } + + fn cached(&self, stream_id: &str) -> StreamResult> { + let writers = self.writers.lock(); + if writers.ended { + return Err(unknown_scope()); + } + Ok(writers.by_stream.get(stream_id).copied()) + } + + /// The scope's writer for `stream`, which the first call opens. + pub(crate) async fn writer(&self, shared: &Shared, stream: &StreamRef) -> StreamResult { + if let Some(id) = self.cached(&stream.id)? { + return Ok(id); + } + let opening = self.opening_lock(&stream.id); + let _opening = opening.lock().await; + if let Some(id) = self.cached(&stream.id)? { + return Ok(id); + } + let request = OpenWriterRequest { + namespace: self.namespace.clone(), + stream: Some(stream.clone()), + producer: Some(self.producer.clone()), + previous_completed_task: self.previous_completed_task, + failed_tasks: self.failed_tasks.clone(), + ..Default::default() + }; + let service = shared.service().await?; + let id = shared.writers.open(service, request, Holder::Scope).await?; + let mut writers = self.writers.lock(); + if writers.ended { + drop(writers); + shared.writers.free(id, Holder::Scope)?; + return Err(unknown_scope()); + } + if let Some(existing) = writers.by_stream.get(&stream.id).copied() { + drop(writers); + // An open that waited on an older lock got there first, so this handle goes. + shared.writers.free(id, Holder::Scope)?; + return Ok(existing); + } + writers.by_stream.insert(stream.id.clone(), id); + Ok(id) + } + + /// Ends the scope and returns its writers, which the caller frees. + fn end(&self) -> Vec { + let mut writers = self.writers.lock(); + writers.ended = true; + self.opening.lock().clear(); + writers.by_stream.drain().map(|(_, id)| id).collect() + } + + fn opening_lock(&self, stream_id: &str) -> Arc> { + let mut opening = self.opening.lock(); + opening.entry(stream_id.to_string()).or_default().clone() + } + + /// Ends the scope and takes one of its writers, which the caller closes. + fn take_writer(&self) -> Option { + let mut writers = self.writers.lock(); + writers.ended = true; + let stream = writers.by_stream.keys().next()?.clone(); + writers.by_stream.remove(&stream) + } +} + +/// Frees what a scope still holds when the call that opens or ends it fails or is dropped. A +/// shutdown puts its scope back in the table instead, so that the next shutdown ends it. +struct Leftovers<'a> { + shared: &'a Shared, + scope: &'a Scope, + requeue: Option<(u64, Arc)>, + armed: bool, +} + +impl Drop for Leftovers<'_> { + fn drop(&mut self) { + if !self.armed { + return; + } + match self.requeue.take() { + Some((id, scope)) => { + self.shared.scopes.table.lock().insert(id, scope); + } + None => self.shared.release(self.scope), + } + } +} + +fn unknown_scope() -> StreamError { + StreamError::new(Reason::UnknownWriter, "No scope has the id") +} + +pub(crate) struct Scopes { + next: AtomicU64, + table: Mutex>>, + /// The runtime's random key. A client scope with no name takes `-`. + client_key: String, + clients: AtomicU64, +} + +impl Default for Scopes { + fn default() -> Self { + Self { + next: AtomicU64::new(0), + table: Mutex::default(), + client_key: random_key(), + clients: AtomicU64::new(0), + } + } +} + +/// Random, so that two processes, or a forked child and its parent, never share a client Producer. +fn random_key() -> String { + let half = |salt: u64| { + let mut hasher = std::collections::hash_map::RandomState::new().build_hasher(); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default(); + hasher.write_u128(now.as_nanos()); + hasher.write_u32(std::process::id()); + hasher.write_u64(salt); + hasher.finish() + }; + format!("{:016x}{:016x}", half(1), half(2)) +} + +impl Scopes { + fn insert(&self, scope: Scope) -> (u64, Arc) { + let id = self.next.fetch_add(1, Ordering::Relaxed) + 1; + let scope = Arc::new(scope); + self.table.lock().insert(id, scope.clone()); + (id, scope) + } + + pub(crate) fn get(&self, id: u64) -> StreamResult> { + self.table + .lock() + .get(&id) + .cloned() + .ok_or_else(unknown_scope) + } + + fn remove(&self, id: u64) -> StreamResult> { + self.table.lock().remove(&id).ok_or_else(unknown_scope) + } + + fn take_any(&self) -> Option<(u64, Arc)> { + let mut table = self.table.lock(); + let id = *table.keys().next()?; + table.remove(&id).map(|scope| (id, scope)) + } +} + +impl Shared { + pub(crate) async fn open_activity_scope( + &self, + request: OpenActivityScopeRequest, + ) -> StreamResult { + let context = request + .context + .ok_or_else(|| StreamError::invalid("The request has no Activity context"))?; + let attempt = context.attempt; + let derived = derive_producer(DeriveProducerRequest { + context: Some(context), + purpose: ProducerPurpose::ActivityWriter as i32, + })?; + let producer = derived.producer.unwrap_or_default(); + let scope = Scope::new( + request.namespace, + producer.clone(), + derived.previous_completed_task, + derived.failed_tasks, + true, + ); + if attempt > 1 { + let mut leftovers = Leftovers { + shared: self, + scope: &scope, + requeue: None, + armed: true, + }; + let mut seen = HashSet::new(); + for stream in request.streams.iter().filter(|s| seen.insert(s.id.clone())) { + // The writer's open ignores a missing, closed or expired Stream. + scope.writer(self, stream).await?; + } + leftovers.armed = false; + } + // Lang gets no id from a dropped call, so the scope joins the table only at the end. + let scope_id = self.admit(scope)?; + Ok(OpenScopeResponse { + scope_id, + producer: Some(producer), + }) + } + + pub(crate) fn open_client_scope( + &self, + request: OpenClientScopeRequest, + ) -> StreamResult { + let (id, attempt) = if request.name.is_empty() { + let n = self.scopes.clients.fetch_add(1, Ordering::Relaxed) + 1; + let key = format!("{}-{n}", self.scopes.client_key); + (ids::client_producer_id(&request.identity, &key), 1) + } else if request.epoch < 1 { + return Err(StreamError::invalid( + "A named Producer's epoch is 1 or more", + )); + } else { + (ids::named_client_producer_id(&request.name), request.epoch) + }; + let producer = Producer { + kind: StreamProducerKind::Client as i32, + id, + epoch: Some(ProducerEpoch { + attempt, + ..Default::default() + }), + run_id: String::new(), + }; + ids::check_producer(&producer)?; + let scope_id = self.admit(Scope::new( + request.namespace, + producer.clone(), + None, + vec![], + false, + ))?; + Ok(OpenScopeResponse { + scope_id, + producer: Some(producer), + }) + } + + /// Adds a scope to the table. A shutdown that already drained the table frees it instead. + fn admit(&self, scope: Scope) -> StreamResult { + let (scope_id, scope) = self.scopes.insert(scope); + if self.is_shut_down() { + if self.scopes.remove(scope_id).is_ok() { + self.release(&scope); + } + return Err(crate::runtime::shut_down()); + } + Ok(scope_id) + } + + pub(crate) async fn publish(&self, request: PublishRequest) -> StreamResult { + let scope = self.scopes.get(request.scope_id)?; + let stream = request + .stream + .ok_or_else(|| StreamError::invalid("The request has no Stream"))?; + let writer = scope.writer(self, &stream).await?; + let (first_offset, last_offset) = self.writers.write(writer, request.records).await?; + Ok(PublishResponse { + first_offset, + last_offset, + }) + } + + pub(crate) async fn close_stream_as( + &self, + request: CloseStreamAsRequest, + ) -> StreamResult { + let scope = self.scopes.get(request.scope_id)?; + if scope.activity { + return Err(StreamError::new( + Reason::NotOwner, + "An Activity doesn't close a Stream. Its Workflow or a client closes it", + )); + } + let mut close = request + .close + .ok_or_else(|| StreamError::invalid("The request has no close"))?; + let stream_id = close + .stream + .as_ref() + .map(|s| s.id.clone()) + .unwrap_or_default(); + // An open of the Stream waits, so that it can't share the writer while the close ends it. + let opening = scope.opening_lock(&stream_id); + let _opening = opening.lock().await; + // Closing the writer first puts this Producer's Records before the close Record. + let writer = scope.writers.lock().by_stream.remove(&stream_id); + if let Some(writer) = writer { + self.writers.close(writer, Holder::Scope).await?; + } + close.namespace = scope.namespace.clone(); + close.producer = Some(scope.producer.clone()); + let closed = self.service().await?.close_stream(close).await; + scope.opening.lock().remove(&stream_id); + closed + } + + pub(crate) async fn close_scope(&self, scope_id: u64) -> StreamResult { + let scope = self.scopes.remove(scope_id)?; + Ok(self.end_scope(&scope, None).await) + } + + async fn end_scope( + &self, + scope: &Scope, + requeue: Option<(u64, Arc)>, + ) -> CloseScopeResponse { + let mut leftovers = Leftovers { + shared: self, + scope, + requeue, + armed: true, + }; + let mut response = CloseScopeResponse::default(); + while let Some(writer) = scope.take_writer() { + if let Err(error) = self.writers.close(writer, Holder::Scope).await { + response.failed_writers += 1; + response.failure.get_or_insert_with(|| error.failure()); + } + } + leftovers.armed = false; + response + } + + pub(crate) fn free_scope(&self, scope_id: u64) -> StreamResult<()> { + let scope = self.scopes.remove(scope_id)?; + self.release(&scope); + Ok(()) + } + + /// Releases what the scope holds, with no wait. + fn release(&self, scope: &Scope) { + for writer in scope.end() { + let _ = self.writers.free(writer, Holder::Scope); + } + } + + /// Closes every scope, then every writer that lang opened with `OpenWriter`. + pub(crate) async fn close_everything(&self) -> ShutdownResponse { + let mut response = ShutdownResponse::default(); + // One scope at a time, so that a dropped shutdown leaves the rest for the next one. + while let Some((id, scope)) = self.scopes.take_any() { + let requeue = Some((id, scope.clone())); + response.failed_writers += self.end_scope(&scope, requeue).await.failed_writers; + } + response.failed_writers += self.writers.close_all().await as i32; + response + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + MemoryStreamRuntime, StreamRuntime, + memory::MemoryStore, + proto::{ + ActivityContext, AppendRecord, CloseScopeRequest, CloseScopeResponse, + CreateStreamRequest, FreeScopeRequest, ShutdownRequest, + }, + testing::support::Flaky, + }; + use prost::Message; + use std::time::Duration; + + #[test] + fn each_runtime_has_its_own_client_key() { + let (a, b) = (Scopes::default(), Scopes::default()); + assert_eq!(a.client_key.len(), 32); + assert_ne!(a.client_key, b.client_key); + } + + /// A runtime on a Store whose appends fail while it is down, with the Streams `a` and `b`. + async fn flaky() -> (Arc, Arc) { + let memory = MemoryStore::new(Duration::from_secs(3600), Duration::from_secs(7200)); + let store = Arc::new(Flaky::new(Arc::new(memory))); + let ext = MemoryStreamRuntime::with_store(store.clone(), None); + for stream in ["a", "b"] { + let create = CreateStreamRequest { + namespace: "ns".into(), + stream_id: stream.into(), + ..Default::default() + }; + ext.call("CreateStream", &create.encode_to_vec()) + .await + .unwrap(); + } + (store, ext) + } + + fn stream(id: &str) -> Option { + Some(StreamRef { + id: id.into(), + ..Default::default() + }) + } + + fn publish(scope_id: u64, id: &str) -> Vec { + PublishRequest { + scope_id, + stream: stream(id), + records: vec![AppendRecord::default()], + } + .encode_to_vec() + } + + fn retry_scope() -> Vec { + retry_scope_on(&["a"]) + } + + /// The open of attempt 2 of an Activity, which opens a writer on each of `streams`. + fn retry_scope_on(streams: &[&str]) -> Vec { + OpenActivityScopeRequest { + namespace: "ns".into(), + context: Some(ActivityContext { + in_workflow: true, + workflow_run_id: "r1".into(), + activity_id: "a".into(), + attempt: 2, + ..Default::default() + }), + streams: streams.iter().map(|id| stream(id).unwrap()).collect(), + } + .encode_to_vec() + } + + #[tokio::test] + async fn a_dropped_scope_close_frees_its_writers() { + let (store, ext) = flaky().await; + let open = OpenClientScopeRequest { + namespace: "ns".into(), + ..Default::default() + }; + let opened = OpenScopeResponse::decode( + ext.call_sync("OpenClientScope", &open.encode_to_vec()) + .unwrap() + .as_slice(), + ) + .unwrap(); + for id in ["a", "b"] { + ext.call("Publish", &publish(opened.scope_id, id)) + .await + .unwrap(); + } + // A write in flight on each writer makes the close wait on the first one it closes. + store.down.store(true, Ordering::SeqCst); + let writes: Vec<_> = ["a", "b"] + .into_iter() + .map(|id| { + let (ext, input) = (ext.clone(), publish(opened.scope_id, id)); + tokio::spawn(async move { ext.call("Publish", &input).await }) + }) + .collect(); + tokio::time::sleep(Duration::from_millis(50)).await; + let close = CloseScopeRequest { + scope_id: opened.scope_id, + }; + let input = close.encode_to_vec(); + let dropped = + tokio::time::timeout(Duration::from_millis(200), ext.call("CloseScope", &input)); + assert!(dropped.await.is_err()); + assert_eq!(ext.shared().writers.open_count(), 0); + store.down.store(false, Ordering::SeqCst); + for write in writes { + let _ = write.await.unwrap(); + } + } + + /// A client scope with a writer on `a` and on `b`, each with a write that waits while the + /// Store is down, so that a close of the scope waits on the first writer that it closes. + async fn scope_with_blocked_writes( + store: &Flaky, + ext: &Arc, + ) -> (u64, Vec>>>) { + let open = OpenClientScopeRequest { + namespace: "ns".into(), + ..Default::default() + }; + let opened = OpenScopeResponse::decode( + ext.call_sync("OpenClientScope", &open.encode_to_vec()) + .unwrap() + .as_slice(), + ) + .unwrap(); + for id in ["a", "b"] { + ext.call("Publish", &publish(opened.scope_id, id)) + .await + .unwrap(); + } + store.down.store(true, Ordering::SeqCst); + let writes = ["a", "b"] + .into_iter() + .map(|id| { + let (ext, input) = (ext.clone(), publish(opened.scope_id, id)); + tokio::spawn(async move { ext.call("Publish", &input).await }) + }) + .collect(); + tokio::time::sleep(Duration::from_millis(50)).await; + (opened.scope_id, writes) + } + + #[tokio::test] + async fn close_writers_leaves_the_writers_of_a_scope_that_is_closing() { + let (store, ext) = flaky().await; + let (scope_id, writes) = scope_with_blocked_writes(&store, &ext).await; + let close = tokio::spawn({ + let ext = ext.clone(); + let input = CloseScopeRequest { scope_id }.encode_to_vec(); + async move { ext.call("CloseScope", &input).await } + }); + tokio::time::sleep(Duration::from_millis(50)).await; + let close_writers = tokio::spawn({ + let ext = ext.clone(); + async move { ext.call("CloseWriters", &[]).await } + }); + tokio::time::sleep(Duration::from_millis(50)).await; + store.down.store(false, Ordering::SeqCst); + let closed = CloseScopeResponse::decode(close.await.unwrap().unwrap().as_slice()).unwrap(); + assert_eq!(closed, CloseScopeResponse::default()); + close_writers.await.unwrap().unwrap(); + for write in writes { + write.await.unwrap().unwrap(); + } + } + + #[tokio::test] + async fn a_shutdown_leaves_the_writers_of_a_scope_that_is_closing() { + let (store, ext) = flaky().await; + let (scope_id, writes) = scope_with_blocked_writes(&store, &ext).await; + let close = tokio::spawn({ + let ext = ext.clone(); + let input = CloseScopeRequest { scope_id }.encode_to_vec(); + async move { ext.call("CloseScope", &input).await } + }); + tokio::time::sleep(Duration::from_millis(50)).await; + let shutdown = ShutdownRequest {}.encode_to_vec(); + ext.call("Shutdown", &shutdown).await.unwrap(); + store.down.store(false, Ordering::SeqCst); + let closed = CloseScopeResponse::decode(close.await.unwrap().unwrap().as_slice()).unwrap(); + assert_eq!(closed, CloseScopeResponse::default()); + for write in writes { + write.await.unwrap().unwrap(); + } + } + + #[tokio::test] + async fn a_free_of_what_a_running_shutdown_took_answers_ok() { + let (store, ext) = flaky().await; + let (scope_id, writes) = scope_with_blocked_writes(&store, &ext).await; + let shutdown = tokio::spawn({ + let ext = ext.clone(); + async move { + ext.call("Shutdown", &ShutdownRequest {}.encode_to_vec()) + .await + } + }); + tokio::time::sleep(Duration::from_millis(100)).await; + let free = FreeScopeRequest { scope_id }.encode_to_vec(); + ext.call_sync("FreeScope", &free).unwrap(); + ext.free_writer(u64::MAX).unwrap(); + store.down.store(false, Ordering::SeqCst); + shutdown.await.unwrap().unwrap(); + for write in writes { + let _ = write.await.unwrap(); + } + } + + #[tokio::test] + async fn a_dropped_shutdown_runs_again() { + let (store, ext) = flaky().await; + let (_, writes) = scope_with_blocked_writes(&store, &ext).await; + let shutdown = ShutdownRequest {}.encode_to_vec(); + let dropped = + tokio::time::timeout(Duration::from_millis(200), ext.call("Shutdown", &shutdown)); + assert!(dropped.await.is_err()); + store.down.store(false, Ordering::SeqCst); + ext.call("Shutdown", &shutdown).await.unwrap(); + assert!(ext.shared().scopes.table.lock().is_empty()); + assert_eq!(ext.shared().writers.open_count(), 0); + for write in writes { + let _ = write.await.unwrap(); + } + } + + #[tokio::test] + async fn a_dropped_activity_scope_open_leaves_no_scope() { + let (store, ext) = flaky().await; + // The writer on `a` opens, and the one on `b` waits while the Store is down. + store.down_after.store(1, Ordering::SeqCst); + let input = retry_scope_on(&["a", "b"]); + let dropped = tokio::time::timeout( + Duration::from_millis(200), + ext.call("OpenActivityScope", &input), + ); + assert!(dropped.await.is_err()); + assert!(ext.shared().scopes.table.lock().is_empty()); + assert_eq!(ext.shared().writers.open_count(), 0); + } + + #[tokio::test] + async fn close_writers_leaves_the_writers_of_a_scope_that_is_opening() { + let (store, ext) = flaky().await; + store.down_after.store(1, Ordering::SeqCst); + let open = tokio::spawn({ + let ext = ext.clone(); + async move { + ext.call("OpenActivityScope", &retry_scope_on(&["a", "b"])) + .await + } + }); + tokio::time::sleep(Duration::from_millis(100)).await; + ext.call("CloseWriters", &[]).await.unwrap(); + store.down.store(false, Ordering::SeqCst); + let opened = OpenScopeResponse::decode(open.await.unwrap().unwrap().as_slice()).unwrap(); + ext.call("Publish", &publish(opened.scope_id, "a")) + .await + .unwrap(); + } + + #[tokio::test] + async fn an_open_that_ends_after_the_shutdown_is_refused() { + let (store, ext) = flaky().await; + store.down.store(true, Ordering::SeqCst); + let open = tokio::spawn({ + let ext = ext.clone(); + async move { ext.call("OpenActivityScope", &retry_scope()).await } + }); + tokio::time::sleep(Duration::from_millis(50)).await; + let shutdown = ShutdownRequest {}.encode_to_vec(); + ext.call("Shutdown", &shutdown).await.unwrap(); + store.down.store(false, Ordering::SeqCst); + let error = open.await.unwrap().unwrap_err(); + assert_eq!(error.reason, Reason::Unsupported); + assert!(ext.shared().scopes.table.lock().is_empty()); + assert_eq!(ext.shared().writers.open_count(), 0); + } +} diff --git a/crates/streams/src/sdk.rs b/crates/streams/src/sdk.rs index a383ea6ec..23d28b7fd 100644 --- a/crates/streams/src/sdk.rs +++ b/crates/streams/src/sdk.rs @@ -35,6 +35,11 @@ pub fn call_static(method: &str, input: &[u8]) -> StreamResult> { } } +/// Whether [call_static] runs the method. +pub(crate) fn is_static(method: &str) -> bool { + matches!(method, "DeriveProducer" | "ScopedStreamId") +} + /// The Producer of an Activity attempt, from the facts that lang copies from its Activity info. A /// Workflow write keeps attempt 1, so the attempts in one Workflow Task resend one batch. pub fn derive_producer(request: DeriveProducerRequest) -> StreamResult { diff --git a/crates/streams/src/testing/mod.rs b/crates/streams/src/testing/mod.rs index 3b4fdef8f..1f7c44b23 100644 --- a/crates/streams/src/testing/mod.rs +++ b/crates/streams/src/testing/mod.rs @@ -1,8 +1,9 @@ //! The shared suites of the Stream rules. Each Store crate runs them on its own backends, so that //! every Store keeps the same rules. This crate runs them on the memory Store. -mod support; +pub(crate) mod support; +pub mod scope; pub mod service; pub mod store; pub mod writer; @@ -158,4 +159,9 @@ mod tests { async fn the_writer_suite_passes_on_the_memory_store() { run(vec![Arc::new(MemoryBackend)], writer::CASES).await; } + + #[tokio::test] + async fn the_scope_suite_passes_on_the_memory_store() { + run(vec![Arc::new(MemoryBackend)], scope::CASES).await; + } } diff --git a/crates/streams/src/testing/scope.rs b/crates/streams/src/testing/scope.rs new file mode 100644 index 000000000..824287ab4 --- /dev/null +++ b/crates/streams/src/testing/scope.rs @@ -0,0 +1,550 @@ +//! Producer scopes of Activity attempts and clients, through the runtime's byte call, as lang sends +//! them. +//! +//! Every Store runs the suite through its own runtime. + +use super::case; +use crate::{ + StreamError, StreamRuntime, + epoch::withdraws, + proto::{ + ActivityContext, AppendRecord, CloseScopeRequest, CloseScopeResponse, CloseStreamAsRequest, + CloseStreamRequest, CloseStreamResponse, CloseWritersRequest, CloseWritersResponse, + CreateStreamRequest, CreateStreamResponse, DeriveProducerRequest, DeriveProducerResponse, + DescribeStreamRequest, DescribeStreamResponse, FreeResponse, FreeScopeRequest, + FreeWriterRequest, FreeWriterResponse, OpenActivityScopeRequest, OpenClientScopeRequest, + OpenScopeResponse, OpenWriterRequest, OpenWriterResponse, Producer, ProducerPurpose, + PublishRequest, PublishResponse, ReadRecordsRequest, ReadRecordsResponse, ShutdownRequest, + ShutdownResponse, StreamErrorReason, StreamRecord, StreamRef, StreamStatus, + WorkflowTaskAttempt, stream_record::Item, + }, +}; +use prost::Message; +use prost_types::Timestamp; +use std::{ + sync::Arc, + time::{SystemTime, UNIX_EPOCH}, +}; +use temporalio_common::protos::temporal::api::common::v1::Payload; + +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}"), + }; + case.create(&case.stream).await; + cases.push(case); + } + cases +} + +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) -> &'static str { + error.reason.token() +} + +/// The context of attempt `attempt` of Activity `a` in run `r1`. +fn activity(attempt: i32) -> ActivityContext { + ActivityContext { + in_workflow: true, + workflow_id: "wf".into(), + workflow_run_id: "r1".into(), + activity_id: "a".into(), + attempt, + current_attempt_scheduled_time: Some(Timestamp { + seconds: 10 * attempt as i64, + nanos: 0, + }), + call_key: "1".into(), + ..Default::default() + } +} + +/// The context of a Local Activity of run `r1` with call key `call`, dispatched by the task that +/// started at `started`. +fn local_activity(call: &str, started: i64, completed: i64, failed: Vec) -> ActivityContext { + ActivityContext { + is_local: true, + in_workflow: true, + workflow_id: "wf".into(), + workflow_run_id: "r1".into(), + activity_id: call.into(), + attempt: 1, + call_key: call.into(), + workflow_task: Some(WorkflowTaskAttempt { + started_event_id: started, + attempt: 1, + started_time: Some(Timestamp { + seconds: started, + nanos: 0, + }), + previous_started_event_id: completed, + failed_tasks: failed, + previous_attempt: 1, + previous_started_time: Some(Timestamp { + seconds: completed, + nanos: 0, + }), + }), + ..Default::default() + } +} + +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()) + } + + fn call_sync( + &self, + method: &str, + request: Req, + ) -> Result { + let bytes = self.ext.call_sync(method, &request.encode_to_vec())?; + Ok(Resp::decode(bytes.as_slice()).unwrap()) + } + + fn stream_ref(&self) -> StreamRef { + StreamRef { + id: self.stream.clone(), + ..Default::default() + } + } + + async fn create(&self, stream: &str) { + let created: CreateStreamResponse = self + .call( + "CreateStream", + CreateStreamRequest { + namespace: self.namespace.clone(), + stream_id: stream.to_string(), + ..Default::default() + }, + ) + .await + .unwrap(); + assert!(created.created); + } + + fn client_scope(&self, name: &str, epoch: i32) -> Result { + self.call_sync( + "OpenClientScope", + OpenClientScopeRequest { + namespace: self.namespace.clone(), + identity: "me@host".into(), + name: name.into(), + epoch, + }, + ) + } + + async fn activity_scope(&self, context: ActivityContext, streams: &[&str]) -> u64 { + let opened: OpenScopeResponse = self + .call( + "OpenActivityScope", + OpenActivityScopeRequest { + namespace: self.namespace.clone(), + context: Some(context), + streams: streams + .iter() + .map(|id| StreamRef { + id: id.to_string(), + ..Default::default() + }) + .collect(), + }, + ) + .await + .unwrap(); + opened.scope_id + } + + async fn publish(&self, scope_id: u64, texts: &[&str]) -> Result<(i64, i64), StreamError> { + let published: PublishResponse = self + .call( + "Publish", + PublishRequest { + scope_id, + stream: Some(self.stream_ref()), + records: records(texts), + }, + ) + .await?; + Ok((published.first_offset, published.last_offset)) + } + + async fn close_as(&self, scope_id: u64) -> Result { + self.call( + "CloseStreamAs", + CloseStreamAsRequest { + scope_id, + close: Some(CloseStreamRequest { + stream: Some(self.stream_ref()), + ..Default::default() + }), + }, + ) + .await + } + + async fn close_scope(&self, scope_id: u64) -> Result { + self.call("CloseScope", CloseScopeRequest { scope_id }) + .await + } + + async fn items(&self) -> Vec { + let read: ReadRecordsResponse = self + .call( + "ReadRecords", + ReadRecordsRequest { + namespace: self.namespace.clone(), + stream: Some(self.stream_ref()), + ..Default::default() + }, + ) + .await + .unwrap(); + read.records + } + + async fn visible(&self) -> Vec { + let mut kept: Vec = vec![]; + for record in self.items().await { + 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 producers(&self) -> Vec { + self.items() + .await + .into_iter() + .filter_map(|record| match record.item { + Some(Item::Data(data)) => data.producer, + Some(Item::Close(close)) => close.producer, + _ => None, + }) + .collect() + } +} + +/// A client scope writes as `client::`, and its close comes after its Records. +async fn a_client_scope_publishes_and_closes_as_one_producer() { + for case in cases().await { + let opened = case.client_scope("", 0).unwrap(); + let producer = opened.producer.unwrap(); + assert!( + producer.id.starts_with("client:me%40host:"), + "{}", + producer.id + ); + assert_eq!(producer.epoch.unwrap().attempt, 1); + let (first, last) = case.publish(opened.scope_id, &["a", "b"]).await.unwrap(); + assert!(0 < first && first < last); + case.close_as(opened.scope_id).await.unwrap(); + let items = case.items().await; + assert!(matches!(items.last().unwrap().item, Some(Item::Close(_)))); + assert!(case.producers().await.iter().all(|p| *p == producer)); + assert_eq!(case.visible().await, ["a", "b"]); + // The scope's writer closed with the Stream, so a later publish opens a new one. + let error = case.publish(opened.scope_id, &["late"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_closed"); + } +} + +/// Each client scope with no name has its own Producer, so two clients never share a sequence. +async fn two_client_scopes_are_two_producers() { + for case in cases().await { + let a = case.client_scope("", 0).unwrap(); + let b = case.client_scope("", 0).unwrap(); + assert_ne!(a.scope_id, b.scope_id); + assert_ne!(a.producer.unwrap().id, b.producer.unwrap().id); + case.publish(a.scope_id, &["a"]).await.unwrap(); + case.publish(b.scope_id, &["b"]).await.unwrap(); + case.publish(a.scope_id, &["c"]).await.unwrap(); + assert_eq!(case.visible().await, ["a", "b", "c"]); + } +} + +/// A named scope at a higher epoch fences the earlier one, and withdraws none of its Records. +async fn a_named_client_scope_fences_its_earlier_epoch() { + for case in cases().await { + let old = case.client_scope("ingest", 1).unwrap(); + assert_eq!(old.producer.as_ref().unwrap().id, "client:ingest"); + case.publish(old.scope_id, &["old"]).await.unwrap(); + let new = case.client_scope("ingest", 2).unwrap(); + case.publish(new.scope_id, &["new"]).await.unwrap(); + let error = case.publish(old.scope_id, &["late"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_fenced"); + assert_eq!(case.visible().await, ["old", "new"]); + let error = case.client_scope("ingest", 0).unwrap_err(); + assert_eq!(reason(&error), "stream_invalid"); + } +} + +/// The Producer of an Activity scope is the one that `DeriveProducer` gives. +async fn an_activity_scope_writes_as_its_derived_producer() { + for case in cases().await { + let scope = case.activity_scope(activity(1), &[]).await; + case.publish(scope, &["a1"]).await.unwrap(); + let derived: DeriveProducerResponse = case + .call_sync( + "DeriveProducer", + DeriveProducerRequest { + context: Some(activity(1)), + purpose: ProducerPurpose::ActivityWriter as i32, + }, + ) + .unwrap(); + assert_eq!(case.producers().await, vec![derived.producer.unwrap()]); + let closed = case.close_scope(scope).await.unwrap(); + assert_eq!(closed, CloseScopeResponse::default()); + } +} + +/// A retry's scope withdraws the earlier attempt's Records on each Stream of its arguments, even +/// if the retry writes nothing. A missing Stream there doesn't fail the open. +async fn an_activity_retry_scope_withdraws_the_streams_of_its_arguments() { + for case in cases().await { + let first = case.activity_scope(activity(1), &[]).await; + case.publish(first, &["a1"]).await.unwrap(); + case.close_scope(first).await.unwrap(); + let missing = format!("{}-missing", case.stream); + let retry = case + .activity_scope(activity(2), &[&case.stream, &case.stream, &missing]) + .await; + assert!(case.visible().await.is_empty()); + case.publish(retry, &["a2"]).await.unwrap(); + assert_eq!(case.visible().await, ["a2"]); + } +} + +/// An Activity doesn't close a Stream, and its refusal calls no Store. +async fn an_activity_scope_doesnt_close_a_stream() { + for case in cases().await { + let scope = case.activity_scope(activity(1), &[]).await; + let error = case.close_as(scope).await.unwrap_err(); + assert_eq!(reason(&error), "stream_not_owner"); + let described: DescribeStreamResponse = case + .call( + "DescribeStream", + DescribeStreamRequest { + namespace: case.namespace.clone(), + stream: Some(case.stream_ref()), + }, + ) + .await + .unwrap(); + assert_eq!(described.status(), StreamStatus::Open); + } +} + +/// A Local Activity scope sends its run's task facts, so its first write withdraws the Records of +/// a Local Activity whose task failed. +async fn a_local_activity_scope_withdraws_the_records_of_a_failed_task() { + for case in cases().await { + let orphan = case + .activity_scope(local_activity("1", 3, 2, vec![]), &[]) + .await; + case.publish(orphan, &["orphan"]).await.unwrap(); + case.close_scope(orphan).await.unwrap(); + let next = case + .activity_scope(local_activity("2", 5, 4, vec![3]), &[]) + .await; + case.publish(next, &["next"]).await.unwrap(); + assert_eq!(case.visible().await, ["next"]); + } +} + +/// The two first publishes of a scope to a Stream share one writer and one sequence. +async fn concurrent_first_publishes_share_one_writer() { + for case in cases().await { + let scope = case.activity_scope(activity(2), &[]).await; + let (a, b) = tokio::join!(case.publish(scope, &["a"]), case.publish(scope, &["b"])); + a.unwrap(); + b.unwrap(); + let mut visible = case.visible().await; + visible.sort(); + assert_eq!(visible, ["a", "b"]); + assert_eq!(case.close_scope(scope).await.unwrap().failed_writers, 0); + } +} + +/// The close of a scope reports a failed writer instead of failing, and frees the scope. +async fn a_scope_close_reports_a_failed_writer_and_frees_the_scope() { + for case in cases().await { + let scope = case.client_scope("", 0).unwrap().scope_id; + case.publish(scope, &["a"]).await.unwrap(); + let other = case.client_scope("", 0).unwrap().scope_id; + case.close_as(other).await.unwrap(); + assert_eq!( + reason(&case.publish(scope, &["late"]).await.unwrap_err()), + "stream_closed" + ); + let closed = case.close_scope(scope).await.unwrap(); + assert_eq!(closed.failed_writers, 1); + assert_eq!(closed.failure.unwrap().reason(), StreamErrorReason::Closed); + let error = case.publish(scope, &["x"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_unknown_writer"); + assert!(case.close_scope(scope).await.is_err()); + } +} + +/// A freed scope releases its writers with no wait and is gone. +async fn a_freed_scope_is_gone() { + for case in cases().await { + let scope = case.client_scope("", 0).unwrap().scope_id; + case.publish(scope, &["a"]).await.unwrap(); + let _: FreeResponse = case + .call_sync("FreeScope", FreeScopeRequest { scope_id: scope }) + .unwrap(); + let error = case.publish(scope, &["x"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_unknown_writer"); + let error = case + .call_sync::<_, FreeResponse>("FreeScope", FreeScopeRequest { scope_id: scope }) + .unwrap_err(); + assert_eq!(reason(&error), "stream_unknown_writer"); + assert_eq!(case.visible().await, ["a"]); + } +} + +/// The shutdown closes every scope. Then every call fails, except the `Free` methods, which do +/// nothing, and the static ones. +async fn the_shutdown_closes_every_scope_and_refuses_later_calls() { + for case in cases().await { + let client = case.client_scope("", 0).unwrap().scope_id; + let attempt = case.activity_scope(activity(1), &[]).await; + case.publish(client, &["c"]).await.unwrap(); + case.publish(attempt, &["a"]).await.unwrap(); + let shut: ShutdownResponse = case.call("Shutdown", ShutdownRequest {}).await.unwrap(); + assert_eq!(shut.failed_writers, 0); + let error = case.publish(client, &["x"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_unsupported"); + assert_eq!( + reason(&case.client_scope("", 0).unwrap_err()), + "stream_unsupported" + ); + let _: FreeResponse = case + .call_sync("FreeScope", FreeScopeRequest { scope_id: client }) + .unwrap(); + case.ext.free_writer(u64::MAX).unwrap(); + let _: DeriveProducerResponse = case + .call_sync( + "DeriveProducer", + DeriveProducerRequest { + context: Some(activity(1)), + purpose: ProducerPurpose::ActivityWriter as i32, + }, + ) + .unwrap(); + } +} + +/// Lang's `CloseWriters` closes only the writers of `OpenWriter`. A scope keeps its own, since +/// another Worker of the client may still write through it. +async fn close_writers_leaves_the_writers_of_scopes() { + for case in cases().await { + let scope = case.client_scope("", 0).unwrap().scope_id; + case.publish(scope, &["a"]).await.unwrap(); + let closed: CloseWritersResponse = case + .call("CloseWriters", CloseWritersRequest {}) + .await + .unwrap(); + assert_eq!(closed.failed, 0); + case.publish(scope, &["b"]).await.unwrap(); + assert_eq!(case.close_scope(scope).await.unwrap().failed_writers, 0); + assert_eq!(case.visible().await, ["a", "b"]); + } +} + +/// A late `FreeWriter` from lang, after `CloseWriters`, takes no handle of a scope that shares the +/// writer. +async fn a_late_lang_free_leaves_the_writer_of_a_scope() { + for case in cases().await { + let opened = case.client_scope("shared", 1).unwrap(); + case.publish(opened.scope_id, &["a"]).await.unwrap(); + let writer: OpenWriterResponse = case + .call( + "OpenWriter", + OpenWriterRequest { + namespace: case.namespace.clone(), + stream: Some(case.stream_ref()), + producer: opened.producer.clone(), + ..Default::default() + }, + ) + .await + .unwrap(); + let _: CloseWritersResponse = case + .call("CloseWriters", CloseWritersRequest {}) + .await + .unwrap(); + let free = FreeWriterRequest { + writer_id: writer.writer_id, + }; + let error = case + .call_sync::<_, FreeWriterResponse>("FreeWriter", free) + .unwrap_err(); + assert_eq!(reason(&error), "stream_unknown_writer"); + case.publish(opened.scope_id, &["b"]).await.unwrap(); + assert_eq!(case.visible().await, ["a", "b"]); + } +} + +/// The cases of the suite, in file order. +pub const CASES: &[super::Case] = &[ + case!(a_client_scope_publishes_and_closes_as_one_producer), + case!(two_client_scopes_are_two_producers), + case!(a_named_client_scope_fences_its_earlier_epoch), + case!(an_activity_scope_writes_as_its_derived_producer), + case!(an_activity_retry_scope_withdraws_the_streams_of_its_arguments), + case!(an_activity_scope_doesnt_close_a_stream), + case!(a_local_activity_scope_withdraws_the_records_of_a_failed_task), + case!(concurrent_first_publishes_share_one_writer), + case!(a_scope_close_reports_a_failed_writer_and_frees_the_scope), + case!(a_freed_scope_is_gone), + case!(the_shutdown_closes_every_scope_and_refuses_later_calls), + case!(close_writers_leaves_the_writers_of_scopes), + case!(a_late_lang_free_leaves_the_writer_of_a_scope), +]; diff --git a/crates/streams/src/testing/support.rs b/crates/streams/src/testing/support.rs index cae7fe7eb..ba57b6cf4 100644 --- a/crates/streams/src/testing/support.rs +++ b/crates/streams/src/testing/support.rs @@ -5,7 +5,7 @@ use super::{RawKeys, backends}; use crate::{ - StreamResult, StreamRuntime, + StreamError, StreamResult, StreamRuntime, digest::batch_digest, epoch::withdraws, ids::local_activity_producer_id, @@ -13,13 +13,19 @@ use crate::{ AppendRecord, Producer, ProducerEpoch, StreamClose, StreamData, StreamProducerKind, StreamRecord, StreamRef, Supersession, stream_record::Item, }, - store::{Appended, Batch, Closed, Entry, Owner, Store, Withdrawal}, + store::{ + Appended, Batch, Capabilities, Closed, Created, Entry, Owner, Store, StreamState, + Withdrawal, + }, }; use prost::Message; use prost_types::Timestamp; -use std::sync::{ - Arc, - atomic::{AtomicU64, Ordering}, +use std::{ + sync::{ + Arc, + atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}, + }, + time::Duration, }; use temporalio_common::protos::temporal::api::{common::v1::Execution, enums::v1::ExecutionType}; @@ -400,3 +406,112 @@ pub(crate) fn named_runtimes(backend: &str) -> Vec> { .map(|test_backend| test_backend.runtime(backend)) .collect() } + +/// 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. +pub(crate) struct Flaky { + pub(crate) inner: Arc, + pub(crate) lose_next: AtomicBool, + pub(crate) down: AtomicBool, + /// The appends that it takes before it goes down. Below zero is never. + pub(crate) down_after: AtomicI64, +} + +impl Flaky { + /// A Store that is up, over `inner`. + pub(crate) fn new(inner: Arc) -> Self { + Self { + inner, + lose_next: AtomicBool::new(false), + down: AtomicBool::new(false), + down_after: AtomicI64::new(-1), + } + } +} + +#[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_after.fetch_sub(1, Ordering::SeqCst) == 0 { + self.down.store(true, Ordering::SeqCst); + } + 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 + } +} diff --git a/crates/streams/src/testing/writer.rs b/crates/streams/src/testing/writer.rs index 61359412b..a9e116832 100644 --- a/crates/streams/src/testing/writer.rs +++ b/crates/streams/src/testing/writer.rs @@ -4,10 +4,10 @@ use super::{ case, - support::{Needs, stores_for}, + support::{Flaky, Needs, stores_for}, }; use crate::{ - MemoryStreamRuntime, StreamError, StreamResult, StreamRuntime, + MemoryStreamRuntime, StreamError, StreamRuntime, epoch::withdraws, ids::{ activity_producer_id, client_producer_id, local_activity_producer_id, @@ -20,10 +20,6 @@ use crate::{ 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; @@ -432,109 +428,6 @@ 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. -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)> { @@ -663,6 +556,20 @@ async fn a_local_activity_writer_withdraws_the_records_of_a_failed_task() { } } +/// A huge retry duration counts as a day, so that the retry math doesn't overflow. +async fn a_huge_retry_duration_doesnt_overflow() { + for case in cases().await { + let huge = Some(prost_types::Duration { + seconds: i64::MAX, + nanos: 0, + }); + let writer = case + .open(activity(2), |r| r.max_retry_duration = huge) + .await; + case.write(writer, &["a"]).await.unwrap(); + } +} + /// 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), @@ -680,4 +587,5 @@ pub const CASES: &[super::Case] = &[ 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), + case!(a_huge_retry_duration_doesnt_overflow), ]; diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index d0af10584..5c8a9ae0f 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -29,6 +29,7 @@ use tokio::time::Instant; const DEFAULT_BATCH: usize = 1000; const DEFAULT_RETRY: Duration = Duration::from_secs(10); +const LONGEST_RETRY: Duration = Duration::from_secs(24 * 60 * 60); 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 @@ -182,7 +183,27 @@ type WriterKey = (String, String, String, String); struct Open { writer: Arc, key: WriterKey, - handles: usize, + /// The handles of each holder. Lang's `CloseWriters` releases lang's only. + lang: usize, + scoped: usize, +} + +impl Open { + fn handles(&mut self, holder: Holder) -> &mut usize { + match holder { + Holder::Lang => &mut self.lang, + Holder::Scope => &mut self.scoped, + } + } +} + +/// Who holds a handle of a writer. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Holder { + /// Lang, through `OpenWriter`. + Lang, + /// A Producer scope. + Scope, } #[derive(Default)] @@ -193,20 +214,30 @@ 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. - 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 { + /// Releases one handle of a writer. Returns the writer when it has no handle left, so that the + /// caller ends it. + fn release(&mut self, id: u64, holder: Holder) -> StreamResult>> { + let unknown = || StreamError::new(Reason::UnknownWriter, format!("No writer {id}")); + let open = self.open.get_mut(&id).ok_or_else(unknown)?; + // A holder with no handle left, such as lang after `CloseWriters`, takes none of another. + let handles = open.handles(holder); + *handles = handles.checked_sub(1).ok_or_else(unknown)?; + if open.lang + open.scoped > 0 { return Ok(None); } - let open = self.open.remove(&id).expect("found above"); + Ok(Some(self.remove(id))) + } + + fn add_handle(&mut self, key: &WriterKey, holder: Holder) -> Option { + let id = self.by_producer.get(key).copied()?; + *self.open.get_mut(&id).expect("indexed").handles(holder) += 1; + Some(id) + } + + fn remove(&mut self, id: u64) -> Arc { + let open = self.open.remove(&id).expect("found by the caller"); self.by_producer.remove(&open.key); - Ok(Some(open.writer)) + open.writer } } @@ -226,6 +257,7 @@ impl Writers { &self, service: &Arc, request: OpenWriterRequest, + holder: Holder, ) -> StreamResult { let stream = request .stream @@ -245,7 +277,7 @@ impl Writers { producer.id.clone(), crate::epoch::epoch_key(&epoch), ); - if let Some(id) = self.share(&key) { + if let Some(id) = self.share(&key, holder) { return Ok(id); } let writer = Arc::new(Writer { @@ -258,7 +290,9 @@ impl Writers { n if n > 0 => n as usize, _ => DEFAULT_BATCH, }, - max_retry: duration(request.max_retry_duration).unwrap_or(DEFAULT_RETRY), + // A retry of more than a day counts as a day, so that `Instant` math can't overflow. + max_retry: (duration(request.max_retry_duration).unwrap_or(DEFAULT_RETRY)) + .min(LONGEST_RETRY), service: service.clone(), state: Mutex::default(), writing: tokio::sync::Mutex::new(()), @@ -286,8 +320,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; + if let Some(id) = table.add_handle(&key, holder) { return Ok(id); } let id = self.next.fetch_add(1, Ordering::Relaxed) + 1; @@ -297,18 +330,21 @@ impl Writers { Open { writer, key, - handles: 1, + lang: usize::from(holder == Holder::Lang), + scoped: usize::from(holder == Holder::Scope), }, ); Ok(id) } + #[cfg(test)] + pub(crate) fn open_count(&self) -> usize { + self.table.lock().open.len() + } + /// 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 share(&self, key: &WriterKey, holder: Holder) -> Option { + self.table.lock().add_handle(key, holder) } fn get(&self, id: u64) -> StreamResult> { @@ -344,15 +380,27 @@ impl Writers { writer.write(records).await } - /// Closes every open writer, whatever its handles. Returns how many had failed. + /// Releases every handle that lang holds, and ends each writer with no handle left. Returns + /// how many of those had failed. Scopes release their own handles when they end. pub(crate) async fn close_all(&self) -> usize { - let ids: Vec = self.table.lock().open.keys().copied().collect(); + let ended: Vec> = { + let mut table = self.table.lock(); + let ids: Vec = table.open.keys().copied().collect(); + let mut ended = vec![]; + for id in ids { + let open = table.open.get_mut(&id).expect("listed"); + open.lang = 0; + if open.scoped == 0 { + let writer = table.remove(id); + // Closed at once, so a dropped call still ends every writer it took. + writer.state.lock().closed = true; + ended.push(writer); + } + } + ended + }; let mut failed = 0; - for id in ids { - let released = self.table.lock().release(id, true); - let Ok(Some(writer)) = released else { - continue; - }; + for writer in ended { if end(&writer).await.is_err() { failed += 1; } @@ -362,9 +410,9 @@ impl Writers { /// 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 { + pub(crate) async fn close(&self, id: u64, holder: Holder) -> StreamResult { let writer = self.get(id)?; - if self.table.lock().release(id, false)?.is_none() { + if self.table.lock().release(id, holder)?.is_none() { let state = writer.state.lock(); return match &state.failed { Some(error) => Err(error.clone()), @@ -376,8 +424,8 @@ impl Writers { /// 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)? { + pub(crate) fn free(&self, id: u64, holder: Holder) -> StreamResult<()> { + if let Some(writer) = self.table.lock().release(id, holder)? { writer.state.lock().closed = true; } Ok(())