feat(streams): operator control server

sentry_streams/Cargo.toml

@@ -10,7 +10,6 @@ sentry_arroyo = { version = "2.40.0", features = ["ssl"] }
 chrono = "0.4.40"
 tracing = "0.1.40"
 tracing-subscriber = "0.3.20"
-ctrlc = "3.4.6"
 rdkafka = { version = "0.37.0", features = ["cmake-build", "tracing"] }
 anyhow = "1.0.98"
 reqwest = "0.12.15"

sentry_streams/sentry_streams/adapters/arroyo/adapter.py

@@ -33,7 +33,11 @@
     RouterStep,
     StreamSinkStep,
 )
-from sentry_streams.adapters.stream_adapter import PipelineConfig, StreamAdapter
+from sentry_streams.adapters.stream_adapter import (
+    PipelineConfig,
+    RuntimeState,
+    StreamAdapter,
+)
 from sentry_streams.config_types import (
     KafkaConsumerConfig,
     KafkaProducerConfig,
@@ -322,7 +326,7 @@ def create_processors(self) -> None:
             for source, consumer in self.__consumers.items()
         }
 
-    def run(self) -> None:
+    def _run(self) -> None:
         """
         Starts the pipeline
         """
@@ -332,14 +336,17 @@ def run(self) -> None:
         source = next(iter(self.__consumers))
 
         processor = self.__processors[source]
+        if self.status.state is RuntimeState.STOPPING:
+            processor.signal_shutdown()
+        else:
+            self._set_status(RuntimeState.CONSUMING)
+
         processor.run()
 
-    def shutdown(self) -> None:
+    def _shutdown(self) -> None:
         """
         Shutdown the arroyo processors allowing them to terminate the inflight
         work.
         """
-        assert len(self.__consumers) == 1, "Only one consumer is supported"
-        source = next(iter(self.__consumers))
-        processor = self.__processors[source]
-        processor.signal_shutdown()
+        for processor in self.__processors.values():
+            processor.signal_shutdown()

sentry_streams/sentry_streams/adapters/arroyo/rust_arroyo.py

@@ -27,7 +27,11 @@
 from sentry_streams.adapters.arroyo.routers import build_branches
 from sentry_streams.adapters.arroyo.routes import Route
 from sentry_streams.adapters.arroyo.steps_chain import TransformChains
-from sentry_streams.adapters.stream_adapter import PipelineConfig, StreamAdapter
+from sentry_streams.adapters.stream_adapter import (
+    PipelineConfig,
+    RuntimeState,
+    StreamAdapter,
+)
 from sentry_streams.config_types import (
     KafkaConsumerConfig,
     KafkaProducerConfig,
@@ -587,18 +591,20 @@ def routing_function(msg: Message[Any]) -> str:
         )
         return build_branches(stream, step.routing_table.values())
 
-    def run(self) -> None:
+    def _run(self) -> None:
         """
         Starts the pipeline
         """
         # TODO: Support multiple consumers
         assert len(self.__consumers) == 1, "Multiple consumers not supported yet"
         consumer = next(iter(self.__consumers.values()))
+        self._set_status(RuntimeState.CONSUMING)
         consumer.run()
 
-    def shutdown(self) -> None:
+    def _shutdown(self) -> None:
         """
         Shutdown the arroyo processors allowing them to terminate the inflight
         work.
         """
-        raise NotImplementedError
+        for consumer in self.__consumers.values():
+            consumer.shutdown()

