diff --git a/src/remote/endpoint.zig b/src/remote/endpoint.zig index 72d70b9..5fd6dea 100644 --- a/src/remote/endpoint.zig +++ b/src/remote/endpoint.zig @@ -61,10 +61,13 @@ const Endpoint = struct { } fn deinit(self: *@This()) void { - // Exit all live proxies (best-effort; endpoint may already be shutting down). + // Shut down all live proxies. Use "endpoint_exit" rather than "exit" + // so the proxy can distinguish this deinit-time message (which has no + // live sender context, i.e. from.instance_id()==0) from a Thespian + // trapped-exit sent by a linked local actor. var proxy_it = self.proxies.valueIterator(); while (proxy_it.next()) |p| { - p.send(.{ "exit", "transport_closed" }) catch {}; + p.send(.{ "endpoint_exit", "transport_closed" }) catch {}; p.deinit(); } self.proxies.deinit(); @@ -87,6 +90,7 @@ const Endpoint = struct { var to_name: []const u8 = ""; var payload: []const u8 = ""; var remote_id: u64 = 0; + var reason: []const u8 = ""; if (try m.match(.{ proc_tag, "stdout", tp.extract(&bytes) })) { if (self.accumulator.feed(bytes)) |frame| try self.dispatch_inbound(frame); @@ -99,6 +103,19 @@ const Endpoint = struct { } else if (try m.match(.{ "send", tp.extract(&from_id), tp.extract(&to_name), cbor.extract_cbor(&payload) })) { // Outbound send_named — from a local actor addressing a well-known remote actor by name. try self.send_wire_named(from_id, to_name, payload); + } else if (try m.match(.{ "link_wire", tp.extract(&from_id), tp.extract(&to_id) })) { + // A local proxy is establishing a remote link: from_id=local handle pointer, + // to_id=remote_id of the target on the remote system. + const wire_from_id = try self.get_or_assign_outbound_id(from_id); + try self.send_wire_link(wire_from_id, to_id); + } else if (try m.match(.{ "local_link_exit", tp.extract(&from_id), tp.extract(&reason) })) { + // A local actor that a proxy was Thespian-linked to has exited. + // from_id is the actor's handle pointer; look up its wire ID and + // forward an exit wire message so the remote side can propagate it. + const key: usize = @intCast(from_id); + if (self.outbound.get(key)) |wire_id| + try self.send_wire_exit(wire_id, reason); + // If not in the outbound table the actor never sent outbound; ignore. } else if (try m.match(.{ "proxy_exit", tp.extract(&remote_id), tp.any })) { // A local proxy has exited; remove it from the proxy table. if (self.proxies.fetchRemove(remote_id)) |entry| entry.value.deinit(); @@ -165,8 +182,17 @@ const Endpoint = struct { if (self.proxies.get(e.id)) |p| p.send(.{ "exit", e.reason }) catch {}; }, + .link => |lnk| { + // Remote side is establishing a link: lnk.local_id is the remote + // actor's wire ID (= our proxy key), lnk.remote_id is the local + // actor's wire ID in our local_actors table. + const prx = try self.get_or_create_proxy(lnk.local_id); + if (self.local_actors.getPtr(lnk.remote_id)) |actor_ptr| + try prx.send(.{ "set_notify", @as(u64, @intFromPtr(actor_ptr.h)) }) + else + return tp.exit_error(error.UnknownLocalActor, null); + }, .transport_error => |te| return tp.exit(te.reason), - .link => return tp.exit_error(error.UnexpectedMessage, null), } } @@ -181,6 +207,28 @@ const Endpoint = struct { try self.proc.send(frame_stream.buffered()); } + fn send_wire_link(self: *@This(), local_id: u64, remote_id: u64) !void { + var msg_buf: [framing.max_frame_size]u8 = undefined; + var msg_stream: std.Io.Writer = .fixed(&msg_buf); + try protocol.encode_link(&msg_stream, local_id, remote_id); + + var frame_buf: [framing.max_frame_size + 4]u8 = undefined; + var frame_stream: std.Io.Writer = .fixed(&frame_buf); + try framing.write_frame(&frame_stream, msg_stream.buffered()); + try self.proc.send(frame_stream.buffered()); + } + + fn send_wire_exit(self: *@This(), id: u64, reason: []const u8) !void { + var msg_buf: [framing.max_frame_size]u8 = undefined; + var msg_stream: std.Io.Writer = .fixed(&msg_buf); + try protocol.encode_exit(&msg_stream, id, reason); + + var frame_buf: [framing.max_frame_size + 4]u8 = undefined; + var frame_stream: std.Io.Writer = .fixed(&frame_buf); + try framing.write_frame(&frame_stream, msg_stream.buffered()); + try self.proc.send(frame_stream.buffered()); + } + fn send_wire_named(self: *@This(), from_id: u64, to_name: []const u8, payload: []const u8) !void { var msg_buf: [framing.max_frame_size]u8 = undefined; var msg_stream: std.Io.Writer = .fixed(&msg_buf); diff --git a/src/remote/proxy.zig b/src/remote/proxy.zig index 79fef76..bf8bbe8 100644 --- a/src/remote/proxy.zig +++ b/src/remote/proxy.zig @@ -8,6 +8,33 @@ /// sends {"exit", reason} to the proxy. Because the proxy traps exits, this /// arrives as a normal message, and the proxy propagates it by exiting with /// the same reason, which in turn signals any local actors linked to it. +/// +/// ## Link protocol +/// +/// To link a local actor A to remote actor B (proxy_B): +/// 1. A sends {"link"} to proxy_B. +/// 2. proxy_B calls from.link() (A ↔ proxy_B are now Thespian-linked), +/// caches A in senders, and sends {"link_wire", A_handle, B_remote_id} +/// to the endpoint. +/// 3. Endpoint assigns A a wire ID and sends ["link", A_wire_id, B_remote_id] +/// over the wire. +/// 4. Remote endpoint receives the link, creates/finds proxy_A, and sends +/// {"set_notify", B_handle} to proxy_A. +/// 5. proxy_A calls target.link() (proxy_A ↔ B are now Thespian-linked). +/// +/// When B exits: +/// - proxy_A gets an exit from B (from ≠ endpoint). +/// - proxy_A forwards {"local_link_exit", B_handle, reason} to its endpoint. +/// - Remote endpoint looks up B's wire ID and sends ["exit", B_wire_id, reason]. +/// - Local endpoint receives, finds proxy_B, sends {"exit", reason}. +/// - proxy_B exits; A gets an exit signal via the Thespian link. +/// +/// When A exits: +/// - proxy_B gets an exit from A (from ≠ endpoint). +/// - proxy_B forwards {"local_link_exit", A_handle, reason} to its endpoint. +/// - Local endpoint looks up A's wire ID and sends ["exit", A_wire_id, reason]. +/// - Remote endpoint receives, finds proxy_A, sends {"exit", reason}. +/// - proxy_A exits; B gets an exit signal via the Thespian link. const std = @import("std"); const tp = @import("thespian"); const protocol = @import("protocol"); @@ -44,7 +71,7 @@ const Proxy = struct { }; errdefer self.deinit(); // Trap exit signals so they arrive as ["exit", reason] messages, - // allowing us to propagate them to linked local actors. + // allowing us to distinguish remote-actor exits from local-actor exits. _ = tp.set_trap(true); tp.receive(&self.receiver); } @@ -67,12 +94,40 @@ const Proxy = struct { var handle_int: u64 = 0; var payload: []const u8 = ""; - if (try m.match(.{ "exit", tp.extract(&reason) })) { - // Notify the endpoint so it can remove us from the proxy table. - // Silently ignore failure — the endpoint may already be dead if - // this exit came from transport collapse. + if (try m.match(.{ "endpoint_exit", tp.extract(&reason) })) { + // Sent by the endpoint's deinit during teardown (no live sender + // context). Exit unconditionally; skip proxy_exit because the + // endpoint is already cleaning up its proxy table. + return tp.exit(reason); + } else if (try m.match(.{ "exit", tp.extract(&reason) })) { + if (from.instance_id() != self.endpoint.instance_id()) { + // A local actor that we are linked to (via "link" or "set_notify") + // has exited. Forward the exit wire message so the remote side can + // propagate it to the actor that initiated the link. Do not exit + // this proxy — it may still serve other remote messages. + self.endpoint.send(.{ "local_link_exit", @as(u64, @intFromPtr(from.h)), reason }) catch {}; + return; + } + // Exit came from the live endpoint: a remote actor exited and the + // endpoint dispatched the wire exit to us. Notify the endpoint so it + // can remove us from the proxy table, then exit. self.endpoint.send(.{ "proxy_exit", self.remote_id, reason }) catch {}; return tp.exit(reason); + } else if (try m.match(.{"link"})) { + // A local actor is requesting a remote link via this proxy. + // Establish a Thespian link so that when either side exits the other + // is notified, then ask the endpoint to send the wire "link" message. + try from.link(); + const actor_id = from.instance_id(); + const gop = try self.senders.getOrPut(actor_id); + if (!gop.found_existing) gop.value_ptr.* = from.clone(); + const stored = gop.value_ptr.*; + try self.endpoint.send(.{ "link_wire", @as(u64, @intFromPtr(stored.h)), self.remote_id }); + } else if (try m.match(.{ "set_notify", tp.extract(&handle_int) })) { + // The endpoint has received a wire "link" and is asking this proxy to + // monitor a local actor. Link proxy to the actor so its exit is trapped. + const target: tp.pid_ref = .{ .h = @ptrFromInt(@as(usize, @intCast(handle_int))) }; + try target.link(); } else if (try m.match(.{ "deliver_named", tp.extract(&to_name), tp.extract_cbor(&payload) })) { // Deliver to a named local actor. Because this call is made from // within the proxy's receive, the recipient sees the proxy as `from`. @@ -88,9 +143,9 @@ const Proxy = struct { // We cache one owned clone per sender (keyed by stable instance_id) // so the endpoint can safely clone from the stored heap handle later. const actor_id = from.instance_id(); - const result = try self.senders.getOrPut(actor_id); - if (!result.found_existing) result.value_ptr.* = from.clone(); - const stored = result.value_ptr.*; + const gop = try self.senders.getOrPut(actor_id); + if (!gop.found_existing) gop.value_ptr.* = from.clone(); + const stored = gop.value_ptr.*; try self.endpoint.send(.{ "send", @as(u64, @intFromPtr(stored.h)), self.remote_id, protocol.RawCbor{ .bytes = m.buf } }); } } diff --git a/test/remote_child_endpoint.zig b/test/remote_child_endpoint.zig index 5fbfe22..77ecfae 100644 --- a/test/remote_child_endpoint.zig +++ b/test/remote_child_endpoint.zig @@ -259,7 +259,13 @@ const StdioEndpoint = struct { return tp.exit_error(error.UnknownWireId, null); }, .transport_error => |te| return tp.exit(te.reason), - .link, .exit => return tp.exit_error(error.UnexpectedMessage, null), + .exit => |e| { + // Remote side notifying that a linked actor has exited. + // Forward to the local actor with that wire ID (best-effort). + if (self.wire_ids.get(e.id)) |actor| + actor.send(.{ "exit", e.reason }) catch {}; + }, + .link => return tp.exit_error(error.LinkNotSupported, null), } }