[Brook] Added stream topics and references.

temporalio/contrib/streams/__init__.py

@@ -20,6 +20,9 @@
    refused with :class:`StreamCursorError`, and one whose record
    retention dropped with :class:`StreamExpiredError`. A read with no cursor
    starts at :data:`BEGINNING` or :data:`END`.
+3. **Topics are defined once.** :func:`topic` defines a topic with the type
+   its records decode to, and every party shares that definition. A call
+   that names no topic addresses :data:`DEFAULT_TOPIC`.
 
 The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in
 :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body``
@@ -46,10 +49,13 @@
     StreamRecord,
     Supersession,
 )
+from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef
+from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic
 
 __all__ = [
     "BEGINNING",
     "Cursor",
+    "DEFAULT_TOPIC",
     "END",
     "RecordKind",
     "StreamClosedError",
@@ -58,11 +64,15 @@
     "StreamExpiredError",
     "StreamNotFoundError",
     "StreamOutcomeUnknownError",
+    "StreamOwnerKind",
     "StreamProducerError",
     "StreamRecord",
     "StreamRecordError",
+    "StreamRef",
     "StreamRefusedError",
     "StreamStorageError",
+    "StreamTopic",
     "StreamUnsupportedError",
     "Supersession",
+    "topic",
 ]

temporalio/contrib/streams/_ref.py

