A relay that never answers the dial no longer holds req and fetch

build.zig.zon

@@ -10,8 +10,8 @@
         // themselves later, and "whatever main happened to be that afternoon"
         // is not that.
         .nostr = .{
-            .url = "https://github.com/zig-nostr/nostr/archive/refs/tags/v0.14.2.tar.gz",
-            .hash = "nostr-0.14.2-CMyPzfnQCABuzpqY9v1y0PPErY5BBDOc5OWIUUerqYbU",
+            .url = "https://github.com/zig-nostr/nostr/archive/refs/tags/v0.14.4.tar.gz",
+            .hash = "nostr-0.14.4-CMyPze4BCQAUoi8JLNWfaR4ZDF1e4C5dhM2vMwGJ_qIT",
         },
     },
     .paths = .{

src/cmd_fetch.zig

@@ -26,6 +26,9 @@ pub const usage =
     \\      --store <path>   keep every event this receives in a local store
     \\      --timeout <ms>   give up on relays still answering (default 30000)
     \\
+    \\A relay that has not accepted the connection within five seconds, or
+    \\within --timeout if that is shorter, is named on stderr and left out.
+    \\
     \\A bare npub fetches that person's profile rather than everything they have
     \\ever written, which is what `npub` on its own can sensibly mean.
     \\

src/cmd_req.zig

@@ -36,6 +36,9 @@ pub const usage =
     \\      --stream         keep reading after the relays have sent what they hold
     \\      --timeout <ms>   give up on relays still answering (default 30000)
     \\
+    \\A relay that has not accepted the connection within five seconds, or
+    \\within --timeout if that is shorter, is named on stderr and left out.
+    \\
     \\Given no relay, it prints what it would send and stops, so a filter can be
     \\read before it is asked of anybody:
     \\
@@ -55,10 +58,11 @@ pub const Request = struct {
     timeout_ms: i64 = default_timeout_ms,
 };
 
-/// How long a run waits on relays still answering.
+/// How long a run waits on relays still answering, once they are reached.
+/// Reaching them has its own, shorter bound (`dial.default_timeout_ms`).
 ///
-/// Thirty seconds rather than nak's forever: long enough for a slow relay on a
-/// slow link, short enough that a script does not hang on one gone quiet.
+/// Thirty seconds: long enough for a slow relay on a slow link, short enough
+/// that a script does not hang on one gone quiet.
 pub const default_timeout_ms: i64 = 30_000;
 
 const max_values = 64;

src/dial.zig

@@ -0,0 +1,146 @@
+//! Reaching a set of relays at once, under one deadline.
+//!
+//! `nostr.relay.dial` takes no deadline of its own: a relay that accepts the
+//! connection and then never answers the websocket upgrade would hold it, and
+//! the process, forever. So every dial runs concurrently, a timer runs beside
+//! them, and whatever is still dialling when the timer fires is cancelled. A
+//! cancelled dial stops where it is and frees what it allocated.
+//!
+//! The one step a cancel cannot cut short is the name lookup, which is a plain
+//! libc call: a resolver that hangs still holds its dial until it returns.
+
+const std = @import("std");
+const nostr = @import("nostr");
+
+const Io = std.Io;
+const relay = nostr.relay;
+
+/// How long a relay gets to accept the connection, finish TLS and answer the
+/// websocket upgrade. A relay that needs longer than this to say hello is not
+/// one a command line should be waiting on.
+pub const default_timeout_ms: i64 = 5_000;
+
+/// The most relays one run will dial.
+pub const max_relays = 32;
+
+pub const Outcome = union(enum) {
+    connected: *relay.Relay,
+    /// The dial failed on its own, before the deadline.
+    failed: anyerror,
+    /// Still dialling when the deadline passed.
+    timed_out,
+};
+
+const DialResult = @typeInfo(@TypeOf(relay.dial)).@"fn".return_type.?;
+
+const Finished = struct {
+    index: usize,
+    result: DialResult,
+};
+
+const Arrival = union(enum) {
+    dial: Finished,
+    timer: Io.Cancelable!void,
+};
+
+fn dialOne(gpa: std.mem.Allocator, io: Io, url: []const u8, index: usize) Finished {
+    return .{ .index = index, .result = relay.dial(gpa, io, url) };
+}
+
+/// Dials every one of `urls` at once and writes url `i`'s outcome to `out[i]`.
+///
+/// Returns once every dial has finished or `timeout_ms` has passed, whichever
+/// comes first. Every `connected` relay belongs to the caller.
+pub fn all(gpa: std.mem.Allocator, io: Io, urls: []const []const u8, timeout_ms: i64, out: []Outcome) void {
+    std.debug.assert(urls.len <= max_relays);
+    std.debug.assert(out.len >= urls.len);
+    for (out[0..urls.len]) |*o| o.* = .timed_out;
+
+    // One slot per task that can finish, or `cancel` below deadlocks waiting
+    // for room to put a result.
+    var buf: [max_relays + 1]Arrival = undefined;
+    var sel = Io.Select(Arrival).init(io, &buf);
+
+    var pending: usize = 0;
+    for (urls, 0..) |url, i| {
+        // `concurrent`, never `async`: past its limit `async` runs the task
+        // inline, and an inline dial is exactly the unbounded wait this avoids.
+        sel.concurrent(.dial, dialOne, .{ gpa, io, url, i }) catch |e| {
+            out[i] = .{ .failed = e };
+            continue;
+        };
+        pending += 1;
+    }
+
+    if (pending > 0) {
+        // Should the timer fail to start, the dials are only as bounded as
+        // they were before this existed, which still beats not dialling.
+        sel.concurrent(.timer, Io.sleep, .{ io, .fromMilliseconds(timeout_ms), .awake }) catch {};
+    }
+
+    while (pending > 0) {
+        const arrival = sel.await() catch break;
+        switch (arrival) {
+            .dial => |f| {
+                out[f.index] = if (f.result) |r| .{ .connected = r } else |e| .{ .failed = e };
+                pending -= 1;
+            },
+            .timer => break,
+        }
+    }
+
+    // Whatever is left missed the deadline. Cancelling joins each task, and a
+    // dial that happened to finish in the same instant is kept rather than
+    // thrown away: it is a live connection, and the caller asked for one.
+    while (sel.cancel()) |late| switch (late) {
+        .dial => |f| if (f.result) |r| {
+            out[f.index] = .{ .connected = r };
+        } else |_| {},
+        .timer => {},
+    };
+}
+
+test "a silent relay times out, a closed port fails, and a live relay connects, all at once" {
+    const io = std.testing.io;
+    const testrelay = @import("testrelay.zig");
+    var da: testrelay.DialAllocator = .init;
+    defer if (da.deinit() == .leak) @panic("the dials leaked");
+    const gpa = da.allocator();
+
+    // Listening, and never accepting: the kernel completes the TCP handshake
+    // from its backlog, so the dial sends its upgrade and waits for an answer
+    // that never comes.
+    var silent = try testrelay.listen(io);
+    defer silent.deinit(io);
+
+    // Bound and closed again, so nothing is listening there.
+    var closed = try testrelay.listen(io);
+    const closed_port = closed.socket.address.ip4.port;
+    closed.deinit(io);
+
+    var live: testrelay.Relay = undefined;
+    try live.start(io, .accept);
+    defer live.stop(io);
+
+    var urls_buf: [3][40]u8 = undefined;
+    const urls = [_][]const u8{
+        try testrelay.url(&urls_buf[0], silent.socket.address.ip4.port),
+        try testrelay.url(&urls_buf[1], closed_port),
+        try testrelay.url(&urls_buf[2], live.port()),
+    };
+
+    var out: [3]Outcome = undefined;
+    const started = Io.Timestamp.now(io, .awake).toMilliseconds();
+    all(gpa, io, &urls, 300, &out);
+    const took = Io.Timestamp.now(io, .awake).toMilliseconds() - started;
+    defer for (out) |o| switch (o) {
+        .connected => |r| r.deinit(),
+        else => {},
+    };
+
+    try std.testing.expect(out[0] == .timed_out);
+    try std.testing.expect(out[1] == .failed);
+    try std.testing.expect(out[2] == .connected);
+    // Bounded by the deadline, not by the slowest relay.
+    try std.testing.expect(took < 3_000);
+}

src/main.zig

@@ -322,5 +322,7 @@ test {
     _ = @import("cmd_publish.zig");
     _ = @import("cmd_req.zig");
     _ = @import("relayset.zig");
+    _ = @import("dial.zig");
+    _ = @import("testrelay.zig");
     _ = @import("cmd_verify.zig");
 }

src/relayset.zig

@@ -11,6 +11,7 @@
 
 const std = @import("std");
 const nostr = @import("nostr");
+const dial = @import("dial.zig");
 
 const filter = nostr.filter;
 const message = nostr.message;
@@ -23,15 +24,13 @@ pub const Options = struct {
     /// so this is the difference between "everything you have" and "everything
     /// you have, and then whatever arrives while I wait".
     until_eose: bool = true,
-    /// The whole run gives up here, whatever the relays are doing.
+    /// The run gives up on relays still answering here, whatever they are
+    /// doing. Reaching them has its own, shorter bound: see `dial`.
     ///
-    /// nak has no equivalent and this is a deliberate difference. Its per-relay
-    /// select waits on EOSE, a close or an event and nothing else, so a relay
-    /// that accepts a subscription and then says nothing holds the process open
-    /// for as long as somebody lets it (go-nostr's pool.go:677-729, no timer in
-    /// that select). That is survivable at an interactive prompt and not
-    /// survivable in a script, which is where a command line spends most of its
-    /// life.
+    /// A relay can stall at any point: while connecting, during TLS, at the
+    /// websocket upgrade, or after accepting the subscription. Waiting on it
+    /// is survivable at an interactive prompt and not in a script, which is
+    /// where a command line spends most of its life.
     deadline_ms: i64,
     /// How long one read may block before the loop moves to the next relay.
     /// Short, because it is a round-robin across relays rather than a wait.
@@ -51,7 +50,7 @@ pub const Outcome = struct {
     complete: usize = 0,
 };
 
-const max_relays = 32;
+const max_relays = dial.max_relays;
 
 /// Whether an event answers any of the questions this run asked.
 ///
@@ -96,14 +95,29 @@ pub fn query(
         }
     };
 
+    // All at once, so the slowest relay costs its own wait and not everybody
+    // else's too, and never longer than the run itself may take.
+    var dialled: [max_relays]dial.Outcome = undefined;
+    const dial_ms = @min(dial.default_timeout_ms, opts.deadline_ms);
+    dial.all(gpa, io, urls[0..n], dial_ms, &dialled);
+
     for (urls[0..n], 0..) |url, i| {
-        const r = relay.dial(gpa, io, url) catch |e| {
-            // Named, not swallowed. A run that quietly asked three relays
-            // instead of four looks like the fourth had nothing.
-            try err.print("deed: {s}: {s}\n", .{ url, @errorName(e) });
-            result.unreachable_count += 1;
-            done[i] = true;
-            continue;
+        // Named, not swallowed. A run that quietly asked three relays instead
+        // of four looks like the fourth had nothing.
+        const r = switch (dialled[i]) {
+            .connected => |r| r,
+            .failed => |e| {
+                try err.print("deed: {s}: {s}\n", .{ url, @errorName(e) });
+                result.unreachable_count += 1;
+                done[i] = true;
+                continue;
+            },
+            .timed_out => {
+                try err.print("deed: {s}: no answer within {d} ms\n", .{ url, dial_ms });
+                result.unreachable_count += 1;
+                done[i] = true;
+                continue;
+            },
         };
         r.subscribe(subscription_id, filters) catch |e| {
             try err.print("deed: {s}: {s}\n", .{ url, @errorName(e) });
@@ -208,6 +222,44 @@ pub fn query(
     return result;
 }
 
+test "a relay that never answers the dial costs the run its deadline, not forever" {
+    const io = std.testing.io;
+    const testrelay = @import("testrelay.zig");
+    var da: testrelay.DialAllocator = .init;
+    defer if (da.deinit() == .leak) @panic("the query leaked");
+    const gpa = da.allocator();
+
+    // Listed first, where a dial that waited on it one relay at a time would
+    // never have reached the one behind it.
+    var silent = try testrelay.listen(io);
+    defer silent.deinit(io);
+    var live: testrelay.Relay = undefined;
+    try live.start(io, .accept);
+    defer live.stop(io);
+
+    var bufs: [2][40]u8 = undefined;
+    const urls = [_][]const u8{
+        try testrelay.url(&bufs[0], silent.socket.address.ip4.port),
+        try testrelay.url(&bufs[1], live.port()),
+    };
+    const filters = [_]filter.Filter{.{ .kinds = &.{1}, .limit = 1 }};
+
+    var out_buf: [1024]u8 = undefined;
+    var out: std.Io.Writer = .fixed(&out_buf);
+    var err_buf: [1024]u8 = undefined;
+    var err: std.Io.Writer = .fixed(&err_buf);
+
+    const started = std.Io.Timestamp.now(io, .awake).toMilliseconds();
+    const outcome = try query(gpa, io, &urls, &filters, &out, &err, .{ .deadline_ms = 500, .poll_ms = 50 });
+    const took = std.Io.Timestamp.now(io, .awake).toMilliseconds() - started;
+
+    try std.testing.expectEqual(@as(usize, 1), outcome.dialled);
+    try std.testing.expectEqual(@as(usize, 1), outcome.unreachable_count);
+    try std.testing.expectEqual(@as(usize, 1), outcome.complete);
+    try std.testing.expect(std.mem.indexOf(u8, err.buffered(), "no answer within 500 ms") != null);
+    try std.testing.expect(took < 3_000);
+}
+
 test {
     // Forces this file to be analysed. `_ = @import("relayset.zig")` alone
     // imports it without ever compiling a function body nobody references, so