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>
pull/109/head
Strycher 4 weeks ago
parent 1f0406d515
commit a124100acc

@ -203,23 +203,41 @@ class ObserverConfigService extends ChangeNotifier {
return BrokerConfig.fromWireFields(slot, kv); 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() { Future<List<BrokerConfig>?> getBrokers() {
return _serialized(() async { return _serialized(() async {
final decoder = BrokerListDecoder(); final decoder = BrokerListDecoder();
final completer = Completer<List<BrokerConfig>>(); 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) { final sub = _connector.receivedFrames.listen((f) {
if (f.isEmpty || f[0] != ObserverConfigClient.respConfig) return; 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)); final list = decoder.add(ObserverConfigClient.parse(f));
if (list != null && !completer.isCompleted) completer.complete(list); if (list != null && !completer.isCompleted) completer.complete(list);
}); });
try { try {
await _connector.sendFrame(ObserverConfigClient.buildBrokers()); 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) { } catch (e) {
_setError('broker list failed: $e'); _setError('broker list failed: $e');
return null; return null;
} finally { } finally {
idle?.cancel();
await sub.cancel(); await sub.cancel();
} }
}); });

@ -134,6 +134,25 @@ Uint8List _respText(int sub, String text) => Uint8List.fromList([
0, 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)); Future<void> _tick() => Future<void>.delayed(const Duration(milliseconds: 10));
void main() { void main() {
@ -373,4 +392,33 @@ void main() {
auto.closeStream(); 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]);
});
} }

Loading…
Cancel
Save

Powered by TurnKey Linux.