[Brook K10] Removed the owner and the withdrawal, and added the Stream close item.

crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto

@@ -20,6 +20,16 @@ enum StreamStatus {
     STREAM_STATUS_EXPIRED = 3;
 }
 
+// Experimental. Might change in the future.
+// How a Stream closed. CLOSED comes from a Producer that called close(). Native Streams add their
+// own close statuses later.
+// (-- api-linter: core::0216::synonyms=disabled
+//     aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --)
+enum StreamCloseStatus {
+    STREAM_CLOSE_STATUS_UNSPECIFIED = 0;
+    STREAM_CLOSE_STATUS_CLOSED = 1;
+}
+
 // Experimental. Might change in the future.
 // Where a read starts when it has no from_offset.
 enum StreamReadStart {
@@ -49,16 +59,14 @@ enum StreamErrorReason {
     STREAM_ERROR_REASON_EXPIRED = 2;
     // The Stream is closed. FAILED_PRECONDITION.
     STREAM_ERROR_REASON_CLOSED = 3;
-    // The Producer's epoch is lower than the newest one, its Workflow Task was withdrawn, or a
-    // Workflow retry withdrew its run. FAILED_PRECONDITION.
+    // The Producer's epoch is lower than the newest one. FAILED_PRECONDITION.
     STREAM_ERROR_REASON_FENCED = 4;
     // The batch has a lower `sequence` than the last one of its epoch. FAILED_PRECONDITION.
     STREAM_ERROR_REASON_SEQUENCE_TOO_LOW = 5;
     // The batch has the `sequence` of the last one of its epoch, and other content.
     // FAILED_PRECONDITION.
     STREAM_ERROR_REASON_SEQ_DIVERGENT = 6;
-    // A create gives another owner than the Stream has, or the owner doesn't allow this close.
-    // FAILED_PRECONDITION.
+    // The caller may not create or close this Stream. FAILED_PRECONDITION.
     STREAM_ERROR_REASON_NOT_OWNER = 7;
     // A newer format version wrote the Stream. FAILED_PRECONDITION.
     STREAM_ERROR_REASON_FORMAT_TOO_NEW = 8;

crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto

@@ -18,11 +18,6 @@ import "temporal/api/enums/v1/stream.proto";
 // Identifies one Stream. The id is opaque to callers.
 message StreamRef {
     string id = 1;
-    // Unset for a standalone Stream. For a Stream bound to a Workflow, business_id is the Workflow
-    // ID and run_id is the first execution run id of its run chain. For a Standalone Activity,
-    // business_id is the activity ID and run_id is its run id. A create and a close of a
-    // Workflow-owned Stream must carry its owner.
-    temporal.api.common.v1.Execution owner = 2;
     // The name of the backend that holds the Stream, as the Worker and client config register
     // it, so every process routes a ref it receives. Empty is the default backend. The Store
     // never reads it.
@@ -65,7 +60,7 @@ message ClientEpoch {
 }
 
 // Experimental. Might change in the future.
-// A Workflow run Producer. It writes in its Workflow Task. A higher epoch withdraws the Records of
+// A Workflow run Producer. It writes in its Workflow Task. A higher epoch supersedes 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.
@@ -74,15 +69,15 @@ message WorkflowProducer {
 }
 
 // Experimental. Might change in the future.
-// A Local Activity Producer. A higher epoch withdraws the Producer's lower epochs.
+// A Local Activity Producer. A higher epoch supersedes 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.
+// supersedes the Producer's lower epochs.
 message ActivityProducer {
     string run_id = 1;
     ActivityEpoch epoch = 2;
@@ -97,7 +92,7 @@ message ClientProducer {
 
 // 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
+// its epochs and picks which earlier Records a higher epoch supersedes. A run-based kind carries its
 // run, so a client can't.
 message Producer {
     // Stable across attempts, such as `wf:<run id>` for the Workflow Producer of a run.
@@ -113,30 +108,11 @@ message Producer {
 }
 
 // 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.
+// The Producer of a write. An append and a close carry the same writer, so the Store orders each
+// write the same way.
 message StreamWriter {
-    // Required on a write. Its Producer's kind picks the withdrawal rule.
+    // Required on a write.
     Producer producer = 1;
-    // 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.
@@ -145,35 +121,34 @@ message AppendRecord {
     temporal.api.common.v1.Payload body = 1;
 }
 
-// Experimental. Might change in the future.
-// Which earlier Records a retried Producer withdraws. The Store keeps these facts for the first
-// 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, with its withdrawal facts.
-    StreamWriter writer = 1;
-}
-
 // Experimental. Might change in the future.
 // One Record that a Producer appended.
 message StreamData {
     Producer producer = 1;
     temporal.api.common.v1.Payload body = 2;
 }
 
+// Experimental. Might change in the future.
+// The close of a Stream. A Stream has at most one, and a read returns it last.
+message StreamClose {
+    // The Producer that closed the Stream.
+    Producer producer = 1;
+    reserved 2, 3;
+    reserved "result", "failure";
+    // Why the Stream closed.
+    temporal.api.enums.v1.StreamCloseStatus status = 4;
+}
+
 // Experimental. Might change in the future.
 // An offset is ordered and unique in one Stream, and it is sparse. Offsets have gaps, so they
 // don't count Records. Zero is never an offset. A read gives its Stream, so the read checks the
 // offset against that Stream.
 message StreamRecord {
     int64 offset = 1;
-    // A read item is a Record or a Superseded item. A close writes no item.
+    // A read item is a Record or the close.
     oneof item {
         StreamData data = 2;
-        // The Store keeps only the withdrawal facts of the first batch of a higher epoch. A read
-        // builds this item from them. It withdraws the earlier Records that the facts match, and a
-        // reader drops them.
-        StreamSuperseded superseded = 3;
+        StreamClose close = 4;
     }
 }
 

crates/protos/protos/api_upstream/temporal/api/streamservice/v1/request_response.proto

@@ -20,31 +20,19 @@ import "temporal/api/stream/v1/message.proto";
 // Experimental. Might change in the future.
 // Creates the Stream. When the Stream exists, FAIL fails with ALREADY_EXISTS, and USE_EXISTING
 // gets it with created false and its own Retention. A retry of the create that made the Stream,
-// with the same request_id, gets created true. Each policy fails with CLOSED on a closed Stream,
-// EXPIRED on an expired one, and NOT_OWNER on one with another owner.
+// with the same request_id, gets created true. Each policy fails with CLOSED on a closed Stream
+// and EXPIRED on an expired one.
 // (-- api-linter: core::0133=disabled
 //     aip.dev/not-precedent: The request identifies a Stream by namespace and ref. --)
 message CreateStreamRequest {
     string namespace = 1;
-    // The ref of the Stream to create. Its id is the Stream id, and is never empty. A bound
-    // Stream's ref has the owner's run chain as its owner.
+    // The ref of the Stream to create. Its id is the Stream id, and is never empty.
     temporal.api.stream.v1.StreamRef stream_ref = 2;
     // The age after which a Record is dropped. Empty is the Store default.
     google.protobuf.Duration retention = 3;
-    // Workflow only. The previous run on a retry run's first create, to withdraw its Records.
-    string withdraw_run = 4;
-    // The Workflow run that owns a bound Stream, with its run_id and a Workflow epoch. It needs a
-    // ref whose owner is the run's chain. Unset for a standalone Stream.
-    temporal.api.stream.v1.Producer owner = 5;
-    // Only read by the server. When set, the server closes the Stream when the owner's run chain
-    // ends, also on a terminate or a timeout. It needs owner, and a ref with a Workflow owner.
-    // (-- api-linter: core::0140::prepositions=disabled
-    //     aip.dev/not-precedent: "with" ties the close to the owner's run chain. --)
-    bool close_with_run = 6;
     // Only read by the server, for a Stream bound to a Standalone Activity. The server checks that
     // the token is the Activity's current attempt, so a caller can't bind a Stream to a Standalone
-    // Activity that it isn't. The server knows the run of a Workflow's call, and a standalone
-    // Stream has no owner, so neither needs a token.
+    // Activity that it isn't.
     bytes task_token = 7;
     // What the create does when the Stream exists. Unspecified is FAIL.
     temporal.api.enums.v1.StreamIdConflictPolicy id_conflict_policy = 8;
@@ -56,7 +44,7 @@ message CreateStreamRequest {
 // Experimental. Might change in the future.
 // The Stream that a create made or got.
 message CreateStreamResponse {
-    // The ref, with the owner of a bound Stream.
+    // The ref of the Stream.
     temporal.api.stream.v1.StreamRef ref = 1;
     // False when USE_EXISTING got a Stream that existed.
     bool created = 2;
@@ -74,9 +62,6 @@ message AppendRecordsRequest {
     // the next number of the epoch. Such a batch has no dedupe, so a retry writes it again.
     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 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.
     bytes task_token = 7;
@@ -92,13 +77,10 @@ message AppendRecordsResponse {
 // Experimental. Might change in the future.
 // Closes a Stream. Readers drain what is left, then end, and later appends fail with CLOSED. A
 // close writes no Record and carries no value, so code that has a last value publishes it first.
-// Only the owner's run chain closes a bound Stream. Only a Workflow or a client closes a
-// standalone Stream.
 message CloseStreamRequest {
     string namespace = 1;
     temporal.api.stream.v1.StreamRef ref = 2;
-    // The writer that closes the Stream. An SDK Store reads only its Producer, and the server
-    // also reads its withdrawal fields.
+    // The writer that closes the Stream. A Store reads only its Producer.
     temporal.api.stream.v1.StreamWriter writer = 3;
     // Only read by the server. It checks the token as on the append.
     bytes task_token = 4;
@@ -122,8 +104,7 @@ message ReadRecordsRequest {
     int64 from_offset = 3;
     // Where to start when from_offset is zero. Unspecified is the beginning.
     temporal.api.enums.v1.StreamReadStart start = 4;
-    // The most items in one page, Superseded items included. Zero is 1000. A Superseded item and
-    // the Record after it stay in one page, so a page of one Record can hold two items.
+    // The most Records in one page. Zero is 1000.
     int32 max_records = 5;
     // How long the read waits for a new Record before it answers with an empty page. Empty answers
     // at once, and a wait longer than 60 s waits 60 s.
@@ -180,7 +161,7 @@ message DescribeStreamRequest {
 // The state of a Stream.
 message DescribeStreamResponse {
     temporal.api.enums.v1.StreamStatus status = 1;
-    // The newest offset, of a Record or a Superseded item. Zero when the Stream has none.
+    // The newest offset, of a Record. Zero when the Stream has none.
     int64 last_offset = 2;
     google.protobuf.Duration retention = 3;
 }

crates/streams/README.md

@@ -15,11 +15,10 @@ re-exports them, so a value needs no conversion between the crates.
 - `crates/protos/protos/local/temporal/sdk/core/streams/streams.proto`, in the package
   `coresdk.streams`, has the crate's own calls, such as the writer calls.
 - `testdata/*.json` are the shared vectors. They cover Stream ids, Producer ids, Store keys and
-  meta fields, offsets, epoch keys, the withdrawal rule and the batch digest of a dedupe entry.
+  meta fields, offsets, epoch keys and the batch digest of a dedupe entry.
   `testdata/gen_vectors.py` writes them from a second implementation, and the crate's tests read
-  each file. The Python SDK also reads `ids.json` and `withdrawals.json`. Each Store reads
-  `fields.json` through the shared Store suite. `digests.json` is the golden of the crate's batch
-  digest.
+  each file. The Python SDK also reads `ids.json`. Each Store reads `fields.json` through the
+  shared Store suite. `digests.json` is the golden of the crate's batch digest.
 
 ## The runtime
 

crates/streams/src/epoch.rs

@@ -1,15 +1,12 @@
-//! Producer epochs, their order, and the withdrawal rule of a `Superseded` item.
+//! Producer epochs and their order.
 //!
 //! A Store keeps an epoch as a key of fixed-width big-endian fields, in the order of comparison.
 //! A byte comparison of two keys then orders the epochs field by field. `testdata/epochs.json` has
 //! the shared vectors of the keys, in hex.
 
 use crate::{
     StreamError, StreamResult,
-    proto::{
-        Producer, StreamRecord, StreamSuperseded, StreamWriter, WorkflowTaskEpoch, producer::Kind,
-        stream_record::Item,
-    },
+    proto::{Producer, WorkflowTaskEpoch, producer::Kind},
 };
 use prost_types::Timestamp;
 use std::cmp::Ordering;
@@ -66,30 +63,6 @@ pub fn producer_run_id(producer: &Producer) -> &str {
     }
 }
 
-/// 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].
 pub const TASK_KEY_LEN: usize = 24;
 /// The length of an [epoch_key].
@@ -161,79 +134,6 @@ pub fn compare_epochs(a: Option<&Kind>, b: Option<&Kind>) -> Ordering {
     epoch_key(a).cmp(&epoch_key(b))
 }
 
-fn compare_tasks(a: &WorkflowTaskEpoch, b: &WorkflowTaskEpoch) -> Ordering {
-    task_key(a).cmp(&task_key(b))
-}
-
-/// Whether a `Superseded` item withdraws an earlier Record. The case of the epoch of the item's
-/// writer picks the rule.
-///
-/// For an Activity or a Local Activity, it withdraws that Producer's Records with a lower epoch.
-/// For a Workflow task t with the previous completed task c, it withdraws the Records of the run
-/// and of its Local Activities whose task is in `failed_tasks` or strictly between c and t, and
-/// those at c or at t with a lower epoch. The item of a retry run withdraws every Record of the
-/// previous run. A client's newer epoch fences the earlier ones but withdraws nothing, since a
-/// restarted client's earlier Records are real messages, not a failed attempt's.
-pub fn withdraws(superseded: &StreamSuperseded, record: &StreamRecord) -> bool {
-    let Some(writer) = superseded.writer.as_ref() else {
-        return false;
-    };
-    let Some(Item::Data(data)) = &record.item else {
-        return false;
-    };
-    let Some(producer) = data.producer.as_ref() else {
-        return false;
-    };
-    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 (new, old) = (withdrawing.kind.as_ref(), producer.kind.as_ref());
-    let (Some(new_kind), Some(old_kind)) = (new, old) else {
-        return false;
-    };
-    match ProducerKind::from(new_kind) {
-        ProducerKind::Client => return false,
-        ProducerKind::Workflow => {}
-        same => {
-            return ProducerKind::from(old_kind) == same
-                && producer.id == withdrawing.id
-                && compare_epochs(old, new) == Ordering::Less;
-        }
-    }
-    if !of_run(producer, withdrawing) {
-        return false;
-    }
-    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(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_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 kind_of(producer) {
-        Some(ProducerKind::Workflow) => producer.id == withdrawing.id,
-        Some(ProducerKind::LocalActivity) => {
-            producer_run_id(producer) == producer_run_id(withdrawing)
-        }
-        _ => false,
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;