@@ -0,0 +1,91 @@
+"""A reference to one stream that crosses a process boundary as data.
+
+A handle is bound to a client and its store, so it cannot be a Workflow
+argument or an Activity result. A :class:`StreamRef` can: it names the owner
+and the topic, nothing more, and whoever receives it opens the stream on the
+store its own client carries. It carries no cursor, because a position
+belongs to a reader, and no store name.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+from dataclasses import dataclass
+from typing import Any, Literal
+
+from temporalio.contrib.streams._topic import (
+    DEFAULT_TOPIC,
+    StreamTopic,
+    resolve_topic,
+)
+
+__all__ = ["StreamOwnerKind", "StreamRef"]
+
+StreamOwnerKind = Literal["workflow"]
+"""The owner kinds this release opens. Only a Workflow owns a stream.
+
+:attr:`StreamRef.kind` is a plain string, so a ref that names an owner kind
+a later release adds still reaches the ref's own check when it is decoded,
+instead of failing the converter's type check.
+"""
+
+
+@dataclass(frozen=True)
+class StreamRef:
+    """One stream, named by its owner and its topic.
+
+    ``kind`` says what owns the stream; only ``"workflow"`` exists in this
+    release. A Workflow ref carries ``workflow_id`` and, when pinned to one
+    run, ``run_id``. The stream belongs to the run chain either way, and a
+    read returns the chain's records. ``run_id`` changes two things: which
+    chain the handle looks up, and that a read ends when that run closes,
+    Continue-as-New included. Without it a reader keeps reading across
+    Continue-as-New. ``topic`` is the topic a handle opened from the ref
+    addresses when a call names none.
+
+    The default data converter carries it as JSON, so it can be a Workflow
+    argument or an Activity result.
+    """
+
+    kind: str
+    workflow_id: str
+    run_id: str | None = None
+    topic: str = DEFAULT_TOPIC
+
+    def __post_init__(self) -> None:
+        """Refuse a ref this release cannot open.
+
+        Raises:
+            ValueError: ``kind`` is not ``"workflow"``, or ``workflow_id`` or
+                ``topic`` is empty.
+        """
+        if self.kind != "workflow":
+            raise ValueError(f"unknown StreamRef kind {self.kind!r}")
+        if not self.workflow_id:
+            raise ValueError("a Workflow StreamRef needs a workflow_id")
+        if not self.topic:
+            raise ValueError("a StreamRef needs a topic name")
+
+    @classmethod
+    def for_workflow(
+        cls,
+        workflow_id: str,
+        *,
+        run_id: str | None = None,
+        topic: str | StreamTopic[Any] | None = None,
+    ) -> StreamRef:
+        """A ref to ``topic`` of ``workflow_id``'s stream.
+
+        Without ``topic`` it names
+        :data:`temporalio.contrib.streams.DEFAULT_TOPIC`.
+        """
+        name, _ = resolve_topic(topic)
+        return cls("workflow", workflow_id, run_id=run_id, topic=name)
+
+    def with_topic(self, topic: str | StreamTopic[Any] | None) -> StreamRef:
+        """The same owner, naming ``topic`` instead.
+
+        ``None`` names :data:`temporalio.contrib.streams.DEFAULT_TOPIC`.
+        """
+        name, _ = resolve_topic(topic)
+        return dataclasses.replace(self, topic=name)

temporalio/contrib/streams/_topic.py

@@ -0,0 +1,108 @@
+"""Typed topic definitions.
+
+A topic is defined once, at module level, with the type its records decode
+to, and the Workflow, its Activities and the client code all refer to that
+one definition, the way they share a Signal or an Activity. A plain string
+names a topic too, for a name decided at runtime; then the decode hint
+travels as ``result_type=`` on each read.
+
+On the wire a topic is a string, and
+:attr:`temporalio.contrib.streams.StreamRecord.topic` is that string.
+"""
+
+from __future__ import annotations
+
+import unicodedata
+from dataclasses import dataclass
+from typing import Any, Generic, TypeVar, overload
+
+__all__ = ["DEFAULT_TOPIC", "StreamTopic", "resolve_topic", "topic"]
+
+T = TypeVar("T")
+
+DEFAULT_TOPIC = "output"
+"""The topic a call addresses when it names none.
+
+It is an ordinary name, so naming it explicitly addresses the same topic.
+"""
+
+
+@dataclass(frozen=True)
+class StreamTopic(Generic[T]):
+    """A topic of a stream, with the type its records decode to.
+
+    Made with :func:`topic`. Pass it wherever a call takes a topic, and the
+    record and value types follow from it.
+    """
+
+    name: str
+    result_type: type[T] | None = None
+    """The type records decode to, or ``None`` for the converter's default."""
+
+
+@overload
+def topic(name: str, result_type: type[T]) -> StreamTopic[T]: ...
+
+
+@overload
+def topic(name: str, result_type: None = None) -> StreamTopic[Any]: ...
+
+
+def topic(name: str, result_type: type | None = None) -> StreamTopic[Any]:
+    """Define a topic.
+
+    Args:
+        name: The topic's name, as it appears on every record.
+        result_type: The type records on this topic decode to. Without one,
+            the payload converter's default applies.
+
+    Raises:
+        ValueError: ``name`` is empty, longer than 256 UTF-8 bytes, or holds
+            a control character.
+    """
+    _check_name(name)
+    return StreamTopic(name, result_type)
+
+
+def resolve_topic(
+    topic: str | StreamTopic[Any] | None = None, result_type: type | None = None
+) -> tuple[str, type | None]:
+    """The name and decode hint a call means, from either form of topic.
+
+    ``None`` means :data:`DEFAULT_TOPIC`. A handle calls this once at the
+    top of each call that takes a topic, so a definition and a string reach
+    the store the same way.
+
+    Raises:
+        ValueError: A definition was given together with ``result_type``,
+            which would name two types for one topic, or the name is empty,
+            longer than 256 UTF-8 bytes, or holds a control character.
+    """
+    if isinstance(topic, StreamTopic):
+        if result_type is not None:
+            raise ValueError(
+                f"topic {topic.name!r} already carries its type; do not pass "
+                "result_type= with a definition"
+            )
+        name, result_type = topic.name, topic.result_type
+    elif topic is None:
+        name = DEFAULT_TOPIC
+    else:
+        name = topic
+    _check_name(name)
+    return name, result_type
+
+
+_MAX_NAME_BYTES = 256
+
+
+def _check_name(name: str) -> None:
+    # The same limit in every SDK: the name goes into store keys, cursor
+    # hashes and Core's manifest budget, and a store may join names with a
+    # control character.
+    if not name:
+        raise ValueError("topic name must not be empty")
+    if len(name.encode()) > _MAX_NAME_BYTES:
+        raise ValueError(f"topic name must be at most {_MAX_NAME_BYTES} UTF-8 bytes")
+    if any(unicodedata.category(c) == "Cc" for c in name):
+        raise ValueError(f"topic name {name!r} must not hold a control character")

