src/server/server_forward.zig
Ref: Size: 34.1 KiB History
//! Dedicated forwarding-role peers and their remote-loopback TCP channels.
//! This is daemon state, but deliberately has no Session or terminal imports:
//! accepting a forwarding role cannot create a shell or claim geometry.
const std = @import("std");
const proto = @import("term").protocol;
const quic_server = @import("quic_server.zig");
const server_os = @import("server_os");
pub const max_peers: usize = 4;
pub const poll_len: usize = max_peers * (1 + proto.forward_channels_max);
const peer_queue_max: usize = 1024 * 1024;
const control_reserve: usize = 64 * 1024;
const channel_queue_max: usize = proto.forward_initial_credit;
const frames_per_service: usize = 64;
const connect_deadline_ms: i64 = 5_000;
fn monoMs() i64 {
const t = std.posix.clock_gettime(.MONOTONIC) catch return std.time.milliTimestamp();
return @as(i64, t.sec) * 1000 + @divFloor(t.nsec, 1_000_000);
}
const Sink = union(enum) {
socket: std.posix.fd_t,
quic: struct { listener: *quic_server.Listener, id: u64 },
fn pollFd(self: Sink) std.posix.fd_t {
return switch (self) {
.socket => |fd| fd,
.quic => -1,
};
}
fn send(self: Sink, bytes: []const u8) !usize {
return switch (self) {
.socket => |fd| server_os.sendNoSigNoWait(fd, bytes),
.quic => |q| q.listener.send(q.id, bytes),
};
}
fn inFlight(self: Sink) usize {
return switch (self) {
.socket => 0,
.quic => |q| q.listener.pendingBytes(q.id),
};
}
fn close(self: Sink) void {
switch (self) {
.socket => |fd| std.posix.close(fd),
.quic => |q| q.listener.closeConn(q.id),
}
}
};
const Channel = struct {
generation: u64,
id: u32,
fd: std.posix.fd_t,
connecting: bool,
/// Monotonic deadline for EINPROGRESS. A descriptor that never wakes must
/// not occupy a forwarding channel forever.
connect_by_ms: ?i64 = null,
send_credit: u32 = 0,
recv_credit: u32 = 0,
to_tcp: std.ArrayList(u8) = .empty,
local_eof: bool = false,
peer_eof: bool = false,
write_shutdown: bool = false,
};
const Peer = struct {
generation: u64,
sink: Sink,
inbound: std.ArrayList(u8) = .empty,
pending: std.ArrayList(u8) = .empty,
channels: [proto.forward_channels_max]?Channel = @splat(null),
};
pub const Relay = struct {
alloc: std.mem.Allocator,
peers: [max_peers]?Peer = @splat(null),
next_generation: u64 = 1,
poll_peer_generation: [max_peers]u64 = @splat(0),
poll_channel_generation: [max_peers * proto.forward_channels_max]u64 = @splat(0),
pub fn init(alloc: std.mem.Allocator) Relay {
return .{ .alloc = alloc };
}
pub fn deinit(self: *Relay) void {
for (0..max_peers) |i| self.dropPeer(i, true);
}
pub fn freePeer(self: *const Relay) ?usize {
for (self.peers, 0..) |peer, i| if (peer == null) return i;
return null;
}
pub const Adopt = enum { adopted, no_room };
pub fn adoptSocket(self: *Relay, fd: std.posix.fd_t, inbound: std.ArrayList(u8)) Adopt {
const i = self.freePeer() orelse return .no_room;
self.peers[i] = .{ .generation = self.takeGeneration(), .sink = .{ .socket = fd }, .inbound = inbound };
self.ready(i);
self.drainFrames(i);
return .adopted;
}
pub fn adoptQuic(self: *Relay, listener: *quic_server.Listener, id: u64, inbound: std.ArrayList(u8)) Adopt {
const i = self.freePeer() orelse return .no_room;
self.peers[i] = .{ .generation = self.takeGeneration(), .sink = .{ .quic = .{ .listener = listener, .id = id } }, .inbound = inbound };
self.ready(i);
self.drainFrames(i);
return .adopted;
}
pub fn pushQuic(self: *Relay, id: u64, bytes: []const u8) bool {
for (self.peers, 0..) |peer, i| {
const p = peer orelse continue;
if (p.sink == .quic and p.sink.quic.id == id) {
self.push(i, bytes);
return true;
}
}
return false;
}
pub fn closeQuic(self: *Relay, id: u64) bool {
for (self.peers, 0..) |peer, i| {
const p = peer orelse continue;
if (p.sink == .quic and p.sink.quic.id == id) {
self.dropPeer(i, false);
return true;
}
}
return false;
}
pub fn flushQuic(self: *Relay) void {
for (0..max_peers) |i| {
const p = if (self.peers[i]) |*peer| peer else continue;
if (p.sink == .quic) self.flushPeer(i);
}
}
pub fn fillPoll(self: *Relay, fds: []std.posix.pollfd) void {
std.debug.assert(fds.len == poll_len);
var at: usize = 0;
for (self.peers, 0..) |peer, pi| {
if (peer) |p| {
self.poll_peer_generation[pi] = p.generation;
var events: i16 = std.posix.POLL.IN;
if (p.pending.items.len != 0) events |= std.posix.POLL.OUT;
fds[at] = .{ .fd = p.sink.pollFd(), .events = events, .revents = 0 };
at += 1;
for (p.channels, 0..) |channel, ci| {
if (channel) |ch| {
var ce: i16 = 0;
if (!ch.local_eof and self.dataReadLimit(p, ch) != 0) ce |= std.posix.POLL.IN;
if (ch.connecting or ch.to_tcp.items.len != 0 or (ch.peer_eof and !ch.write_shutdown)) ce |= std.posix.POLL.OUT;
fds[at] = .{ .fd = if (ce == 0) -1 else ch.fd, .events = ce, .revents = 0 };
self.poll_channel_generation[pi * proto.forward_channels_max + ci] = ch.generation;
} else fds[at] = .{ .fd = -1, .events = 0, .revents = 0 };
at += 1;
}
} else {
self.poll_peer_generation[pi] = 0;
for (0..1 + proto.forward_channels_max) |_| {
fds[at] = .{ .fd = -1, .events = 0, .revents = 0 };
at += 1;
}
}
}
}
pub fn servicePoll(self: *Relay, fds: []const std.posix.pollfd) void {
std.debug.assert(fds.len == poll_len);
self.reapConnecting(monoMs());
var at: usize = 0;
for (0..max_peers) |pi| {
const peer_events = fds[at].revents;
at += 1;
const channel_start = at;
at += proto.forward_channels_max;
if (self.peers[pi] == null or self.peers[pi].?.generation != self.poll_peer_generation[pi]) continue;
// Service the descriptors represented by this poll snapshot before
// peer frames can reset and reuse their table slots.
for (0..proto.forward_channels_max) |ci| {
if (self.peers[pi] == null) break;
if (self.peers[pi].?.channels[ci] == null) continue;
if (self.peers[pi].?.channels[ci].?.generation != self.poll_channel_generation[pi * proto.forward_channels_max + ci]) continue;
const revents = fds[channel_start + ci].revents;
if (revents == 0) continue;
self.serviceChannel(pi, ci, revents);
}
if (self.peers[pi] == null or self.peers[pi].?.generation != self.poll_peer_generation[pi]) continue;
if (peer_events & std.posix.POLL.OUT != 0) self.flushPeer(pi);
if (self.peers[pi] == null) continue;
if (peer_events & ~@as(i16, std.posix.POLL.OUT) != 0) self.readPeer(pi);
if (self.peers[pi] != null) self.drainFrames(pi);
}
}
pub fn backlog(self: *const Relay) bool {
for (self.peers) |peer| {
const p = peer orelse continue;
const framed = proto.delimitFrame(p.inbound.items) catch return true;
if (framed != null) return true;
}
return false;
}
fn ready(self: *Relay, pi: usize) void {
const payload = proto.encodeForwardHello();
self.queue(pi, .forward_ready, &payload);
}
fn push(self: *Relay, pi: usize, bytes: []const u8) void {
const p = if (self.peers[pi]) |*peer| peer else return;
if (p.inbound.items.len + bytes.len > peer_queue_max) return self.dropPeer(pi, true);
p.inbound.appendSlice(self.alloc, bytes) catch return self.dropPeer(pi, true);
self.drainFrames(pi);
}
fn readPeer(self: *Relay, pi: usize) void {
const p = if (self.peers[pi]) |*peer| peer else return;
if (p.sink != .socket) return;
var buf: [64 * 1024]u8 = undefined;
const n = std.posix.read(p.sink.socket, &buf) catch |err| switch (err) {
error.WouldBlock => return,
else => return self.dropPeer(pi, true),
};
if (n == 0) return self.dropPeer(pi, true);
self.push(pi, buf[0..n]);
}
fn drainFrames(self: *Relay, pi: usize) void {
for (0..frames_per_service) |_| {
const p = if (self.peers[pi]) |*peer| peer else return;
const frame = proto.takeFrame(self.alloc, &p.inbound) catch return self.dropPeer(pi, true);
if (frame == null) return;
const f = frame.?;
defer f.deinit(self.alloc);
if (!self.handleFrame(pi, f)) return self.dropPeer(pi, true);
}
}
fn handleFrame(self: *Relay, pi: usize, frame: proto.Frame) bool {
switch (frame.type) {
.forward_open => {
const request = proto.decodeForwardOpen(frame.payload) catch return false;
self.openChannel(pi, request);
},
.forward_data => {
if (proto.forwardDataOversize(frame.payload)) return false;
const id = proto.decodeForwardId(frame.payload) catch return false;
const ci = self.findChannel(pi, id) orelse return true; // late frame for a reset channel
const ch = &self.peers[pi].?.channels[ci].?;
const data = frame.payload[proto.forward_id_len..];
if (ch.peer_eof or data.len > ch.recv_credit or ch.to_tcp.items.len + data.len > channel_queue_max) {
self.resetChannel(pi, ci, true);
return true;
}
ch.recv_credit -= @intCast(data.len);
ch.to_tcp.appendSlice(self.alloc, data) catch self.resetChannel(pi, ci, true);
if (self.peers[pi] != null and self.peers[pi].?.channels[ci] != null) self.flushTcp(pi, ci);
},
.forward_credit => {
const credit = proto.decodeForwardCredit(frame.payload) catch return false;
const ci = self.findChannel(pi, credit.id) orelse return true;
const ch = &self.peers[pi].?.channels[ci].?;
if (credit.amount > proto.forward_initial_credit -| ch.send_credit) return false;
ch.send_credit += credit.amount;
},
.forward_half_close => {
const id = proto.decodeForwardId(frame.payload) catch return false;
if (frame.payload.len != proto.forward_id_len) return false;
const ci = self.findChannel(pi, id) orelse return true;
self.peers[pi].?.channels[ci].?.peer_eof = true;
self.flushTcp(pi, ci);
},
.forward_reset => {
const id = proto.decodeForwardId(frame.payload) catch return false;
if (frame.payload.len != proto.forward_id_len) return false;
const ci = self.findChannel(pi, id) orelse return true;
self.resetChannel(pi, ci, false);
},
else => return false,
}
return self.peers[pi] != null;
}
fn openChannel(self: *Relay, pi: usize, request: proto.ForwardOpen) void {
if (self.findChannel(pi, request.id) != null) return self.refuseOpen(pi, request.id);
var ci: ?usize = null;
for (self.peers[pi].?.channels, 0..) |ch, i| if (ch == null) {
ci = i;
break;
};
const slot = ci orelse return self.refuseOpen(pi, request.id);
const fd = std.posix.socket(std.posix.AF.INET, std.posix.SOCK.STREAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC, std.posix.IPPROTO.TCP) catch return self.refuseOpen(pi, request.id);
const addr = std.net.Address.parseIp4("127.0.0.1", request.port) catch unreachable;
var connecting = false;
std.posix.connect(fd, &addr.any, addr.getOsSockLen()) catch |err| switch (err) {
error.WouldBlock => connecting = true,
else => {
std.posix.close(fd);
return self.refuseOpen(pi, request.id);
},
};
self.peers[pi].?.channels[slot] = .{ .generation = self.takeGeneration(), .id = request.id, .fd = fd, .connecting = connecting, .connect_by_ms = if (connecting) monoMs() + connect_deadline_ms else null };
if (!connecting) self.acceptOpen(pi, slot);
}
fn acceptOpen(self: *Relay, pi: usize, ci: usize) void {
const id = self.peers[pi].?.channels[ci].?.id;
self.peers[pi].?.channels[ci].?.connecting = false;
self.peers[pi].?.channels[ci].?.connect_by_ms = null;
self.peers[pi].?.channels[ci].?.recv_credit = proto.forward_initial_credit;
const result = proto.encodeForwardOpenResult(.{ .id = id, .ok = true });
self.queue(pi, .forward_open_result, &result);
const credit = proto.encodeForwardCredit(.{ .id = id, .amount = proto.forward_initial_credit });
self.queue(pi, .forward_credit, &credit);
}
fn refuseOpen(self: *Relay, pi: usize, id: u32) void {
const result = proto.encodeForwardOpenResult(.{ .id = id, .ok = false });
self.queue(pi, .forward_open_result, &result);
}
/// Reaped from the regular poll service even if a nonblocking connect
/// never produces POLLOUT/POLLERR. Only the expired channel is refused.
fn reapConnecting(self: *Relay, now_ms: i64) void {
for (0..max_peers) |pi| {
for (0..proto.forward_channels_max) |ci| {
const ch = self.peers[pi] orelse continue;
const channel = ch.channels[ci] orelse continue;
const by = channel.connect_by_ms orelse continue;
if (channel.connecting and now_ms >= by) {
const id = channel.id;
self.resetChannel(pi, ci, false);
self.refuseOpen(pi, id);
}
}
}
}
fn serviceChannel(self: *Relay, pi: usize, ci: usize, revents: i16) void {
var ch = &self.peers[pi].?.channels[ci].?;
// Error and hangup bits are reported even when they were not requested.
// SO_ERROR is the verdict for every connect wakeup; requiring POLLOUT
// can otherwise leave a refused channel permanently "connecting".
if (ch.connecting and revents != 0) {
std.posix.getsockoptError(ch.fd) catch {
const id = ch.id;
self.resetChannel(pi, ci, false);
return self.refuseOpen(pi, id);
};
self.acceptOpen(pi, ci);
}
if (self.peers[pi] == null or self.peers[pi].?.channels[ci] == null) return;
ch = &self.peers[pi].?.channels[ci].?;
if (!ch.connecting and revents & std.posix.POLL.OUT != 0) self.flushTcp(pi, ci);
if (self.peers[pi] == null or self.peers[pi].?.channels[ci] == null) return;
ch = &self.peers[pi].?.channels[ci].?;
if (!ch.connecting and revents & ~@as(i16, std.posix.POLL.OUT) != 0) self.readTcp(pi, ci);
}
fn flushTcp(self: *Relay, pi: usize, ci: usize) void {
const ch = &self.peers[pi].?.channels[ci].?;
if (ch.connecting) return;
if (ch.to_tcp.items.len != 0) {
const n = server_os.sendNoSigNoWait(ch.fd, ch.to_tcp.items) catch |err| switch (err) {
error.WouldBlock => return,
else => return self.resetChannel(pi, ci, true),
};
ch.to_tcp.replaceRangeAssumeCapacity(0, n, &.{});
ch.recv_credit += @intCast(n);
const credit = proto.encodeForwardCredit(.{ .id = ch.id, .amount = @intCast(n) });
self.queue(pi, .forward_credit, &credit);
}
if (self.peers[pi] == null or self.peers[pi].?.channels[ci] == null) return;
const current = &self.peers[pi].?.channels[ci].?;
if (current.peer_eof and current.to_tcp.items.len == 0 and !current.write_shutdown) {
std.posix.shutdown(current.fd, .send) catch {};
current.write_shutdown = true;
}
self.maybeFinish(pi, ci);
}
fn readTcp(self: *Relay, pi: usize, ci: usize) void {
const ch = &self.peers[pi].?.channels[ci].?;
if (ch.local_eof or ch.send_credit == 0) return;
// All readable channels in this poll pass observed one shared queue.
// Recheck it for this channel before taking bytes from the TCP socket.
const want = self.dataReadLimit(self.peers[pi].?, ch.*);
if (want == 0) return;
var buf: [proto.forward_data_max]u8 = undefined;
const n = std.posix.read(ch.fd, buf[0..want]) catch |err| switch (err) {
error.WouldBlock => return,
else => return self.resetChannel(pi, ci, true),
};
if (n == 0) {
ch.local_eof = true;
const id = proto.encodeForwardId(ch.id);
self.queue(pi, .forward_half_close, &id);
if (self.peers[pi] == null or self.peers[pi].?.channels[ci] == null) return;
return self.maybeFinish(pi, ci);
}
ch.send_credit -= @intCast(n);
var payload: [proto.forward_id_len + proto.forward_data_max]u8 = undefined;
std.mem.writeInt(u32, payload[0..4], ch.id, .little);
@memcpy(payload[4..][0..n], buf[0..n]);
self.queue(pi, .forward_data, payload[0 .. 4 + n]);
}
fn maybeFinish(self: *Relay, pi: usize, ci: usize) void {
if (self.peers[pi] == null or self.peers[pi].?.channels[ci] == null) return;
const ch = &self.peers[pi].?.channels[ci].?;
if (ch.local_eof and ch.peer_eof and ch.to_tcp.items.len == 0) self.resetChannel(pi, ci, false);
}
fn findChannel(self: *const Relay, pi: usize, id: u32) ?usize {
const p = self.peers[pi] orelse return null;
for (p.channels, 0..) |ch, i| if (ch != null and ch.?.id == id) return i;
return null;
}
fn resetChannel(self: *Relay, pi: usize, ci: usize, notify: bool) void {
const p = if (self.peers[pi]) |*peer| peer else return;
if (p.channels[ci]) |*ch| {
const id = ch.id;
std.posix.close(ch.fd);
ch.to_tcp.deinit(self.alloc);
p.channels[ci] = null;
if (notify) {
const payload = proto.encodeForwardId(id);
self.queue(pi, .forward_reset, &payload);
}
}
}
fn queue(self: *Relay, pi: usize, kind: proto.MsgType, payload: []const u8) void {
const p = if (self.peers[pi]) |*peer| peer else return;
const frame_len = proto.frame_header_len + payload.len;
if (p.pending.items.len + p.sink.inFlight() > peer_queue_max -| frame_len) return self.dropPeer(pi, true);
proto.appendFrame(&p.pending, self.alloc, kind, payload) catch return self.dropPeer(pi, true);
self.flushPeer(pi);
}
fn flushPeer(self: *Relay, pi: usize) void {
const p = if (self.peers[pi]) |*peer| peer else return;
if (p.pending.items.len == 0) return;
const n = p.sink.send(p.pending.items) catch |err| switch (err) {
error.WouldBlock => return,
else => return self.dropPeer(pi, true),
};
p.pending.replaceRangeAssumeCapacity(0, n, &.{});
}
fn dropPeer(self: *Relay, pi: usize, close_sink: bool) void {
if (self.peers[pi]) |*p| {
for (0..proto.forward_channels_max) |ci| self.resetChannel(pi, ci, false);
p.inbound.deinit(self.alloc);
p.pending.deinit(self.alloc);
if (close_sink) p.sink.close();
}
self.peers[pi] = null;
}
fn takeGeneration(self: *Relay) u64 {
const result = self.next_generation;
self.next_generation +%= 1;
if (self.next_generation == 0) self.next_generation = 1;
return result;
}
fn dataReadLimit(_: *const Relay, peer: Peer, ch: Channel) usize {
const overhead = proto.frame_header_len + proto.forward_id_len;
const pending = peer.pending.items.len + peer.sink.inFlight();
if (pending >= peer_queue_max - control_reserve - overhead) return 0;
const room = peer_queue_max - control_reserve - overhead - pending;
return @min(proto.forward_data_max, @min(@as(usize, ch.send_credit), room));
}
};
fn testSocketPair() ![2]std.posix.fd_t {
var pair: [2]std.posix.fd_t = undefined;
if (std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair) != 0) return error.SocketPairFailed;
return pair;
}
fn testListener() !struct { fd: std.posix.fd_t, port: u16 } {
const fd = try std.posix.socket(std.posix.AF.INET, std.posix.SOCK.STREAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC, std.posix.IPPROTO.TCP);
errdefer std.posix.close(fd);
const addr = try std.net.Address.parseIp4("127.0.0.1", 0);
try std.posix.bind(fd, &addr.any, addr.getOsSockLen());
try std.posix.listen(fd, 8);
var actual: std.posix.sockaddr.storage = undefined;
var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
try std.posix.getsockname(fd, @ptrCast(&actual), &len);
return .{ .fd = fd, .port = std.net.Address.initPosix(@ptrCast(@alignCast(&actual))).getPort() };
}
fn testPush(relay: *Relay, kind: proto.MsgType, payload: []const u8) !void {
var bytes: std.ArrayList(u8) = .empty;
defer bytes.deinit(std.testing.allocator);
try proto.appendFrame(&bytes, std.testing.allocator, kind, payload);
relay.push(0, bytes.items);
}
fn testPump(relay: *Relay, timeout_ms: i32) !void {
var fds: [poll_len]std.posix.pollfd = undefined;
relay.fillPoll(&fds);
_ = try std.posix.poll(&fds, timeout_ms);
relay.servicePoll(&fds);
}
fn testReadFrame(fd: std.posix.fd_t, expected: proto.MsgType) !proto.Frame {
const frame = (try proto.readFrame(std.testing.allocator, fd)) orelse return error.UnexpectedEof;
errdefer frame.deinit(std.testing.allocator);
try std.testing.expectEqual(expected, frame.type);
return frame;
}
test "relay carries request response credit and orderly half-closes over loopback" {
const pair = try testSocketPair();
defer std.posix.close(pair[1]);
var relay = Relay.init(std.testing.allocator);
defer relay.deinit();
try std.testing.expectEqual(Relay.Adopt.adopted, relay.adoptSocket(pair[0], .empty));
const ready = try testReadFrame(pair[1], .forward_ready);
defer ready.deinit(std.testing.allocator);
try std.testing.expectEqual(proto.forward_version, try proto.decodeForwardHello(ready.payload));
const listener = try testListener();
defer std.posix.close(listener.fd);
const open = proto.encodeForwardOpen(.{ .id = 7, .port = listener.port });
try testPush(&relay, .forward_open, &open);
var remote: ?std.posix.fd_t = null;
for (0..20) |_| {
try testPump(&relay, 20);
remote = std.posix.accept(listener.fd, null, null, std.posix.SOCK.CLOEXEC) catch |err| switch (err) {
error.WouldBlock => null,
else => return err,
};
if (remote != null and relay.peers[0].?.channels[0].?.connecting == false) break;
}
const tcp = remote orelse return error.AcceptTimedOut;
defer std.posix.close(tcp);
try std.testing.expect(!relay.peers[0].?.channels[0].?.connecting);
const opened = try testReadFrame(pair[1], .forward_open_result);
defer opened.deinit(std.testing.allocator);
try std.testing.expect((try proto.decodeForwardOpenResult(opened.payload)).ok);
const initial = try testReadFrame(pair[1], .forward_credit);
defer initial.deinit(std.testing.allocator);
try std.testing.expectEqual(proto.forward_initial_credit, (try proto.decodeForwardCredit(initial.payload)).amount);
const outbound_credit = proto.encodeForwardCredit(.{ .id = 7, .amount = proto.forward_initial_credit });
try testPush(&relay, .forward_credit, &outbound_credit);
var request: [proto.forward_id_len + 4]u8 = undefined;
std.mem.writeInt(u32, request[0..4], 7, .little);
@memcpy(request[4..], "ping");
try testPush(&relay, .forward_data, &request);
var got_request: [4]u8 = undefined;
try std.testing.expectEqual(@as(usize, 4), try std.posix.read(tcp, &got_request));
try std.testing.expectEqualStrings("ping", &got_request);
const replenished = try testReadFrame(pair[1], .forward_credit);
defer replenished.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(u32, 4), (try proto.decodeForwardCredit(replenished.payload)).amount);
try std.testing.expectEqual(@as(usize, 4), try std.posix.write(tcp, "pong"));
for (0..20) |_| {
try testPump(&relay, 20);
if (relay.peers[0].?.channels[0].?.send_credit == proto.forward_initial_credit - 4) break;
}
const response = try testReadFrame(pair[1], .forward_data);
defer response.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(u32, 7), try proto.decodeForwardId(response.payload));
try std.testing.expectEqualStrings("pong", response.payload[proto.forward_id_len..]);
const id = proto.encodeForwardId(7);
try testPush(&relay, .forward_half_close, &id);
var eof_byte: [1]u8 = undefined;
try std.testing.expectEqual(@as(usize, 0), try std.posix.read(tcp, &eof_byte));
try std.posix.shutdown(tcp, .send);
for (0..20) |_| {
try testPump(&relay, 20);
if (relay.peers[0].?.channels[0] == null) break;
}
try std.testing.expect(relay.peers[0].?.channels[0] == null);
const half = try testReadFrame(pair[1], .forward_half_close);
defer half.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(u32, 7), try proto.decodeForwardId(half.payload));
}
test "expired nonblocking connects refuse only their channels and release slots" {
const pair = try testSocketPair();
defer std.posix.close(pair[1]);
var relay = Relay.init(std.testing.allocator);
defer relay.deinit();
try std.testing.expectEqual(Relay.Adopt.adopted, relay.adoptSocket(pair[0], .empty));
const ready = try testReadFrame(pair[1], .forward_ready);
ready.deinit(std.testing.allocator);
const first = try testSocketPair();
defer std.posix.close(first[1]);
const second = try testSocketPair();
defer std.posix.close(second[1]);
const live = try testSocketPair();
defer std.posix.close(live[1]);
relay.peers[0].?.channels[0] = .{ .generation = relay.takeGeneration(), .id = 1, .fd = first[0], .connecting = true, .connect_by_ms = 10 };
relay.peers[0].?.channels[1] = .{ .generation = relay.takeGeneration(), .id = 2, .fd = second[0], .connecting = true, .connect_by_ms = 10 };
relay.peers[0].?.channels[2] = .{ .generation = relay.takeGeneration(), .id = 3, .fd = live[0], .connecting = false };
// `servicePoll`, rather than a direct helper call, is the production
// deadline path: it runs even when every descriptor is quiet. A stuck
// EINPROGRESS must therefore be reaped by a zero-event poll snapshot.
var fds: [poll_len]std.posix.pollfd = undefined;
relay.fillPoll(&fds);
relay.servicePoll(&fds);
try std.testing.expect(relay.peers[0].?.channels[0] == null);
try std.testing.expect(relay.peers[0].?.channels[1] == null);
try std.testing.expect(relay.peers[0].?.channels[2] != null);
// resetChannel owns the descriptor, not just the table entry. Leaving
// this open turns repeated timed-out opens into an fd leak.
// std.posix.fcntl deliberately treats EBADF as unreachable, so ask its
// raw syscall wrapper for this OS-level ownership assertion.
try std.testing.expectEqual(std.posix.E.BADF, std.posix.errno(std.posix.system.fcntl(first[0], std.posix.F.GETFD, @as(c_int, 0))));
try std.testing.expectEqual(std.posix.E.BADF, std.posix.errno(std.posix.system.fcntl(second[0], std.posix.F.GETFD, @as(c_int, 0))));
for ([_]u32{ 1, 2 }) |id| {
const refused = try testReadFrame(pair[1], .forward_open_result);
defer refused.deinit(std.testing.allocator);
const result = try proto.decodeForwardOpenResult(refused.payload);
try std.testing.expectEqual(id, result.id);
try std.testing.expect(!result.ok);
}
const reuse = try testSocketPair();
defer std.posix.close(reuse[1]);
relay.peers[0].?.channels[0] = .{ .generation = relay.takeGeneration(), .id = 4, .fd = reuse[0], .connecting = false };
try std.testing.expect(relay.peers[0].?.channels[0] != null);
}
test "channel flow violation resets only that channel and credit overflow drops the peer" {
const pair = try testSocketPair();
defer std.posix.close(pair[1]);
var relay = Relay.init(std.testing.allocator);
defer relay.deinit();
try std.testing.expectEqual(Relay.Adopt.adopted, relay.adoptSocket(pair[0], .empty));
const ready = try testReadFrame(pair[1], .forward_ready);
ready.deinit(std.testing.allocator);
const first = try testSocketPair();
defer std.posix.close(first[1]);
const second = try testSocketPair();
defer std.posix.close(second[1]);
relay.peers[0].?.channels[0] = .{ .generation = relay.takeGeneration(), .id = 1, .fd = first[0], .connecting = false, .recv_credit = 4 };
relay.peers[0].?.channels[1] = .{ .generation = relay.takeGeneration(), .id = 2, .fd = second[0], .connecting = false };
var excess: [proto.forward_id_len + 5]u8 = undefined;
std.mem.writeInt(u32, excess[0..4], 1, .little);
@memcpy(excess[4..], "12345");
try testPush(&relay, .forward_data, &excess);
try std.testing.expect(relay.peers[0] != null);
try std.testing.expect(relay.peers[0].?.channels[0] == null);
try std.testing.expect(relay.peers[0].?.channels[1] != null);
const reset = try testReadFrame(pair[1], .forward_reset);
defer reset.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(u32, 1), try proto.decodeForwardId(reset.payload));
const overflow = proto.encodeForwardCredit(.{ .id = 2, .amount = proto.forward_initial_credit + 1 });
try testPush(&relay, .forward_credit, &overflow);
try std.testing.expect(relay.peers[0] == null);
}
fn testNonblocking(fd: std.posix.fd_t) !void {
const flags = try std.posix.fcntl(fd, std.posix.F.GETFL, 0);
const bits: u32 = @bitCast(std.posix.O{ .NONBLOCK = true });
_ = try std.posix.fcntl(fd, std.posix.F.SETFL, flags | bits);
}
fn testFillSocket(fd: std.posix.fd_t) !usize {
var buf: [4096]u8 = @splat(0xa5);
var total: usize = 0;
while (true) total += std.posix.write(fd, &buf) catch |err| switch (err) {
error.WouldBlock => return total,
else => return err,
};
}
fn testDrainPrefix(relay: *Relay, fd: std.posix.fd_t, count: usize) !void {
var left = count;
var buf: [64 * 1024]u8 = undefined;
while (left != 0) {
relay.flushPeer(0);
const n = try std.posix.read(fd, buf[0..@min(buf.len, left)]);
if (n == 0) return error.UnexpectedEof;
left -= n;
}
}
test "concurrent remote channels pause at shared peer capacity and resume intact" {
const alloc = std.testing.allocator;
const transport_pair = try testSocketPair();
defer std.posix.close(transport_pair[1]);
try testNonblocking(transport_pair[0]);
var relay = Relay.init(alloc);
defer relay.deinit();
try std.testing.expectEqual(Relay.Adopt.adopted, relay.adoptSocket(transport_pair[0], .empty));
const ready = try testReadFrame(transport_pair[1], .forward_ready);
ready.deinit(alloc);
const kernel_fill = try testFillSocket(transport_pair[0]);
const frame_len = proto.frame_header_len + proto.forward_id_len + proto.forward_data_max;
const prefix_len = peer_queue_max - control_reserve - frame_len;
try relay.peers[0].?.pending.resize(alloc, prefix_len);
@memset(relay.peers[0].?.pending.items, 0x5a);
const channel_count = 6;
var channel_peers: [channel_count]std.posix.fd_t = undefined;
var made: usize = 0;
defer for (channel_peers[0..made]) |fd| std.posix.close(fd);
var payloads: [channel_count][proto.forward_data_max]u8 = undefined;
for (0..channel_count) |i| {
const pair = try testSocketPair();
channel_peers[i] = pair[1];
made += 1;
@memset(&payloads[i], @intCast(i + 1));
try proto.writeAllFd(pair[1], &payloads[i]);
relay.peers[0].?.channels[i] = .{
.generation = relay.takeGeneration(),
.id = @intCast(i + 1),
.fd = pair[0],
.connecting = false,
.send_credit = proto.forward_initial_credit,
};
}
var fds: [poll_len]std.posix.pollfd = undefined;
relay.fillPoll(&fds);
_ = try std.posix.poll(&fds, 1000);
relay.servicePoll(&fds);
try std.testing.expect(relay.peers[0] != null);
try std.testing.expectEqual(prefix_len + frame_len, relay.peers[0].?.pending.items.len);
try std.testing.expect(relay.peers[0].?.pending.items.len <= peer_queue_max - control_reserve);
for (0..channel_count) |i| try std.testing.expect(relay.peers[0].?.channels[i] != null);
try testDrainPrefix(&relay, transport_pair[1], kernel_fill + prefix_len);
relay.flushPeer(0);
var first = try testReadFrame(transport_pair[1], .forward_data);
defer first.deinit(alloc);
try std.testing.expectEqual(@as(u32, 1), try proto.decodeForwardId(first.payload));
try std.testing.expectEqualSlices(u8, &payloads[0], first.payload[proto.forward_id_len..]);
for (1..channel_count) |i| {
relay.readTcp(0, i);
try std.testing.expect(relay.peers[0] != null);
try std.testing.expect(relay.peers[0].?.pending.items.len + relay.peers[0].?.sink.inFlight() <= peer_queue_max - control_reserve);
relay.flushPeer(0);
var frame = try testReadFrame(transport_pair[1], .forward_data);
defer frame.deinit(alloc);
try std.testing.expectEqual(@as(u32, @intCast(i + 1)), try proto.decodeForwardId(frame.payload));
try std.testing.expectEqualSlices(u8, &payloads[i], frame.payload[proto.forward_id_len..]);
}
}