diff --git a/src/remote/endpoint.zig b/src/remote/endpoint.zig index af5dc60..9bb882e 100644 --- a/src/remote/endpoint.zig +++ b/src/remote/endpoint.zig @@ -1,10 +1,12 @@ /// Parent-side endpoint actor. /// /// Wraps a subprocess, accumulates framed CBOR from its stdout, and dispatches -/// decoded send_named messages to named local actors via the env proc table. +/// decoded messages to local actors or proxies. /// -/// Local actors send outbound messages to the endpoint with: -/// {"send", from_id: u64, to_name: text, payload: raw_cbor} +/// Outbound messages (local → endpoint): +/// {"send", from_id: u64, to_id: u64, payload} — via proxy, by remote ID +/// {"send", from_id: u64, to_name: text, payload} — direct by name +/// {"proxy_exit", remote_id: u64, reason: text} — proxy notifying of exit const std = @import("std"); const tp = @import("thespian"); const cbor = @import("cbor"); @@ -27,6 +29,8 @@ const Endpoint = struct { allocator: std.mem.Allocator, proc: subprocess, accumulator: framing.Accumulator, + /// Inbound proxy table: remote actor ID → local proxy pid. + proxies: std.AutoHashMap(u64, tp.pid), receiver: tp.Receiver(*@This()), fn start(args: Args) tp.result { @@ -41,6 +45,7 @@ const Endpoint = struct { .allocator = args.allocator, .proc = proc, .accumulator = .{}, + .proxies = std.AutoHashMap(u64, tp.pid).init(args.allocator), .receiver = .init(receive_fn, deinit, self), }; errdefer self.deinit(); @@ -48,6 +53,13 @@ const Endpoint = struct { } fn deinit(self: *@This()) void { + // Exit all live proxies (best-effort; endpoint may already be shutting down). + var it = self.proxies.valueIterator(); + while (it.next()) |p| { + p.send(.{ "exit", "transport_closed" }) catch {}; + p.deinit(); + } + self.proxies.deinit(); self.proc.deinit(); self.allocator.destroy(self); } @@ -62,6 +74,8 @@ const Endpoint = struct { var to_id: u64 = 0; var to_name: []const u8 = ""; var payload: []const u8 = ""; + var remote_id: u64 = 0; + if (try m.match(.{ proc_tag, "stdout", tp.extract(&bytes) })) { if (self.accumulator.feed(bytes)) |frame| try self.dispatch_inbound(frame); } else if (try m.match(.{ proc_tag, "stderr", tp.any })) { @@ -71,10 +85,9 @@ 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(.{ "proxy_exit", tp.any, tp.any })) { - // A local proxy has exited; remove it from the proxy table (stub until table is implemented). - } else if (try m.match(.{ proc_tag, "term", "exited", tp.any })) { - return tp.exit("transport_closed"); + } 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(); } else if (try m.match(.{ proc_tag, "term", tp.any, tp.any })) { return tp.exit("transport_closed"); } else { @@ -82,18 +95,42 @@ const Endpoint = struct { } } - fn dispatch_inbound(_: *@This(), frame: []const u8) !void { + /// Get the local proxy for remote_id, creating one if it doesn't exist yet. + fn get_or_create_proxy(self: *@This(), remote_id: u64) !tp.pid_ref { + if (self.proxies.getPtr(remote_id)) |p| return p.ref(); + const p = try tp.spawn(self.allocator, proxy.Args{ + .allocator = self.allocator, + .endpoint = tp.self_pid().clone(), + .remote_id = remote_id, + }, proxy.start, "proxy"); + try self.proxies.put(remote_id, p); + return self.proxies.getPtr(remote_id).?.ref(); + } + + fn dispatch_inbound(self: *@This(), frame: []const u8) !void { const msg = try protocol.decode(frame); switch (msg) { .send => |s| { - _ = s; - return tp.exit_error(error.UnexpectedSend, null); + // Ensure a local proxy exists for the sender. + _ = try self.get_or_create_proxy(s.from_id); + // Deliver payload to the local actor identified by to_id. + // Requires the outbound ID table (not yet implemented); error for now. + _ = s.to_id; + _ = s.payload; + return tp.exit_error(error.OutboundTableNotImplemented, null); }, .send_named => |s| { + // Ensure a local proxy exists for the sender so it can receive replies. + _ = try self.get_or_create_proxy(s.from_id); const actor = tp.env.get().proc(s.to_name); try actor.send_raw(tp.message{ .buf = s.payload }); }, - .link, .exit, .proxy_id, .transport_error => return tp.exit_error(error.UnexpectedMessage, null), + .exit => |e| { + // Remote actor has exited; notify its local proxy. + if (self.proxies.get(e.id)) |p| + p.send(.{ "exit", e.reason }) catch {}; + }, + .proxy_id, .link, .transport_error => return tp.exit_error(error.UnexpectedMessage, null), } }