tests/contrib/streams/test_public_surface.py

@@ -12,6 +12,7 @@
     "decode_body",
     "encode_bodies",
     "error_from_failure",
+    "resolve_topic",
 }
 
 

tests/contrib/streams/test_topics_refs.py

@@ -0,0 +1,88 @@
+"""Typed topics and the stream reference that crosses process boundaries."""
+
+from __future__ import annotations
+
+import pytest
+
+from temporalio.api.common.v1 import Payload
+from temporalio.contrib.streams import (
+    DEFAULT_TOPIC,
+    StreamRef,
+    StreamTopic,
+    topic,
+)
+from temporalio.contrib.streams._topic import (
+    resolve_topic,
+)
+from temporalio.converter import DataConverter
+
+OUT = topic("out", dict)
+
+
+def test_a_definition_carries_its_name_and_type():
+    assert OUT == StreamTopic("out", dict)
+    assert resolve_topic(OUT) == ("out", dict)
+    assert resolve_topic("free", int) == ("free", int)
+    assert resolve_topic(None) == (DEFAULT_TOPIC, None)
+
+
+def test_topic_mistakes_are_value_errors():
+    with pytest.raises(ValueError):
+        topic("")
+    with pytest.raises(ValueError):
+        resolve_topic("")
+    with pytest.raises(ValueError, match="already carries its type"):
+        resolve_topic(OUT, dict)
+
+
+def test_a_ref_names_a_workflow_stream():
+    ref = StreamRef.for_workflow("wf", topic=OUT)
+    assert ref == StreamRef("workflow", "wf", topic="out")
+    assert StreamRef.for_workflow("wf").topic == DEFAULT_TOPIC
+    assert ref.with_topic(None).topic == DEFAULT_TOPIC
+    assert ref.with_topic("other").workflow_id == "wf"
+
+
+def test_a_ref_round_trips_through_the_default_converter():
+    ref = StreamRef.for_workflow("wf", run_id="run", topic=OUT)
+    converter = DataConverter.default.payload_converter
+    payload = converter.to_payloads([ref])[0]
+    assert converter.from_payloads([payload], [StreamRef])[0] == ref
+
+
+def test_a_ref_refuses_what_it_cannot_name():
+    with pytest.raises(ValueError, match="unknown"):
+        StreamRef("nexus", "x")
+    with pytest.raises(ValueError):
+        StreamRef("workflow", "")
+    with pytest.raises(ValueError):
+        StreamRef("workflow", "wf", topic="")
+
+
+def test_a_ref_from_a_later_release_decodes_to_the_kind_check():
+    # The kind is a plain string, so the converter builds the ref and the ref
+    # itself refuses the kind, rather than the converter failing a type check.
+    converter = DataConverter.default.payload_converter
+    payload = Payload(
+        metadata={"encoding": b"json/plain"},
+        data=b'{"kind": "activity", "workflow_id": "wf", "topic": "out"}',
+    )
+    with pytest.raises(ValueError, match="'activity'"):
+        converter.from_payloads([payload], [StreamRef])
+
+
+@pytest.mark.parametrize(
+    "name",
+    ["x" * 257, "é" * 129, "a\x1fb", "line\nbreak", "\x00"],
+)
+def test_a_topic_name_that_is_too_long_or_has_a_control_character_is_refused(name: str):
+    # Every store puts the name in keys, cursors and Core's manifest budget.
+    with pytest.raises(ValueError):
+        topic(name)
+    with pytest.raises(ValueError):
+        resolve_topic(name)
+
+
+def test_a_topic_name_of_256_bytes_is_accepted():
+    assert topic("x" * 256).name == "x" * 256
+    assert resolve_topic("é" * 128)[0] == "é" * 128