From 3e69f9bb3c7d068c6989d2425097b0af05fee16f Mon Sep 17 00:00:00 2001 From: sepehr-safari Date: Wed, 23 Sep 2026 12:54:54 +0300 Subject: [PATCH] fix: a relay that never answers the dial no longer holds req and fetch The relays were dialled one after another with no deadline, and --timeout only started counting once every dial had returned. A relay that accepted the connection and never answered the websocket upgrade held the run forever, and the relays listed after it were never reached. Every relay is now dialled at once, under a bound of five seconds or --timeout if that is shorter. A relay still dialling at the bound is cancelled, named on stderr, and left out, and the run carries on with the ones that answered. This needs nostr 0.14.4, where a cancelled dial stops at once and frees what it allocated, so the pin moves to it. Tests dial a real relay on loopback: one that never answers, a closed port and a live one, all at once and inside the deadline. Refs #19. --- build.zig.zon | 4 +- src/cmd_fetch.zig | 3 + src/cmd_req.zig | 10 ++- src/dial.zig | 146 ++++++++++++++++++++++++++++++++++++++++ src/main.zig | 2 + src/relayset.zig | 84 ++++++++++++++++++----- src/testrelay.zig | 168 ++++++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 396 insertions(+), 21 deletions(-) create mode 100644 src/dial.zig create mode 100644 src/testrelay.zig diff --git a/build.zig.zon b/build.zig.zon index 95c8fd3..e5ae9c4 100644 --- a/build.zig.zon +++ b/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 = .{ diff --git a/src/cmd_fetch.zig b/src/cmd_fetch.zig index 077860f..04a929c 100644 --- a/src/cmd_fetch.zig +++ b/src/cmd_fetch.zig @@ -26,6 +26,9 @@ pub const usage = \\ --store keep every event this receives in a local store \\ --timeout 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. \\ diff --git a/src/cmd_req.zig b/src/cmd_req.zig index 899a8d8..3ed1299 100644 --- a/src/cmd_req.zig +++ b/src/cmd_req.zig @@ -36,6 +36,9 @@ pub const usage = \\ --stream keep reading after the relays have sent what they hold \\ --timeout 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; diff --git a/src/dial.zig b/src/dial.zig new file mode 100644 index 0000000..d9a049c --- /dev/null +++ b/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); +} diff --git a/src/main.zig b/src/main.zig index f96fed6..ad2bd6b 100644 --- a/src/main.zig +++ b/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"); } diff --git a/src/relayset.zig b/src/relayset.zig index 4f5bd75..ba7ba42 100644 --- a/src/relayset.zig +++ b/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 diff --git a/src/testrelay.zig b/src/testrelay.zig new file mode 100644 index 0000000..fc87691 --- /dev/null +++ b/src/testrelay.zig @@ -0,0 +1,168 @@ +//! A relay for tests: it listens on loopback, answers the websocket upgrade, +//! and then does what its mode says. Nothing outside tests uses it. +//! +//! It is real enough to dial. The client side of every test that uses it is +//! the same `nostr.relay.dial` a user's run goes through, over a real socket. + +const std = @import("std"); +const nostr = @import("nostr"); + +const Io = std.Io; +const ws = nostr.websocket; + +pub const Mode = enum { + /// OK true to every EVENT, EOSE to every REQ. + accept, + /// OK false to every EVENT, with a reason. + refuse, + /// OK true to every EVENT, saying it already had it. + duplicate, + /// Answers the upgrade and then never sends another byte. + silent, + /// Never answers an EVENT, and sends a NOTICE every 100 ms, so a wait + /// that restarts on every message would never end. + chatty, + /// Sends a message the parser has no case for before each OK true. + unreadable_first, +}; + +/// The allocator a test hands to anything that dials. +/// +/// Still leak-checked, but it captures no stack traces. On macOS capturing one +/// takes a lock that notices a pending cancel and swallows it, so a dial that +/// is cancelled while it allocates never finds out and never returns, and the +/// test hangs instead of failing. `std.testing.allocator` captures them. +pub const DialAllocator = std.heap.DebugAllocator(.{ .stack_trace_frames = 0 }); + +/// A loopback listener on a port the kernel picks. +pub fn listen(io: Io) !Io.net.Server { + var address: Io.net.IpAddress = .{ .ip4 = .loopback(0) }; + return address.listen(io, .{ .reuse_address = true }); +} + +/// `ws://127.0.0.1:`, written into `buf`. +pub fn url(buf: []u8, port: u16) ![]const u8 { + return std.fmt.bufPrint(buf, "ws://127.0.0.1:{d}", .{port}); +} + +pub const Relay = struct { + server: Io.net.Server, + mode: Mode, + /// EVENT messages this relay has received, across every connection. + events: std.atomic.Value(usize) = .init(0), + task: Io.Future(void), + + /// Starts serving in the background. `self` must not move until `stop`. + pub fn start(self: *Relay, io: Io, mode: Mode) !void { + self.* = .{ .server = try listen(io), .mode = mode, .task = undefined }; + errdefer self.server.deinit(io); + self.task = try io.concurrent(serve, .{ self, io }); + } + + pub fn stop(self: *Relay, io: Io) void { + self.task.cancel(io); + self.server.deinit(io); + } + + pub fn port(self: *const Relay) u16 { + return self.server.socket.address.ip4.port; + } + + /// One connection, then done. A loop back to `accept` would be the one + /// place a cancel could be lost: a cancel that lands while the connection + /// is being read comes back as a read error, and once a task has seen its + /// cancel, a blocking call it makes afterwards can no longer be woken, so + /// `stop` would wait on that `accept` forever. + fn serve(self: *Relay, io: Io) void { + const conn = self.server.accept(io) catch return; + defer conn.close(io); + self.handle(io, conn) catch {}; + } + + fn handle(self: *Relay, io: Io, conn: Io.net.Stream) !void { + // Not `std.testing.allocator`, for the reason `DialAllocator` gives: + // this task is cancelled when the test stops the relay, and a cancel + // swallowed by a stack capture would leave `stop` waiting forever. + const gpa = std.heap.page_allocator; + var rbuf: [8192]u8 = undefined; + var wbuf: [8192]u8 = undefined; + var r = conn.reader(io, &rbuf); + var w = conn.writer(io, &wbuf); + + var req: std.ArrayList(u8) = .empty; + defer req.deinit(gpa); + while (std.mem.indexOf(u8, req.items, "\r\n\r\n") == null) { + try r.interface.fillMore(); + const got = r.interface.buffered(); + try req.appendSlice(gpa, got); + r.interface.toss(got.len); + } + const prefix = "Sec-WebSocket-Key: "; + const ks = (std.mem.indexOf(u8, req.items, prefix) orelse return error.NoKey) + prefix.len; + const ke = std.mem.indexOfPos(u8, req.items, ks, "\r\n") orelse return error.NoKey; + const accept_key = ws.acceptKey(req.items[ks..ke]); + try w.interface.print("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {s}\r\n\r\n", .{&accept_key}); + try w.interface.flush(); + + if (self.mode == .chatty) { + while (true) { + try io.sleep(.fromMilliseconds(100), .awake); + try sendText(&w.interface, "[\"NOTICE\",\"still here\"]"); + } + } + + var frames: std.ArrayList(u8) = .empty; + defer frames.deinit(gpa); + while (true) { + // Not `readSliceShort`: it blocks until its destination is full, + // so a frame smaller than the buffer would never surface. + r.interface.fillMore() catch return; + const avail = r.interface.buffered(); + try frames.appendSlice(gpa, avail); + r.interface.toss(avail.len); + while (try ws.decodeFrame(frames.items)) |f| { + if (f.opcode == .close) return; + if (f.opcode == .text) try self.answer(gpa, &w.interface, f.payload); + const n = f.frame_len; + std.mem.copyForwards(u8, frames.items, frames.items[n..]); + frames.shrinkRetainingCapacity(frames.items.len - n); + } + } + } + + fn answer(self: *Relay, gpa: std.mem.Allocator, w: *Io.Writer, payload: []const u8) !void { + if (self.mode == .silent) return; + var parsed = try std.json.parseFromSlice(std.json.Value, gpa, payload, .{}); + defer parsed.deinit(); + const items = parsed.value.array.items; + const kind = items[0].string; + var text: [512]u8 = undefined; + if (std.mem.eql(u8, kind, "REQ")) { + try sendText(w, try std.fmt.bufPrint(&text, "[\"EOSE\",\"{s}\"]", .{items[1].string})); + } else if (std.mem.eql(u8, kind, "EVENT")) { + _ = self.events.fetchAdd(1, .monotonic); + const id = items[1].object.get("id").?.string; + switch (self.mode) { + .accept => try sendText(w, try std.fmt.bufPrint(&text, "[\"OK\",\"{s}\",true,\"\"]", .{id})), + .refuse => try sendText(w, try std.fmt.bufPrint(&text, "[\"OK\",\"{s}\",false,\"invalid: test refusal\"]", .{id})), + .duplicate => try sendText(w, try std.fmt.bufPrint(&text, "[\"OK\",\"{s}\",true,\"duplicate: already have it\"]", .{id})), + .unreadable_first => { + try sendText(w, "[\"COUNT\",\"x\",{\"count\":1}]"); + try sendText(w, try std.fmt.bufPrint(&text, "[\"OK\",\"{s}\",true,\"\"]", .{id})); + }, + .silent, .chatty => {}, + } + } + } +}; + +/// One unmasked server text frame. +fn sendText(w: *Io.Writer, text: []const u8) !void { + if (text.len < 126) { + try w.writeAll(&.{ 0x81, @intCast(text.len) }); + } else { + try w.writeAll(&.{ 0x81, 126, @intCast(text.len >> 8), @intCast(text.len & 0xff) }); + } + try w.writeAll(text); + try w.flush(); +}