fix(watcher): cancel gRPC stream in stop() to prevent thread leak (#76)

Storing the active stream as self._stream and calling cancel() before
thread.join() ensures the background Subscribe call unblocks promptly.
Without this, the daemon thread blocks indefinitely inside the gRPC
iterator and only exits at process shutdown.

Closes #46

Co-authored-by: Claude <noreply@anthropic.com>

Author: zeevdr

sdk/src/opendecree/watcher.py

@@ -148,6 +148,7 @@ def __init__(self, stub: Any, pb2: Any, tenant_id: str, timeout: float) -> None:
         self._timeout = timeout
         self._fields: dict[str, WatchedField] = {}  # type: ignore[type-arg]
         self._thread: threading.Thread | None = None
+        self._stream: grpc.Future | None = None
         self._stop_event = threading.Event()
 
     def field(self, path: str, type_: type[T], *, default: T) -> WatchedField[T]:
@@ -184,6 +185,8 @@ def start(self) -> None:
     def stop(self) -> None:
         """Stop watching and clean up the background thread."""
         self._stop_event.set()
+        if self._stream is not None:
+            self._stream.cancel()
         if self._thread is not None:
             self._thread.join(timeout=5.0)
             self._thread = None
@@ -215,19 +218,21 @@ def _subscribe_loop(self) -> None:
 
         while not self._stop_event.is_set():
             try:
-                stream = self._stub.Subscribe(
+                self._stream = self._stub.Subscribe(
                     self._pb2.SubscribeRequest(
                         tenant_id=self._tenant_id,
                         field_paths=field_paths,
                     ),
                 )
                 backoff = _RECONNECT_INITIAL  # reset on successful connect
 
-                for response in stream:
+                for response in self._stream:
                     if self._stop_event.is_set():
                         return
                     self._process_change(response.change)
 
+                self._stream = None
+
             except grpc.RpcError as e:
                 if self._stop_event.is_set():
                     return

sdk/tests/test_watcher.py

@@ -264,3 +264,41 @@ def test_non_retryable_error_stops_loop(self):
 
         # Thread should have exited on its own due to non-retryable error.
         assert w._thread is None
+
+    def test_stop_cancels_stream_and_joins_thread(self):
+        """stop() cancels the gRPC stream so the background thread exits cleanly."""
+        import threading
+
+        w = self._make_watcher()
+        w.field("fee", float, default=0.0)
+
+        # A blocking iterator that only unblocks when cancel() is called.
+        cancelled = threading.Event()
+
+        class _BlockingIter:
+            def __iter__(self):
+                return self
+
+            def __next__(self):
+                # Block until cancelled.
+                cancelled.wait(timeout=10.0)
+                raise StopIteration
+
+            def cancel(self):
+                cancelled.set()
+
+        blocking_stream = _BlockingIter()
+        w._stub.Subscribe.return_value = blocking_stream
+
+        w.start()
+        time.sleep(0.1)  # let the thread reach the blocking iterator
+
+        thread_ref = w._thread
+        assert thread_ref is not None
+        assert thread_ref.is_alive()
+
+        w.stop()
+
+        # Thread must have joined within the timeout.
+        assert not thread_ref.is_alive()
+        assert w._thread is None