From a124100acccc9e0c4c8bfc2d719a71ea2df9d0c6 Mon Sep 17 00:00:00 2001
From: Strycher <strycher@gmail.com>
Date: Wed, 24 Jun 2026 11:38:12 -0400
Subject: [PATCH] fix(#103): re-arm broker-pool GET timeout per frame
 (inactivity watchdog)

The broker-pool GET put a single fixed 6s deadline on the entire paginated dump. A large pool (6+ brokers) streams field-by-field over ~8s and blew the deadline, so the client falsely reported the pool as not finished even though the device sent it all (the END terminator arrives after the client gave up). Re-arm an inactivity Timer on each received frame: a steadily-streaming dump never times out; only a genuine stall (no frame for the timeout window) fails. Adds a TDD test - a slow-but-steady dump, frames under the timeout but total far over it, must complete.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 lib/services/observer_config_service.dart     | 22 ++++++++-
 .../observer_config_service_test.dart         | 48 +++++++++++++++++++
 2 files changed, 68 insertions(+), 2 deletions(-)

diff --git a/lib/services/observer_config_service.dart b/lib/services/observer_config_service.dart
index b17ad1d..7025481 100644
--- a/lib/services/observer_config_service.dart
+++ b/lib/services/observer_config_service.dart
@@ -203,23 +203,41 @@ class ObserverConfigService extends ChangeNotifier {
     return BrokerConfig.fromWireFields(slot, kv);
   }
 
-  /// Read the broker pool (paginated START → KV → END). Null on timeout.
+  /// Read the broker pool (paginated START → KV → END). Null on a stall or
+  /// error. The completion deadline is an INACTIVITY watchdog re-armed on every
+  /// frame — a large pool streams field-by-field and outlasts any single fixed
+  /// deadline, so only a genuine stall (no frame for [timeout]) fails (#103).
   Future<List<BrokerConfig>?> getBrokers() {
     return _serialized(() async {
       final decoder = BrokerListDecoder();
       final completer = Completer<List<BrokerConfig>>();
+      Timer? idle;
+      void armIdle() {
+        idle?.cancel();
+        idle = Timer(timeout, () {
+          if (!completer.isCompleted) {
+            completer.completeError(
+              TimeoutException('broker pool stalled', timeout),
+            );
+          }
+        });
+      }
+
       final sub = _connector.receivedFrames.listen((f) {
         if (f.isEmpty || f[0] != ObserverConfigClient.respConfig) return;
+        armIdle(); // re-arm: a steadily-streaming dump must never time out.
         final list = decoder.add(ObserverConfigClient.parse(f));
         if (list != null && !completer.isCompleted) completer.complete(list);
       });
       try {
         await _connector.sendFrame(ObserverConfigClient.buildBrokers());
-        return await completer.future.timeout(timeout);
+        armIdle(); // initial deadline: if the device never replies at all.
+        return await completer.future;
       } catch (e) {
         _setError('broker list failed: $e');
         return null;
       } finally {
+        idle?.cancel();
         await sub.cancel();
       }
     });
diff --git a/test/services/observer_config_service_test.dart b/test/services/observer_config_service_test.dart
index a975da9..fabb63f 100644
--- a/test/services/observer_config_service_test.dart
+++ b/test/services/observer_config_service_test.dart
@@ -134,6 +134,25 @@ Uint8List _respText(int sub, String text) => Uint8List.fromList([
   0,
 ]);
 
+Uint8List _brokersStart(int count) => Uint8List.fromList([
+  ObserverConfigClient.respConfig,
+  ObserverConfigClient.rBrokersStart,
+  count,
+]);
+
+Uint8List _brokerKv(int slot, String key, String value) => Uint8List.fromList([
+  ObserverConfigClient.respConfig,
+  ObserverConfigClient.rBrokerKv,
+  slot,
+  ...utf8.encode('$key=$value'),
+  0,
+]);
+
+Uint8List _brokersEnd() => Uint8List.fromList([
+  ObserverConfigClient.respConfig,
+  ObserverConfigClient.rBrokersEnd,
+]);
+
 Future<void> _tick() => Future<void>.delayed(const Duration(milliseconds: 10));
 
 void main() {
@@ -373,4 +392,33 @@ void main() {
       auto.closeStream();
     },
   );
+
+  // ---- #103: a slow-but-steady dump must outlast the fixed timeout ----
+  test('getBrokers completes a dump that outlasts the per-frame timeout while '
+      'frames keep arriving (re-armed inactivity watchdog) — #103', () async {
+    svc.timeout = const Duration(milliseconds: 80);
+    final future = svc.getBrokers();
+    final frames = <Uint8List>[
+      _brokersStart(2),
+      _brokerKv(0, 'url', 'mqtt://a'),
+      _brokerKv(0, 'enabled', '1'),
+      _brokerKv(1, 'url', 'mqtt://b'),
+      _brokerKv(1, 'enabled', '0'),
+      _brokersEnd(),
+    ];
+    // Each 50ms gap is under the 80ms timeout, but the whole dump (~300ms)
+    // far exceeds it: a fixed deadline kills it, a re-armed watchdog does not.
+    for (final f in frames) {
+      await Future<void>.delayed(const Duration(milliseconds: 50));
+      c.inject(f);
+    }
+    final list = await future;
+    expect(
+      list,
+      isNotNull,
+      reason: 'a steadily-streaming dump must not time out (#103)',
+    );
+    expect(list!.length, 2);
+    expect(list.map((b) => b.slot).toList(), [0, 1]);
+  });
 }