From 7850f4957dfaf73b97fc8972255e6780c272d30a Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 16:46:00 -0700 Subject: [PATCH 1/7] Documented the Redis provider's guarantees and setup. Durability depends on the customer's Redis settings, and access control is Redis's job, so the provider's promises, the settings they need and an ACL that was tested with the conformance suite are written down. --- temporalio/contrib/streams/README.md | 4 + .../contrib/streams/docs/redis-guarantees.md | 122 ++++++++++++++++++ 2 files changed, 126 insertions(+) create mode 100644 temporalio/contrib/streams/docs/redis-guarantees.md diff --git a/temporalio/contrib/streams/README.md b/temporalio/contrib/streams/README.md index a2d087ade7..45a54994f9 100644 --- a/temporalio/contrib/streams/README.md +++ b/temporalio/contrib/streams/README.md @@ -101,3 +101,7 @@ Both samples run in full in `tests/contrib/streams/samples/`. `RedisStreams.delete_workflow_streams` removes them at once. - **Not in this release.** Reading a stream inside a Workflow, and streams owned by an Activity or by no one, raise `StreamUnsupportedError`. + +## Further reading + +- [Redis provider: guarantees and setup](docs/redis-guarantees.md) diff --git a/temporalio/contrib/streams/docs/redis-guarantees.md b/temporalio/contrib/streams/docs/redis-guarantees.md new file mode 100644 index 0000000000..d5520f01d8 --- /dev/null +++ b/temporalio/contrib/streams/docs/redis-guarantees.md @@ -0,0 +1,122 @@ +# Redis provider: guarantees and setup + +> This package is experimental and may change in future versions. + +Stream records live in your Redis. Temporal never sees them, so what a +stream guarantees is what your Redis guarantees. This page says what the +provider promises, which Redis settings those promises depend on, and who +is responsible for what. + +## What an acknowledged append means + +An append returns once one Lua script has written the whole batch to the +stream's log. A batch is written in full or not at all, in order. A retry +of the newest batch on the same producer returns the original position and +writes nothing. + +A Workflow's own publish is staged first and made visible only after +Temporal accepts the Workflow Task that published it. A task that fails +publishes nothing. + +How long an acknowledged record survives a failure depends on persistence: + +| Redis setting | What an acknowledged record survives | +|---|---| +| `appendonly yes`, `appendfsync always` | A Redis process crash and a host power loss. Every write waits for an fsync. | +| `appendonly yes`, `appendfsync everysec` (Redis default for AOF) | A Redis process crash. A host power loss can lose about the last second of writes. | +| RDB snapshots only | Nothing written since the last snapshot. | +| No persistence | Nothing, after a restart. | + +Replicas copy writes asynchronously. A failover can lose writes the primary +acknowledged but had not yet sent to the replica. The provider does not use +`WAIT`. If a failover must not lose acknowledged records, use a store with +a durable write log, such as Amazon MemoryDB, or `appendfsync always` on a +primary without automatic failover. + +Recommended for production: `appendonly yes` with `appendfsync everysec` or +`always`, at least one replica, and the settings below. + +## Settings the provider needs + +- **Redis 7.0 or later.** The provider refuses an older server the first + time it talks to it. Tested with Redis 7.0, 7.4, 8.10 and Valkey 8. +- **`maxmemory-policy noeviction`.** Under memory pressure, an evicting + policy drops whole keys: a stream's records, its dedupe state, or a + staged Workflow publish. The provider logs a warning when it can read + another policy. With `noeviction`, a full Redis refuses writes instead, + and producers get an error they can retry. +- **Enough memory for retention.** A stream keeps records for `retention` + (7 days by default) after they were written. Size Redis for your write + rate times the retention. + +## Retention and cleanup + +The provider cleans up after itself, with no job to run: + +- Every write trims the log to `retention` and refreshes the log's expiry, + so a stream disappears `retention` after its last write. That also covers + Workflows that were terminated or timed out. +- Each log's metadata outlives the log by 30 days. That lets a reader tell + a cursor whose records expired (`StreamExpiredError`) from a stream that + never existed. +- A staged Workflow publish that is never made visible expires with + retention. +- Deleting a Workflow does not delete its streams. To remove them at once, + for example for a deletion request, call + `RedisStreams.delete_workflow_streams(namespace, workflow_id)`. It walks + the keyspace with `SCAN`, so run it rarely. + +## When a Workflow closes + +Redis cannot see a Workflow close, so the provider marks the stream closed +when someone notices: the Worker after a publishing Workflow's final +Workflow Task, a producer when it first writes, or a reader when its read +ends. After that, appends fail with `StreamClosedError`. + +Between the Workflow's final Workflow Task and that mark there is a closing +window in which appends are still accepted. It is usually short. If the +Worker stops before it marks the stream, the window lasts until a reader or +a new producer notices. Records written in the window stay readable. +Continue-as-New does not close a stream: the next run keeps publishing to +it. + +## A Worker that stops + +If a Worker stops after Temporal accepted a Workflow Task but before it +made that task's records visible, the records are not lost. They stay +staged in Redis. The next Worker that replays the Workflow makes them +visible, and so does any reader of the stream, after it checks the +Workflow's History. Each record becomes visible exactly once. + +## Access control + +Temporal does not authorize or meter record traffic. Records go straight +between your applications and your Redis, so access is Redis's job. + +Every key of a namespace's streams starts with +`:{:`, where `` is the provider's `key_prefix` +(default `temporal-streams`). Each part is percent-encoded, so a namespace +`my-ns` gives keys under `temporal-streams:{my-ns:`. A Redis ACL user for +one namespace's applications: + +``` +ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hincrby +hdel +rpush +lrange +exists +del +unlink +pexpire +time +info +config|get +ping +hello +client|setinfo +``` + +The conformance suite passes as a user with exactly these commands. The +delete helper also needs `+scan`. `+config|get` is only used to read +`maxmemory-policy`, and the provider tolerates a server that refuses it. + +What crosses the wire: + +- **Record bodies** go through your data converter's payload codec before + they are stored, so an encrypting codec keeps them encrypted in Redis. +- **Workflow ids, topic names and producer ids** appear in key names and + record fields in clear text. Do not put secrets in them. +- **Transport.** Use TLS (`rediss://`) to encrypt traffic to Redis. + +## Redis Cluster + +All keys of one Workflow's streams share a hash tag, the Workflow's run +chain, so every script touches one slot. Different Workflows spread across +the cluster. From 2cd4fcc7f4b0cba071a2e7cad8f25489b876b2d4 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 19:06:32 -0700 Subject: [PATCH 2/7] Documented unencrypted content hashes and Redis Cluster support. The content hash and the dedupe digest are taken before the codec, so they let a reader of the keys compare values. Cluster support needs a cluster client and names what is checked and tested. --- .../contrib/streams/docs/redis-guarantees.md | 25 ++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/temporalio/contrib/streams/docs/redis-guarantees.md b/temporalio/contrib/streams/docs/redis-guarantees.md index d5520f01d8..4cfb68cdff 100644 --- a/temporalio/contrib/streams/docs/redis-guarantees.md +++ b/temporalio/contrib/streams/docs/redis-guarantees.md @@ -113,10 +113,29 @@ What crosses the wire: they are stored, so an encrypting codec keeps them encrypted in Redis. - **Workflow ids, topic names and producer ids** appear in key names and record fields in clear text. Do not put secrets in them. +- **Content hashes are not encrypted.** Each record carries the SHA-256 of + its body as the converter produced it, before the codec, under the + metadata key `temporal.io/content-hash`. Each producer's dedupe state in + the log's metadata holds a SHA-256 digest of its newest batch, also taken + before the codec. Neither reveals a body, but anyone who can read the + keys can tell when two records carry the same value, and can confirm a + guess of a value. If that matters for your data, add something unique, + such as a random field, to each value. - **Transport.** Use TLS (`rediss://`) to encrypt traffic to Redis. ## Redis Cluster -All keys of one Workflow's streams share a hash tag, the Workflow's run -chain, so every script touches one slot. Different Workflows spread across -the cluster. +The provider supports Redis Cluster. Pass a `redis.asyncio.RedisCluster` +client; a URL string always makes a single-server client. + +All keys of one run chain's streams share a hash tag, so every script and +transaction touches one slot. Different Workflows, and different run +chains of one Workflow id, spread across the cluster. The version check +and the `maxmemory-policy` check ask every primary, and +`delete_workflow_streams` scans every primary. + +The conformance suite and the provider tests run against a Redis 7.4 +cluster with three primaries. The cluster client also sends `CLUSTER SLOTS` +and `COMMAND` to discover the cluster, so an ACL user for it needs +`+cluster|slots +command` as well. The ACL above was tested on a single +server only. From a2c62a2fc59056fbb224330b9a904d017a4d4ab3 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 9 Oct 2026 22:13:27 -0700 Subject: [PATCH 3/7] Allowed PTTL in the documented ACL and tested the provider under it. The stage script reads the pending hash's expiry, so every Workflow publish failed under the documented user. The conformance suite and a publish test now run as that user, built from the page's own line. --- .../contrib/streams/docs/redis-guarantees.md | 10 ++-- tests/contrib/streams/_redis_acl.py | 60 +++++++++++++++++++ tests/contrib/streams/test_redis_acl.py | 59 ++++++++++++++++++ .../streams/test_streams_conformance.py | 30 ++++++++-- 4 files changed, 151 insertions(+), 8 deletions(-) create mode 100644 tests/contrib/streams/_redis_acl.py create mode 100644 tests/contrib/streams/test_redis_acl.py diff --git a/temporalio/contrib/streams/docs/redis-guarantees.md b/temporalio/contrib/streams/docs/redis-guarantees.md index 4cfb68cdff..49cfa7e6fe 100644 --- a/temporalio/contrib/streams/docs/redis-guarantees.md +++ b/temporalio/contrib/streams/docs/redis-guarantees.md @@ -100,12 +100,14 @@ Every key of a namespace's streams starts with one namespace's applications: ``` -ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hincrby +hdel +rpush +lrange +exists +del +unlink +pexpire +time +info +config|get +ping +hello +client|setinfo +ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hincrby +hdel +rpush +lrange +exists +del +unlink +pexpire +pttl +time +info +config|get +ping +hello +client|setinfo ``` -The conformance suite passes as a user with exactly these commands. The -delete helper also needs `+scan`. `+config|get` is only used to read -`maxmemory-policy`, and the provider tolerates a server that refuses it. +The conformance suite and a Workflow publish test run as a user with +exactly these rules, so a provider change that needs another command fails +the tests. The delete helper also needs `+scan`. `+config|get` is only used +to read `maxmemory-policy`, and the provider tolerates a server that refuses +it. What crosses the wire: diff --git a/tests/contrib/streams/_redis_acl.py b/tests/contrib/streams/_redis_acl.py new file mode 100644 index 0000000000..51c52c5626 --- /dev/null +++ b/tests/contrib/streams/_redis_acl.py @@ -0,0 +1,60 @@ +"""A Redis ACL user with exactly the rules the guarantees page documents.""" + +from __future__ import annotations + +import re +import uuid +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from pathlib import Path +from urllib.parse import quote, urlsplit, urlunsplit + +import redis.asyncio + +GUARANTEES = ( + Path(__file__).parents[3] + / "temporalio" + / "contrib" + / "streams" + / "docs" + / "redis-guarantees.md" +) + +_DOC_KEYS = "~temporal-streams:{my-ns:*" + + +def documented_rules() -> list[str]: + """The rules of the page's ``ACL SETUSER`` line, after the user name.""" + lines = re.findall(r"^ACL SETUSER (\S+) (.+)$", GUARANTEES.read_text(), re.M) + assert len(lines) == 1, "the guarantees page should give one ACL SETUSER line" + return lines[0][1].split() + + +@asynccontextmanager +async def documented_user( + url: str, namespace: str, key_prefix: str, *extra: str +) -> AsyncIterator[str]: + """Create the documented user for one prefix and namespace; yield its URL. + + The page's key pattern and password are swapped for this test's own, and + nothing else changes. ``extra`` adds rules the page names separately. + """ + rules = documented_rules() + assert _DOC_KEYS in rules and ">secret" in rules + name = f"streams-acl-{uuid.uuid4().hex}" + password = uuid.uuid4().hex + # Key names percent-encode each part, as the provider writes them. + keys = f"~{quote(key_prefix, safe='')}:{{{quote(namespace, safe='')}:*" + rules = [ + keys if r == _DOC_KEYS else f">{password}" if r == ">secret" else r + for r in rules + ] + admin = redis.asyncio.Redis.from_url(url) + await admin.execute_command("ACL", "SETUSER", name, *rules, *extra) + parts = urlsplit(url) + netloc = f"{name}:{password}@{parts.hostname}:{parts.port or 6379}" + try: + yield urlunsplit(parts._replace(netloc=netloc)) + finally: + await admin.execute_command("ACL", "DELUSER", name) + await admin.aclose() diff --git a/tests/contrib/streams/test_redis_acl.py b/tests/contrib/streams/test_redis_acl.py new file mode 100644 index 0000000000..2dcd3eaf2d --- /dev/null +++ b/tests/contrib/streams/test_redis_acl.py @@ -0,0 +1,59 @@ +"""The Redis provider works as a user with exactly the documented ACL.""" + +from __future__ import annotations + +import os +import uuid +from datetime import timedelta + +import pytest + +from temporalio import workflow +from temporalio.client import Client +from temporalio.contrib.streams import StreamRef, topic, workflow_writer +from temporalio.contrib.streams.redis import RedisStreams +from tests.contrib.streams._redis_acl import documented_user +from tests.helpers import new_worker + +pytestmark = pytest.mark.skipif( + not os.environ.get("STREAMS_REDIS_URL"), + reason="set STREAMS_REDIS_URL to run the Redis ACL test", +) + +EVENTS = topic("events", dict) + + +@workflow.defn +class PublishTwice: + @workflow.run + async def run(self) -> None: + workflow_writer(EVENTS).publish({"n": 1}) + await workflow.sleep(timedelta(milliseconds=10)) + workflow_writer(EVENTS).publish({"n": 2}) + + +async def test_publish_read_and_delete_as_the_documented_user(client: Client): + prefix = f"test-acl-{uuid.uuid4().hex}" + url = os.environ["STREAMS_REDIS_URL"] + # The page says the delete helper also needs SCAN. + async with documented_user(url, client.namespace, prefix, "+scan") as acl_url: + provider = RedisStreams(acl_url, key_prefix=prefix) + config = client.config() + config["plugins"] = [provider] + streams_client = Client(**config) + workflow_id = f"redis-acl-{uuid.uuid4().hex}" + async with new_worker(streams_client, PublishTwice) as worker: + handle = await streams_client.start_workflow( + PublishTwice.run, id=workflow_id, task_queue=worker.task_queue + ) + stream = provider.get_stream_handle( + client, StreamRef.for_workflow(workflow_id) + ) + producer = stream.producer(topic=EVENTS, producer_id="p", attempt=1) + await producer.append({"n": "outside"}) + await handle.result() + values = [r.value async for r in stream.read(topic=EVENTS)] + assert {"n": 1} in values and {"n": 2} in values + assert {"n": "outside"} in values + assert await provider.delete_workflow_streams(client.namespace, workflow_id) > 0 + await provider.close() diff --git a/tests/contrib/streams/test_streams_conformance.py b/tests/contrib/streams/test_streams_conformance.py index 2392b82e2a..c586acc0c4 100644 --- a/tests/contrib/streams/test_streams_conformance.py +++ b/tests/contrib/streams/test_streams_conformance.py @@ -129,10 +129,15 @@ async def truncate(workflow_id: str, name: str, keep: int) -> None: @asynccontextmanager async def _redis_case( - client: Client, cluster: bool = False + client: Client, + cluster: bool = False, + *, + url: str | None = None, + prefix: str | None = None, + name: str | None = None, ) -> AsyncIterator[ProviderCase]: variable = "STREAMS_REDIS_CLUSTER_URL" if cluster else "STREAMS_REDIS_URL" - url = os.environ.get(variable) + url = url or os.environ.get(variable) if not url: pytest.skip(f"set {variable} to run these Redis provider cases") from redis.asyncio.cluster import RedisCluster @@ -140,7 +145,7 @@ async def _redis_case( from temporalio.contrib.streams.redis import RedisStreams # A prefix per case keeps cases apart in one Redis. - prefix = f"conformance-{uuid.uuid4().hex}" + prefix = prefix or f"conformance-{uuid.uuid4().hex}" cluster_client = RedisCluster.from_url(url) if cluster else None provider = RedisStreams(cluster_client or url, key_prefix=prefix) owners: list[WorkflowHandle] = [] @@ -154,7 +159,7 @@ async def host(workflow_id: str) -> str: assert handle.result_run_id is not None return handle.result_run_id - name = "redis-cluster" if cluster else "redis" + name = name or ("redis-cluster" if cluster else "redis") case = ProviderCase(name, provider, client, host=host) async def truncate(workflow_id: str, name: str, keep: int) -> None: @@ -180,9 +185,26 @@ async def truncate(workflow_id: str, name: str, keep: int) -> None: await cluster_client.aclose() +@asynccontextmanager +async def _redis_acl_case(client: Client) -> AsyncIterator[ProviderCase]: + # The guarantees page says this suite passes as its documented ACL user. + url = os.environ.get("STREAMS_REDIS_URL") + if not url: + pytest.skip("set STREAMS_REDIS_URL to run these Redis provider cases") + from tests.contrib.streams._redis_acl import documented_user + + prefix = f"conformance-{uuid.uuid4().hex}" + async with documented_user(url, client.namespace, prefix) as acl_url: + async with _redis_case( + client, url=acl_url, prefix=prefix, name="redis-acl" + ) as case: + yield case + + SETUPS: dict[str, Callable[[Client], Any]] = { "memory": _memory_case, "redis": _redis_case, + "redis-acl": _redis_acl_case, "redis-cluster": functools.partial(_redis_case, cluster=True), } From f586277cec0a1add43aefa4ccb822305f4bda5f0 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 14:51:24 -0700 Subject: [PATCH 4/7] Brought the guarantees page in line with the provider. It now covers forward-only upgrades, where committed output can be lost, what readers need, store dependence and the closing bound. The ACL allows HSCAN, which the session sweep uses. --- .../contrib/streams/docs/redis-guarantees.md | 75 +++++++++++++++---- 1 file changed, 60 insertions(+), 15 deletions(-) diff --git a/temporalio/contrib/streams/docs/redis-guarantees.md b/temporalio/contrib/streams/docs/redis-guarantees.md index 49cfa7e6fe..50a56b7533 100644 --- a/temporalio/contrib/streams/docs/redis-guarantees.md +++ b/temporalio/contrib/streams/docs/redis-guarantees.md @@ -16,7 +16,21 @@ writes nothing. A Workflow's own publish is staged first and made visible only after Temporal accepts the Workflow Task that published it. A task that fails -publishes nothing. +publishes nothing. While a task publishes, its progress depends on Redis: +if the stage call fails or outlasts the Workflow Task timeout, the task +fails and retries until Redis answers. Workflows that don't publish in that +task aren't affected. + +A body larger than the data converter's external storage threshold is +stored as a claim, and the object behind it lives in your external store. +Stream retention doesn't expire that object, and the external store's own +lifecycle decides when it goes. If it goes first, a read of that record +raises `StreamRecordError`. + +Promoting a large publish runs one script on Redis's single thread, which +reads the whole stage and writes every record. Thousands of large records +in one Workflow Task hold up every other client of that shard for the +script's run time. How long an acknowledged record survives a failure depends on persistence: @@ -30,8 +44,12 @@ How long an acknowledged record survives a failure depends on persistence: Replicas copy writes asynchronously. A failover can lose writes the primary acknowledged but had not yet sent to the replica. The provider does not use `WAIT`. If a failover must not lose acknowledged records, use a store with -a durable write log, such as Amazon MemoryDB, or `appendfsync always` on a -primary without automatic failover. +a durable write log, or `appendfsync always` on a primary without automatic +failover. Amazon MemoryDB has a durable write log, but this release hasn't +run its tests against MemoryDB. After a failover that lost writes, a +reader's cursor can name an entry the new primary never had, and new +entries can get lower ids, so the reader skips them. Read again from +`latest()` after a known failover. Recommended for production: `appendonly yes` with `appendfsync everysec` or `always`, at least one replica, and the settings below. @@ -42,8 +60,11 @@ Recommended for production: `appendonly yes` with `appendfsync everysec` or time it talks to it. Tested with Redis 7.0, 7.4, 8.10 and Valkey 8. - **`maxmemory-policy noeviction`.** Under memory pressure, an evicting policy drops whole keys: a stream's records, its dedupe state, or a - staged Workflow publish. The provider logs a warning when it can read - another policy. With `noeviction`, a full Redis refuses writes instead, + staged Workflow publish. If it drops a log's metadata, a reader can no + longer tell that a cursor expired, and it reads on across the gap. The + provider logs a warning when it can read another policy. It checks the + policy, and the server version, once when it starts, so a later change + goes unnoticed. With `noeviction`, a full Redis refuses writes instead, and producers get an error they can retry. - **Enough memory for retention.** A stream keeps records for `retention` (7 days by default) after they were written. Size Redis for your write @@ -59,12 +80,18 @@ The provider cleans up after itself, with no job to run: - Each log's metadata outlives the log by 30 days. That lets a reader tell a cursor whose records expired (`StreamExpiredError`) from a stream that never existed. -- A staged Workflow publish that is never made visible expires with - retention. +- A staged Workflow publish that is never made visible expires `retention` + plus 30 days after it was staged. +- Each producer attempt keeps one dedupe field in the log's metadata. Once + its last write is older than `retention`, the records it guards are gone, + and writes to the topic drop it a few fields at a time. - Deleting a Workflow does not delete its streams. To remove them at once, for example for a deletion request, call `RedisStreams.delete_workflow_streams(namespace, workflow_id)`. It walks - the keyspace with `SCAN`, so run it rarely. + the keyspace with `SCAN`, so run it rarely. Run it only after the chain + closed and its producers stopped: it also removes the close flag and the + dedupe state, so a late append writes a fresh log, and a Worker that + hasn't promoted its last stage loses that output. ## When a Workflow closes @@ -75,24 +102,42 @@ ends. After that, appends fail with `StreamClosedError`. Between the Workflow's final Workflow Task and that mark there is a closing window in which appends are still accepted. It is usually short. If the -Worker stops before it marks the stream, the window lasts until a reader or -a new producer notices. Records written in the window stay readable. +Worker stops before it marks the stream, or the Workflow never published +itself, the window lasts until a reader or a producer notices. A producer +checks again at least once a minute, so a running producer notices within +a minute. Records written in the window stay readable. Continue-as-New does not close a stream: the next run keeps publishing to it. ## A Worker that stops If a Worker stops after Temporal accepted a Workflow Task but before it -made that task's records visible, the records are not lost. They stay -staged in Redis. The next Worker that replays the Workflow makes them -visible, and so does any reader of the stream, after it checks the -Workflow's History. Each record becomes visible exactly once. +made that task's records visible, the records stay staged in Redis. The +next Worker that replays the Workflow makes them visible, and so does any +reader of the stream, after it checks the Workflow's History. Each record +becomes visible once. Two cases lose such output. A stage nobody promotes +within `retention` plus 30 days expires, and the promotion that finds it +gone logs a warning that names its token. A stage whose run History no +longer holds is never promoted, and it waits for that expiry. + +A publishing Workflow can only move forward. Once it has published, only a +Worker with this release and the stream provider plugin can replay it, so +keep the plugin registered while such Workflows run, and don't roll a +fleet back to a release without them. Their next Workflow Task would fail +as nondeterministic until the rollback is undone. ## Access control Temporal does not authorize or meter record traffic. Records go straight between your applications and your Redis, so access is Redis's job. +Readers write too. A read marks the chain closed when it ends, and it +promotes or aborts the stages a stopped Worker left. So a reader needs the +same Redis rules as a writer, and the Temporal permissions to describe the +Workflow and read its History. A read holds one Redis connection while it +waits, so size the client's `max_connections` to the streams a process +follows. + Every key of a namespace's streams starts with `:{:`, where `` is the provider's `key_prefix` (default `temporal-streams`). Each part is percent-encoded, so a namespace @@ -100,7 +145,7 @@ Every key of a namespace's streams starts with one namespace's applications: ``` -ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hincrby +hdel +rpush +lrange +exists +del +unlink +pexpire +pttl +time +info +config|get +ping +hello +client|setinfo +ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hincrby +hdel +hscan +rpush +lrange +exists +del +unlink +pexpire +pttl +time +info +config|get +ping +hello +client|setinfo ``` The conformance suite and a Workflow publish test run as a user with From 5b97353fac1094067551dc3a6b5c10e11006a066 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sat, 10 Oct 2026 22:09:44 -0700 Subject: [PATCH 5/7] Pointed the Redis guarantees at Core's store. Core runs the Redis store for every SDK now, with its own connection pool and cluster settings. --- temporalio/contrib/streams/README.md | 2 +- .../contrib/streams/docs/redis-guarantees.md | 59 +++++++++--------- tests/contrib/streams/_redis_acl.py | 60 ------------------- tests/contrib/streams/test_redis_acl.py | 59 ------------------ 4 files changed, 30 insertions(+), 150 deletions(-) delete mode 100644 tests/contrib/streams/_redis_acl.py delete mode 100644 tests/contrib/streams/test_redis_acl.py diff --git a/temporalio/contrib/streams/README.md b/temporalio/contrib/streams/README.md index 479c578c62..bacf3b295f 100644 --- a/temporalio/contrib/streams/README.md +++ b/temporalio/contrib/streams/README.md @@ -115,4 +115,4 @@ Both samples run in full in `tests/contrib/streams/samples/`. ## Further reading -- [Redis provider: guarantees and setup](docs/redis-guarantees.md) +- [Redis store: guarantees and setup](docs/redis-guarantees.md) diff --git a/temporalio/contrib/streams/docs/redis-guarantees.md b/temporalio/contrib/streams/docs/redis-guarantees.md index 50a56b7533..c77ff708ba 100644 --- a/temporalio/contrib/streams/docs/redis-guarantees.md +++ b/temporalio/contrib/streams/docs/redis-guarantees.md @@ -1,11 +1,12 @@ -# Redis provider: guarantees and setup +# Redis store: guarantees and setup > This package is experimental and may change in future versions. Stream records live in your Redis. Temporal never sees them, so what a -stream guarantees is what your Redis guarantees. This page says what the -provider promises, which Redis settings those promises depend on, and who -is responsible for what. +stream guarantees is what your Redis guarantees. Core runs the store for +every SDK, with the scripts in `crates/streams/lua/` of the Core repository. +This page says what the store promises, which Redis settings those promises +depend on, and who is responsible for what. ## What an acknowledged append means @@ -42,7 +43,7 @@ How long an acknowledged record survives a failure depends on persistence: | No persistence | Nothing, after a restart. | Replicas copy writes asynchronously. A failover can lose writes the primary -acknowledged but had not yet sent to the replica. The provider does not use +acknowledged but had not yet sent to the replica. The store does not use `WAIT`. If a failover must not lose acknowledged records, use a store with a durable write log, or `appendfsync always` on a primary without automatic failover. Amazon MemoryDB has a durable write log, but this release hasn't @@ -54,17 +55,17 @@ entries can get lower ids, so the reader skips them. Read again from Recommended for production: `appendonly yes` with `appendfsync everysec` or `always`, at least one replica, and the settings below. -## Settings the provider needs +## Settings the store needs -- **Redis 7.0 or later.** The provider refuses an older server the first - time it talks to it. Tested with Redis 7.0, 7.4, 8.10 and Valkey 8. +- **Redis 7.0 or later.** Core refuses an older server when the client + connects the store. Core's tests run against Redis 7. - **`maxmemory-policy noeviction`.** Under memory pressure, an evicting policy drops whole keys: a stream's records, its dedupe state, or a staged Workflow publish. If it drops a log's metadata, a reader can no - longer tell that a cursor expired, and it reads on across the gap. The - provider logs a warning when it can read another policy. It checks the - policy, and the server version, once when it starts, so a later change - goes unnoticed. With `noeviction`, a full Redis refuses writes instead, + longer tell that a cursor expired, and it reads on across the gap. Core + logs a warning when it can read another policy. It checks the policy, + and the server version, once when it connects, so a later change goes + unnoticed. With `noeviction`, a full Redis refuses writes instead, and producers get an error they can retry. - **Enough memory for retention.** A stream keeps records for `retention` (7 days by default) after they were written. Size Redis for your write @@ -72,7 +73,7 @@ Recommended for production: `appendonly yes` with `appendfsync everysec` or ## Retention and cleanup -The provider cleans up after itself, with no job to run: +The store cleans up after itself, with no job to run: - Every write trims the log to `retention` and refreshes the log's expiry, so a stream disappears `retention` after its last write. That also covers @@ -95,7 +96,7 @@ The provider cleans up after itself, with no job to run: ## When a Workflow closes -Redis cannot see a Workflow close, so the provider marks the stream closed +Redis cannot see a Workflow close, so Core marks the stream closed when someone notices: the Worker after a publishing Workflow's final Workflow Task, a producer when it first writes, or a reader when its read ends. After that, appends fail with `StreamClosedError`. @@ -121,9 +122,9 @@ gone logs a warning that names its token. A stage whose run History no longer holds is never promoted, and it waits for that expiry. A publishing Workflow can only move forward. Once it has published, only a -Worker with this release and the stream provider plugin can replay it, so -keep the plugin registered while such Workflows run, and don't roll a -fleet back to a release without them. Their next Workflow Task would fail +Worker with this release can replay it, so don't roll a fleet back to a +release without streams while such Workflows run. Replaying needs no store, +but a Worker that runs such a Workflow on needs the store on its client. Their next Workflow Task would fail as nondeterministic until the rollback is undone. ## Access control @@ -135,11 +136,11 @@ Readers write too. A read marks the chain closed when it ends, and it promotes or aborts the stages a stopped Worker left. So a reader needs the same Redis rules as a writer, and the Temporal permissions to describe the Workflow and read its History. A read holds one Redis connection while it -waits, so size the client's `max_connections` to the streams a process -follows. +waits, from a pool Core keeps per node, so set `blocking_reads_per_node` on +`RedisStreams` to the streams a process follows at once. Every key of a namespace's streams starts with -`:{:`, where `` is the provider's `key_prefix` +`:{:`, where `` is the store's `key_prefix` (default `temporal-streams`). Each part is percent-encoded, so a namespace `my-ns` gives keys under `temporal-streams:{my-ns:`. A Redis ACL user for one namespace's applications: @@ -149,10 +150,9 @@ ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetcha ``` The conformance suite and a Workflow publish test run as a user with -exactly these rules, so a provider change that needs another command fails +exactly these rules, so a store change that needs another command fails the tests. The delete helper also needs `+scan`. `+config|get` is only used -to read `maxmemory-policy`, and the provider tolerates a server that refuses -it. +to read `maxmemory-policy`, and the store goes on when a server refuses it. What crosses the wire: @@ -172,8 +172,8 @@ What crosses the wire: ## Redis Cluster -The provider supports Redis Cluster. Pass a `redis.asyncio.RedisCluster` -client; a URL string always makes a single-server client. +The store supports Redis Cluster. Pass the seed nodes and +`cluster=True` to `RedisStreams`. All keys of one run chain's streams share a hash tag, so every script and transaction touches one slot. Different Workflows, and different run @@ -181,8 +181,7 @@ chains of one Workflow id, spread across the cluster. The version check and the `maxmemory-policy` check ask every primary, and `delete_workflow_streams` scans every primary. -The conformance suite and the provider tests run against a Redis 7.4 -cluster with three primaries. The cluster client also sends `CLUSTER SLOTS` -and `COMMAND` to discover the cluster, so an ACL user for it needs -`+cluster|slots +command` as well. The ACL above was tested on a single -server only. +Core's conformance suite runs against a Redis 7 cluster too. The cluster +client also sends `CLUSTER SLOTS` and `COMMAND` to discover the cluster, so +an ACL user for it needs `+cluster|slots +command` as well. The ACL above +was tested on a single server only. diff --git a/tests/contrib/streams/_redis_acl.py b/tests/contrib/streams/_redis_acl.py deleted file mode 100644 index 51c52c5626..0000000000 --- a/tests/contrib/streams/_redis_acl.py +++ /dev/null @@ -1,60 +0,0 @@ -"""A Redis ACL user with exactly the rules the guarantees page documents.""" - -from __future__ import annotations - -import re -import uuid -from collections.abc import AsyncIterator -from contextlib import asynccontextmanager -from pathlib import Path -from urllib.parse import quote, urlsplit, urlunsplit - -import redis.asyncio - -GUARANTEES = ( - Path(__file__).parents[3] - / "temporalio" - / "contrib" - / "streams" - / "docs" - / "redis-guarantees.md" -) - -_DOC_KEYS = "~temporal-streams:{my-ns:*" - - -def documented_rules() -> list[str]: - """The rules of the page's ``ACL SETUSER`` line, after the user name.""" - lines = re.findall(r"^ACL SETUSER (\S+) (.+)$", GUARANTEES.read_text(), re.M) - assert len(lines) == 1, "the guarantees page should give one ACL SETUSER line" - return lines[0][1].split() - - -@asynccontextmanager -async def documented_user( - url: str, namespace: str, key_prefix: str, *extra: str -) -> AsyncIterator[str]: - """Create the documented user for one prefix and namespace; yield its URL. - - The page's key pattern and password are swapped for this test's own, and - nothing else changes. ``extra`` adds rules the page names separately. - """ - rules = documented_rules() - assert _DOC_KEYS in rules and ">secret" in rules - name = f"streams-acl-{uuid.uuid4().hex}" - password = uuid.uuid4().hex - # Key names percent-encode each part, as the provider writes them. - keys = f"~{quote(key_prefix, safe='')}:{{{quote(namespace, safe='')}:*" - rules = [ - keys if r == _DOC_KEYS else f">{password}" if r == ">secret" else r - for r in rules - ] - admin = redis.asyncio.Redis.from_url(url) - await admin.execute_command("ACL", "SETUSER", name, *rules, *extra) - parts = urlsplit(url) - netloc = f"{name}:{password}@{parts.hostname}:{parts.port or 6379}" - try: - yield urlunsplit(parts._replace(netloc=netloc)) - finally: - await admin.execute_command("ACL", "DELUSER", name) - await admin.aclose() diff --git a/tests/contrib/streams/test_redis_acl.py b/tests/contrib/streams/test_redis_acl.py deleted file mode 100644 index 2dcd3eaf2d..0000000000 --- a/tests/contrib/streams/test_redis_acl.py +++ /dev/null @@ -1,59 +0,0 @@ -"""The Redis provider works as a user with exactly the documented ACL.""" - -from __future__ import annotations - -import os -import uuid -from datetime import timedelta - -import pytest - -from temporalio import workflow -from temporalio.client import Client -from temporalio.contrib.streams import StreamRef, topic, workflow_writer -from temporalio.contrib.streams.redis import RedisStreams -from tests.contrib.streams._redis_acl import documented_user -from tests.helpers import new_worker - -pytestmark = pytest.mark.skipif( - not os.environ.get("STREAMS_REDIS_URL"), - reason="set STREAMS_REDIS_URL to run the Redis ACL test", -) - -EVENTS = topic("events", dict) - - -@workflow.defn -class PublishTwice: - @workflow.run - async def run(self) -> None: - workflow_writer(EVENTS).publish({"n": 1}) - await workflow.sleep(timedelta(milliseconds=10)) - workflow_writer(EVENTS).publish({"n": 2}) - - -async def test_publish_read_and_delete_as_the_documented_user(client: Client): - prefix = f"test-acl-{uuid.uuid4().hex}" - url = os.environ["STREAMS_REDIS_URL"] - # The page says the delete helper also needs SCAN. - async with documented_user(url, client.namespace, prefix, "+scan") as acl_url: - provider = RedisStreams(acl_url, key_prefix=prefix) - config = client.config() - config["plugins"] = [provider] - streams_client = Client(**config) - workflow_id = f"redis-acl-{uuid.uuid4().hex}" - async with new_worker(streams_client, PublishTwice) as worker: - handle = await streams_client.start_workflow( - PublishTwice.run, id=workflow_id, task_queue=worker.task_queue - ) - stream = provider.get_stream_handle( - client, StreamRef.for_workflow(workflow_id) - ) - producer = stream.producer(topic=EVENTS, producer_id="p", attempt=1) - await producer.append({"n": "outside"}) - await handle.result() - values = [r.value async for r in stream.read(topic=EVENTS)] - assert {"n": 1} in values and {"n": 2} in values - assert {"n": "outside"} in values - assert await provider.delete_workflow_streams(client.namespace, workflow_id) > 0 - await provider.close() From 87935d8ccb3e7dc9d56c11f661f12fae7202d064 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 00:45:43 -0700 Subject: [PATCH 6/7] Split the rollback note and tested the ACL line against Core's. The rollback paragraph ran two rules together, and the page's ACL line is a copy of the one Core's ACL lane uses, so a test now keeps them equal. The page also says what a full read pool does and which TLS works. --- .../contrib/streams/docs/redis-guarantees.md | 21 +++++++++++++----- tests/contrib/streams/test_redis_docs.py | 22 +++++++++++++++++++ 2 files changed, 37 insertions(+), 6 deletions(-) create mode 100644 tests/contrib/streams/test_redis_docs.py diff --git a/temporalio/contrib/streams/docs/redis-guarantees.md b/temporalio/contrib/streams/docs/redis-guarantees.md index c77ff708ba..0c17b62f10 100644 --- a/temporalio/contrib/streams/docs/redis-guarantees.md +++ b/temporalio/contrib/streams/docs/redis-guarantees.md @@ -122,10 +122,13 @@ gone logs a warning that names its token. A stage whose run History no longer holds is never promoted, and it waits for that expiry. A publishing Workflow can only move forward. Once it has published, only a -Worker with this release can replay it, so don't roll a fleet back to a -release without streams while such Workflows run. Replaying needs no store, -but a Worker that runs such a Workflow on needs the store on its client. Their next Workflow Task would fail -as nondeterministic until the rollback is undone. +Worker with this release can replay it. A Worker rolled back to a release +without streams fails the Workflow's next task as nondeterministic, until the +rollback is undone, so don't roll a fleet back while such Workflows run. + +Replaying a publishing Workflow needs no store, since nothing is stored +again. A Worker that runs such a Workflow on, and publishes live, needs the +store on its client. ## Access control @@ -137,7 +140,10 @@ promotes or aborts the stages a stopped Worker left. So a reader needs the same Redis rules as a writer, and the Temporal permissions to describe the Workflow and read its History. A read holds one Redis connection while it waits, from a pool Core keeps per node, so set `blocking_reads_per_node` on -`RedisStreams` to the streams a process follows at once. +`RedisStreams` to the streams a process follows at once. A read that finds +the pool full waits for a connection until its own poll ends, then answers +empty, so a process that follows more streams than the pool holds sees +records late rather than failing. Every key of a namespace's streams starts with `:{:`, where `` is the store's `key_prefix` @@ -168,7 +174,10 @@ What crosses the wire: keys can tell when two records carry the same value, and can confirm a guess of a value. If that matters for your data, add something unique, such as a random field, to each value. -- **Transport.** Use TLS (`rediss://`) to encrypt traffic to Redis. +- **Transport.** Use TLS (`rediss://`) to encrypt traffic to Redis. In this + release the store checks the server against the system's root + certificates only. A private CA and client certificates are not supported + yet. ## Redis Cluster diff --git a/tests/contrib/streams/test_redis_docs.py b/tests/contrib/streams/test_redis_docs.py new file mode 100644 index 0000000000..96e3b562db --- /dev/null +++ b/tests/contrib/streams/test_redis_docs.py @@ -0,0 +1,22 @@ +"""The Redis docs this package ships stay in step with Core's store.""" + +from __future__ import annotations + +import re +from pathlib import Path + +_ROOT = Path(__file__).resolve().parents[3] +_PYTHON_DOC = _ROOT / "temporalio/contrib/streams/docs/redis-guarantees.md" +# The line Core's ACL test lane builds its Redis user from. +_CORE_DOC = _ROOT / "temporalio/bridge/sdk-core/crates/streams/docs/redis.md" + + +def acl_line(doc: Path) -> str: + (line,) = re.findall(r"^ACL SETUSER .*$", doc.read_text(), re.MULTILINE) + return line.strip() + + +def test_the_documented_acl_line_is_cores(): + # A user built from an older copy of the line fails appends with NOPERM + # once Core's store needs another command. + assert acl_line(_PYTHON_DOC) == acl_line(_CORE_DOC) From 457b055cf545ebef31c87066465c0a72a4501648 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 11 Oct 2026 01:41:14 -0700 Subject: [PATCH 7/7] Took Core's ACL line with cluster slots, for a user on every node. Core's documented user now runs on a cluster too, which needs CLUSTER SLOTS. An ACL user lives on one node, and a delete during a reshard can miss keys. --- .../contrib/streams/docs/redis-guarantees.md | 21 +++++++++++-------- 1 file changed, 12 insertions(+), 9 deletions(-) diff --git a/temporalio/contrib/streams/docs/redis-guarantees.md b/temporalio/contrib/streams/docs/redis-guarantees.md index 0c17b62f10..b16c73958e 100644 --- a/temporalio/contrib/streams/docs/redis-guarantees.md +++ b/temporalio/contrib/streams/docs/redis-guarantees.md @@ -152,13 +152,16 @@ Every key of a namespace's streams starts with one namespace's applications: ``` -ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hincrby +hdel +hscan +rpush +lrange +exists +del +unlink +pexpire +pttl +time +info +config|get +ping +hello +client|setinfo +ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hincrby +hdel +hscan +rpush +lrange +exists +del +unlink +pexpire +pttl +time +info +config|get +ping +hello +client|setinfo +cluster|slots ``` The conformance suite and a Workflow publish test run as a user with -exactly these rules, so a store change that needs another command fails -the tests. The delete helper also needs `+scan`. `+config|get` is only used -to read `maxmemory-policy`, and the store goes on when a server refuses it. +exactly these rules, on a single server and on a cluster, so a store change +that needs another command fails the tests. A cluster client reads the +topology with `CLUSTER SLOTS` first. An ACL user lives on one node, so on a +cluster you create the user on every node. The delete helper also needs +`+scan`. `+config|get` is only used to read `maxmemory-policy`, and the +store goes on when a server refuses it. What crosses the wire: @@ -188,9 +191,9 @@ All keys of one run chain's streams share a hash tag, so every script and transaction touches one slot. Different Workflows, and different run chains of one Workflow id, spread across the cluster. The version check and the `maxmemory-policy` check ask every primary, and -`delete_workflow_streams` scans every primary. +`delete_workflow_streams` scans every primary. A scan makes no promise about +keys that move to another node while it runs, so a delete during a reshard +can miss some. Run it again once the reshard is done. -Core's conformance suite runs against a Redis 7 cluster too. The cluster -client also sends `CLUSTER SLOTS` and `COMMAND` to discover the cluster, so -an ACL user for it needs `+cluster|slots +command` as well. The ACL above -was tested on a single server only. +Core's conformance suite runs against a Redis 7 cluster too, also as a +user with exactly the ACL above.