[Brook K2] Added the Stream ids, offsets, epochs and batch digest, with vectors.

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:<run id>` 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;
 }
 

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.

crates/protos/protos/local/temporal/sdk/core/streams/streams.proto

@@ -20,9 +20,10 @@ import "temporal/api/stream/v1/message.proto";
 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 is 1000.
     int32 max_batch_size = 4;

crates/streams/Cargo.toml

@@ -11,9 +11,14 @@ repository = "https://github.com/temporalio/sdk-rust"
 keywords = ["temporal", "workflow", "streams"]
 categories = ["development-tools"]
 
+[features]
+# The test helpers, for the crates of other Stores.
+test-support = []
+
 [dependencies]
 prost = { workspace = true }
 prost-types = { workspace = true }
+sha2 = "0.10"
 thiserror = { workspace = true }
 tonic = { workspace = true }
 
@@ -22,5 +27,8 @@ path = "../common"
 version = "~1.1.0"
 default-features = false
 
+[dev-dependencies]
+serde_json = { workspace = true }
+
 [lints]
 workspace = true

crates/streams/README.md

@@ -4,7 +4,7 @@ Experimental. The Stream runtime that the Temporal SDKs link into their bridges.
 the runtime per client, as a plain object, and calls it with a method name and proto bytes. A
 Workflow makes each Stream call from a Local Activity. Core has no Stream code.
 
-## Protos
+## Protos and shared vectors
 
 The crate defines no protos. `temporalio-protos` generates them, and `temporalio_streams::proto`
 re-exports them, so a value needs no conversion between the crates.
@@ -14,3 +14,8 @@ re-exports them, so a value needs no conversion between the crates.
   `crates/protos/protos/api_upstream`.
 - `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.
+  `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`. `digests.json` is the
+  golden of the crate's batch digest.

crates/streams/src/digest.rs

@@ -0,0 +1,44 @@
+//! The content digest of a batch. Each dedupe entry stores it, so that a resend with the same
+//! `sequence` and other content fails instead of being dropped without an error. The Store compares
+//! the digest that this function gives, and `testdata/digests.json` has its expected values.
+
+use crate::proto::AppendRecord;
+use sha2::{Digest, Sha256};
+
+/// The length of a [batch_digest].
+pub const DIGEST_LEN: usize = 16;
+
+/// The first 16 bytes of a SHA-256 over the Records in order. Each part has its length before it,
+/// so that two different batches never hash the same bytes. The metadata is hashed in key order,
+/// since a map has no order on the wire.
+pub fn batch_digest(records: &[AppendRecord]) -> [u8; DIGEST_LEN] {
+    let mut hasher = Sha256::new();
+    for record in records {
+        // A Record with no body and one with an empty body are two different stored Records.
+        let Some(body) = &record.body else {
+            hasher.update([0]);
+            continue;
+        };
+        hasher.update([1]);
+        let mut metadata: Vec<_> = body.metadata.iter().collect();
+        metadata.sort();
+        hasher.update((metadata.len() as u64).to_be_bytes());
+        for (key, value) in metadata {
+            part(&mut hasher, key.as_bytes());
+            part(&mut hasher, value);
+        }
+        part(&mut hasher, &body.data);
+        hasher.update((body.external_payloads.len() as u64).to_be_bytes());
+        for external in &body.external_payloads {
+            hasher.update(external.size_bytes.to_be_bytes());
+        }
+    }
+    let mut digest = [0; DIGEST_LEN];
+    digest.copy_from_slice(&hasher.finalize()[..DIGEST_LEN]);
+    digest
+}
+
+fn part(hasher: &mut Sha256, bytes: &[u8]) {
+    hasher.update((bytes.len() as u64).to_be_bytes());
+    hasher.update(bytes);
+}