Bound the relay connect handshake so a fetch can't hang forever
This commit is contained in:
parent
0e70aba755
commit
60f9e4ccfa
3 changed files with 41 additions and 1 deletions
|
|
@ -31,15 +31,23 @@ final class EventsFetched extends FetchResult {
|
||||||
class NostrFetchActor with Handler<FetchMessage, FetchResult> {
|
class NostrFetchActor with Handler<FetchMessage, FetchResult> {
|
||||||
final List<String> _relayUrls;
|
final List<String> _relayUrls;
|
||||||
final Duration _eoseDeadline;
|
final Duration _eoseDeadline;
|
||||||
|
final Duration _connectDeadline;
|
||||||
final RelayConnection Function(String url) _connectionFactory;
|
final RelayConnection Function(String url) _connectionFactory;
|
||||||
|
|
||||||
NostrFetchActor({
|
NostrFetchActor({
|
||||||
required List<String> relayUrls,
|
required List<String> relayUrls,
|
||||||
Duration eoseDeadline = const Duration(seconds: 5),
|
Duration eoseDeadline = const Duration(seconds: 5),
|
||||||
|
// Bounds the WebSocket/TLS handshake. `connect()` awaits `channel.ready`
|
||||||
|
// with no deadline of its own, so a relay that accepts the socket but never
|
||||||
|
// completes the upgrade would hang the fetch forever — and, for a polling
|
||||||
|
// consumer, silently wedge its whole loop. On timeout the `finally` below
|
||||||
|
// closes the half-open socket and the relay is treated as down.
|
||||||
|
Duration connectDeadline = const Duration(seconds: 10),
|
||||||
RelayConnection Function(String url) connectionFactory =
|
RelayConnection Function(String url) connectionFactory =
|
||||||
WebSocketRelayConnection.new,
|
WebSocketRelayConnection.new,
|
||||||
}) : _relayUrls = relayUrls,
|
}) : _relayUrls = relayUrls,
|
||||||
_eoseDeadline = eoseDeadline,
|
_eoseDeadline = eoseDeadline,
|
||||||
|
_connectDeadline = connectDeadline,
|
||||||
_connectionFactory = connectionFactory;
|
_connectionFactory = connectionFactory;
|
||||||
|
|
||||||
@override
|
@override
|
||||||
|
|
@ -78,7 +86,7 @@ class NostrFetchActor with Handler<FetchMessage, FetchResult> {
|
||||||
final collected = <String>[];
|
final collected = <String>[];
|
||||||
|
|
||||||
try {
|
try {
|
||||||
await conn.connect();
|
await conn.connect().timeout(_connectDeadline);
|
||||||
await conn.send(jsonEncode(['REQ', subId, jsonDecode(filter)]));
|
await conn.send(jsonEncode(['REQ', subId, jsonDecode(filter)]));
|
||||||
try {
|
try {
|
||||||
await for (final raw in conn.messages.timeout(_eoseDeadline)) {
|
await for (final raw in conn.messages.timeout(_eoseDeadline)) {
|
||||||
|
|
|
||||||
|
|
@ -9,11 +9,15 @@ final class FakeRelayConnection implements RelayConnection {
|
||||||
bool isConnected = false;
|
bool isConnected = false;
|
||||||
bool autoRespond = true;
|
bool autoRespond = true;
|
||||||
bool throwOnConnect = false;
|
bool throwOnConnect = false;
|
||||||
|
// When set, connect() never completes — simulates a handshake that hangs
|
||||||
|
// (accepted socket, no WebSocket upgrade).
|
||||||
|
bool hangOnConnect = false;
|
||||||
void Function(String subId)? onReq;
|
void Function(String subId)? onReq;
|
||||||
|
|
||||||
@override
|
@override
|
||||||
Future<void> connect() async {
|
Future<void> connect() async {
|
||||||
if (throwOnConnect) throw Exception('connection failed');
|
if (throwOnConnect) throw Exception('connection failed');
|
||||||
|
if (hangOnConnect) await Completer<void>().future;
|
||||||
isConnected = true;
|
isConnected = true;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -120,6 +120,34 @@ void main() {
|
||||||
expect((result as EventsFetched).events, hasLength(1));
|
expect((result as EventsFetched).events, hasLength(1));
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('bounds a hung connect so the fetch cannot stall forever', () async {
|
||||||
|
final fakeHang = FakeRelayConnection()..hangOnConnect = true;
|
||||||
|
final fakeUp = FakeRelayConnection();
|
||||||
|
addTearDown(fakeHang.dispose);
|
||||||
|
addTearDown(fakeUp.dispose);
|
||||||
|
|
||||||
|
final event = _makeEvent('id1');
|
||||||
|
fakeUp.onReq = (subId) {
|
||||||
|
fakeUp.push(jsonEncode(['EVENT', subId, event]));
|
||||||
|
fakeUp.push(jsonEncode(['EOSE', subId]));
|
||||||
|
};
|
||||||
|
|
||||||
|
final fakes = {'ws://hang': fakeHang, 'ws://up': fakeUp};
|
||||||
|
final actor = NostrFetchActor(
|
||||||
|
relayUrls: ['ws://hang', 'ws://up'],
|
||||||
|
connectDeadline: const Duration(milliseconds: 50),
|
||||||
|
connectionFactory: (url) => fakes[url]!,
|
||||||
|
);
|
||||||
|
|
||||||
|
// The outer timeout is a guard: without the connect deadline this hangs.
|
||||||
|
final result = await actor
|
||||||
|
.handle(const FetchEvents(_filter))
|
||||||
|
.timeout(const Duration(seconds: 2));
|
||||||
|
|
||||||
|
expect(result, isA<EventsFetched>());
|
||||||
|
expect((result as EventsFetched).events, hasLength(1));
|
||||||
|
});
|
||||||
|
|
||||||
test('sends CLOSE after EOSE', () async {
|
test('sends CLOSE after EOSE', () async {
|
||||||
final fake = FakeRelayConnection();
|
final fake = FakeRelayConnection();
|
||||||
addTearDown(fake.dispose);
|
addTearDown(fake.dispose);
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue