feat: add take, filter and reduce to a Stream, and a pipeline on top

.github/workflows/ci.yml

@@ -228,8 +228,8 @@ jobs:
       - run: uv tool install poethepoet
       - run: uv sync --all-extras
       - run: cargo install --locked nexgen --version 0.2.8 --features advanced --force
-      # The Stream package needs the `main` of nexgen, which no release has yet.
-      - run: cargo install --locked --git https://github.com/temporalio/nexgen --rev 9653a0f5a845df0c5059554f1b5813db6a74cd1a --features advanced --root "$RUNNER_TEMP/nexgen-stream" nexgen
+      # The Stream package needs `@nexus.helper`, which only a fork of nexgen has yet.
+      - run: cargo install --locked --git https://github.com/moetemp/nexgen --rev 4eab992027a449192a18ee3b3ab190a23e5ae174 --features advanced --root "$RUNNER_TEMP/nexgen-stream" nexgen
       - run: echo "NEX_GEN_STREAM_BIN=$RUNNER_TEMP/nexgen-stream/bin/nexgen" >> "$GITHUB_ENV"
       - run: poe build-develop
       - run: poe gen-protos

changelog/added/chained-pipeline-wren.md

@@ -0,0 +1 @@
+Add `temporalio.contrib.streams.pipeline`, which chains `filter`, `map`, `take_until` and `limit` steps over a Stream and runs the whole chain as one `take`, `filter` or `reduce` call. In Workflow code a chain is one Local Activity and one marker in History.

changelog/added/patient-reader-heron.md

@@ -0,0 +1,2 @@
+Add the `take`, `filter` and `reduce` methods to the client `Stream` and the Workflow `Stream`. Client, Activity and Workflow code call them to scan a Stream until a predicate matches, the Stream is done, or a limit is reached.
+Add `StreamConfig.workflow_read_replay_from`. With `StreamReplaySource.STORE`, a `take` or a `filter` of Workflow code keeps only the offsets and a digest of its Records in History, and a replay reads the Records from the Store again.

scripts/gen_nexus_system_api.py

@@ -28,8 +28,9 @@
 workflow_init_path = base_dir / "temporalio" / "workflow" / "__init__.py"
 NEX_GEN_VERSION = "0.2.8"
 # The Stream package needs a newer nexgen, which rejects the other WITs. One pin is enough
-# again once the api WITs and a nexgen release agree. No release has nexgen commit 9653a0f5,
-# the `main` of nexgen, yet, so NEX_GEN_STREAM_BIN points at a build of it.
+# again once the api WITs and a nexgen release agree. The Stream WIT uses `@nexus.helper`,
+# which only the fork https://github.com/moetemp/nexgen has, at commit 4eab9920. So
+# NEX_GEN_STREAM_BIN points at a build of that commit.
 NEX_GEN_STREAM_VERSION = "0.2.12"
 proto_files = [
     gen_protos.api_proto_dir
@@ -87,7 +88,8 @@ def nex_gen_stream_command() -> list[str]:
     bin_path = os.environ.get("NEX_GEN_STREAM_BIN")
     if not bin_path:
         raise RuntimeError(
-            "NEX_GEN_STREAM_BIN must point at a nexgen build of commit 9653a0f5"
+            "NEX_GEN_STREAM_BIN must point at a nexgen build of commit 4eab9920 "
+            "of https://github.com/moetemp/nexgen"
         )
     version = subprocess.check_output([bin_path, "--version"], text=True).strip()
     if version != f"nexgen {NEX_GEN_STREAM_VERSION}":

temporalio/client/__init__.py

@@ -187,6 +187,7 @@
 )
 from ._stream_store import (
     StreamConfig,
+    StreamReplaySource,
     StreamSelectContext,
     StreamStore,
     StreamStoreCallOptions,
@@ -357,6 +358,7 @@
     "CloudOperationsClient",
     "Plugin",
     "StreamConfig",
+    "StreamReplaySource",
     "StreamSelectContext",
     "StreamStore",
     "StreamStoreCallOptions",

temporalio/client/_impl.py

@@ -24,6 +24,7 @@
 import temporalio.api.errordetails.v1
 import temporalio.api.failure.v1
 import temporalio.api.schedule.v1
+import temporalio.api.streamservice.v1
 import temporalio.api.taskqueue.v1
 import temporalio.api.update.v1
 import temporalio.api.workflowservice.v1
@@ -126,6 +127,7 @@
     ScheduleUpdate,
     ScheduleUpdateInput,
 )
+from ._stream_read import _note_page_sizes
 from ._stream_store import (
     _STREAM_SERVICE,
     StreamConfig,
@@ -1686,6 +1688,8 @@ async def _start_stream_store_operation(
         response = await _call_stream_store(
             store, data_converter, input.operation, request
         )
+        if isinstance(response, temporalio.api.streamservice.v1.ReadRecordsResponse):
+            _note_page_sizes(response)
         handle: NexusOperationHandle[Any] = NexusOperationHandle(
             self._client,
             input.id,