sentry_streams/sentry_streams/adapters/stream_adapter.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+import threading
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from enum import StrEnum
 from typing import (
     Any,
     Callable,
@@ -41,13 +44,116 @@
 StreamSinkT = TypeVar("StreamSinkT")
 
 
+class RuntimeStateError(RuntimeError):
+    """
+    Raised when an operation conflicts with the runtime's lifecycle state.
+    """
+
+
+class RuntimeState(StrEnum):
+    IDLE = "idle"
+    STARTING = "starting"
+    CONSUMING = "consuming"
+    STOPPING = "stopping"
+    STOPPED = "stopped"
+    ERRORED = "errored"
+
+    @property
+    def is_terminal(self) -> bool:
+        return self in (RuntimeState.STOPPED, RuntimeState.ERRORED)
+
+    @property
+    def rank(self) -> int:
+        if self is RuntimeState.IDLE:
+            return 0
+        if self is RuntimeState.STARTING:
+            return 1
+        if self is RuntimeState.CONSUMING:
+            return 2
+        if self is RuntimeState.STOPPING:
+            return 3
+        return 4
+
+    def can_transition_to(self, state: RuntimeState) -> bool:
+        return self.rank < state.rank
+
+
+@dataclass(frozen=True)
+class RuntimeStatus:
+    state: RuntimeState
+    error: Exception | None = None
+
+    @property
+    def is_terminal(self) -> bool:
+        return self.state.is_terminal
+
+    def as_dict(self) -> dict[str, str | None]:
+        return {
+            "state": self.state.value,
+            "error": str(self.error) if self.error is not None else None,
+        }
+
+
 class StreamAdapter(ABC, Generic[StreamT, StreamSinkT]):
     """
     A generic adapter for mapping sentry_streams APIs
     and primitives to runtime-specific ones. This can
     be extended to specific runtimes.
     """
 
+    def __init__(self) -> None:
+        self.__status_lock = threading.Lock()
+        self.__status = RuntimeStatus(RuntimeState.IDLE)
+
+    @property
+    def status(self) -> RuntimeStatus:
+        with self.__status_lock:
+            return self.__status
+
+    def _set_status(self, state: RuntimeState, error: Exception | None = None) -> None:
+        with self.__status_lock:
+            if not self.__status.state.can_transition_to(state):
+                return
+            self.__status = RuntimeStatus(state, error)
+
+    def begin_start(self) -> RuntimeStatus:
+        with self.__status_lock:
+            state = self.__status.state
+            if state.is_terminal:
+                raise RuntimeStateError(f"cannot restart runtime that is {state}")
+            if state is not RuntimeState.IDLE:
+                raise RuntimeStateError(f"cannot start runtime while it is {state}")
+            self.__status = RuntimeStatus(RuntimeState.STARTING)
+            return self.__status
+
+    def run(self) -> None:
+        state = self.status.state
+        if state is RuntimeState.STOPPING:
+            self._set_status(RuntimeState.STOPPED)
+            return
+        if state is not RuntimeState.STARTING:
+            raise RuntimeStateError(f"cannot run runtime while it is {state}")
+
+        try:
+            self._run()
+        except Exception as exc:
+            self._set_status(RuntimeState.ERRORED, exc)
+            raise
+        else:
+            self._set_status(RuntimeState.STOPPED)
+
+    def shutdown(self) -> RuntimeStatus:
+        with self.__status_lock:
+            state = self.__status.state
+            if state is RuntimeState.STOPPING or state.is_terminal:
+                return self.__status
+            self.__status = RuntimeStatus(
+                RuntimeState.STOPPED if state is RuntimeState.IDLE else RuntimeState.STOPPING
+            )
+
+        self._shutdown()
+        return self.status
+
     @classmethod
     @abstractmethod
     def build(cls, config: PipelineConfig) -> Self:
@@ -145,14 +251,14 @@ def broadcast(
         raise NotImplementedError
 
     @abstractmethod
-    def run(self) -> None:
+    def _run(self) -> None:
         """
         Starts the pipeline
         """
         raise NotImplementedError
 
     @abstractmethod
-    def shutdown(self) -> None:
+    def _shutdown(self) -> None:
         """
         Cleanly shutdown the application.
         """

sentry_streams/sentry_streams/control.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import threading
+from concurrent.futures import Future, ThreadPoolExecutor
+from typing import Any
+
+from sentry_streams.adapters.stream_adapter import (
+    RuntimeStatus,
+    StreamAdapter,
+)
+
+
+class PipelineController:
+    """
+    Control panel for one streaming pipeline. Starts the adapter's blocking run loop
+    in a background thread and allows concurrent requests to start/stop the pipeline.
+
+    A pipeline is single-use: after it stops or fails, it cannot be restarted.
+    A replacement deployment must create a new controller and adapter.
+
+    Start and stop requests can arrive concurrently from HTTP handler and process
+    shutdown threads. The lock prevents state transition race conditions.
+    """
+
+    def __init__(self, runtime: StreamAdapter[Any, Any]) -> None:
+        self._runtime = runtime
+        self._lock = threading.Lock()
+        self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="pipeline-run")
+        self._run_future: Future[None] | None = None
+        self._finished = threading.Event()
+
+    @property
+    def snapshot(self) -> RuntimeStatus:
+        return self._runtime.status
+
+    def request_start(self) -> RuntimeStatus:
+        """
+        Ask the pipeline to start (non-blocking).
+        """
+        with self._lock:
+            status = self._runtime.begin_start()
+            if self._run_future is None:
+                self._run_future = self._executor.submit(self._runtime.run)
+                self._run_future.add_done_callback(self._run_finished)
+            return status
+
+    def request_stop(self) -> RuntimeStatus:
+        """
+        Ask the pipeline to stop (non-blocking).
+        """
+        with self._lock:
+            status = self._runtime.shutdown()
+            if self._run_future is None:
+                self._finished.set()
+                self._executor.shutdown(wait=False)
+
+            return status
+
+    def wait_until_finished(self, timeout: float | None = None) -> RuntimeStatus:
+        """
+        Wait until this controller is completely finished.
+        """
+        if not self._finished.wait(timeout):
+            return self._runtime.status
+
+        with self._lock:
+            run_future = self._run_future
+
+        if run_future is not None:
+            run_future.result()
+
+        return self._runtime.status
+
+    def wait_until_stopped(self, timeout: float | None = None) -> RuntimeStatus:
+        """
+        Wait for a started pipeline's background thread to exit.
+        """
+        self._finished.wait(timeout)
+        return self._runtime.status
+
+    def _run_finished(self, _future: Future[None]) -> None:
+        self._finished.set()
+        self._executor.shutdown(wait=False)

sentry_streams/sentry_streams/dummy/dummy_adapter.py

@@ -33,6 +33,7 @@ class DummyAdapter(StreamAdapter[DummyInput, DummyOutput]):
     """
 
     def __init__(self, _: PipelineConfig) -> None:
+        super().__init__()
         self.input_streams: list[str] = []
         self.branches: list[str] = []
 
@@ -94,8 +95,8 @@ def router(self, step: Router[RoutingFuncReturnType, Any], stream: Any) -> Any:
             ret[branch.root.name] = branch
         return ret
 
-    def run(self) -> None:
+    def _run(self) -> None:
         pass
 
-    def shutdown(self) -> None:
+    def _shutdown(self) -> None:
         pass