fix(watcher): apply backoff on clean stream close

When the server closed the Subscribe stream normally, the outer reconnect
loop re-issued Subscribe immediately with no delay, causing a tight
reconnect loop on a server FIN.

Ports the backoff branch from async_watcher.py: on a clean stream close,
the watcher sleeps backoff × jitter seconds (in 0.1 s intervals so the
stop event is honoured) and grows the backoff the same way error
reconnects do.

Closes #51

Author: zeevdr

sdk/src/opendecree/watcher.py

@@ -233,6 +233,16 @@ def _subscribe_loop(self) -> None:
 
                 self._stream = None
 
+                # Stream ended normally (server closed) — reconnect with backoff.
+                if not self._stop_event.is_set():
+                    jitter = random.uniform(0.5, 1.5)
+                    sleep_time = backoff * jitter
+                    logger.warning("Stream closed by server, reconnecting in %.1fs", sleep_time)
+                    deadline = time.monotonic() + sleep_time
+                    while time.monotonic() < deadline and not self._stop_event.is_set():
+                        time.sleep(0.1)
+                    backoff = min(backoff * _RECONNECT_MULTIPLIER, _RECONNECT_MAX)
+
             except grpc.RpcError as e:
                 if self._stop_event.is_set():
                     return

sdk/tests/test_watcher.py

@@ -265,6 +265,64 @@ 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_reconnects_after_clean_stream_close(self):
+        """Clean server-side stream close triggers a reconnect."""
+        import unittest.mock as mock
+
+        w = self._make_watcher()
+        w.field("fee", float, default=0.0)
+
+        call_count = 0
+
+        def _subscribe_side_effect(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                return iter([])  # clean close — server FIN
+            # Stop after the second call so the thread exits.
+            w._stop_event.set()
+            return iter([])
+
+        w._stub.Subscribe.side_effect = _subscribe_side_effect
+
+        with mock.patch("opendecree.watcher._RECONNECT_INITIAL", 0.2):
+            w.start()
+            time.sleep(1.0)
+            w.stop()
+
+        assert call_count >= 2
+
+    def test_clean_close_applies_backoff(self):
+        """Reconnect after a clean close is delayed, not immediate."""
+        import unittest.mock as mock
+
+        w = self._make_watcher()
+        w.field("fee", float, default=0.0)
+
+        call_count = 0
+        timestamps: list[float] = []
+
+        def _subscribe_side_effect(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            timestamps.append(time.monotonic())
+            if call_count == 1:
+                return iter([])  # clean close
+            w._stop_event.set()
+            return iter([])
+
+        w._stub.Subscribe.side_effect = _subscribe_side_effect
+
+        with mock.patch("opendecree.watcher._RECONNECT_INITIAL", 0.2):
+            w.start()
+            time.sleep(1.0)
+            w.stop()
+
+        assert len(timestamps) >= 2
+        gap = timestamps[1] - timestamps[0]
+        # Minimum gap is jitter_min (0.5) * RECONNECT_INITIAL (0.2) = 0.1s.
+        assert gap >= 0.05
+
     def test_stop_cancels_stream_and_joins_thread(self):
         """stop() cancels the gRPC stream so the background thread exits cleanly."""
         import threading