diff --git a/CHANGELOG.md b/CHANGELOG.md index 0d6d628..3a9f195 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -68,6 +68,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **A Call pipelined on an already-failed answer never received a Return.** + Found on the wire by the cross-impl L3 lane (`just e2e-l3-vatc`, + pipelined-provide scenarios): the C++ recipient pipelines its Call on the + Accept question without waiting, the host refuses the Accept + (`CrossPeerReceiverHostedTargetUnsupported`), and the Call — arriving after + the refusal already went out — parked in `pending_promises` forever. The + queued-call drain in `sendReturnException` only reaches calls queued + *before* the exception Return; results Returns are recorded in + `resolved_answers` precisely so late pipelined calls can resolve, but + exception Returns were recorded nowhere, so a late call on a failed answer + was indistinguishable from one on a still-pending answer. The C++ recipient + hung on the spec's exactly-one-Return-per-call guarantee. Not Accept-specific: + any answer failed synchronously during dispatch had the same window. + + Exception Returns are now recorded in a `failed_answers` map (reason + type, + kept until Finish, the exact lifecycle of `resolved_answers`), and a + promised-target call that would otherwise queue is answered immediately with + a **copy of the recorded exception**, preserving the retryability signal — + the broken-pipeline behavior of the C++ reference. The record is written in + the one funnel every exception Return passes through, so transitive + pipelines (a call pipelined on a failed pipelined call) are covered, and the + drain path is unchanged. Ablation-proven unit tests pin both the refused- + Accept shape and the plain two-party shape; the e2e driver now observes the + refusal *through* a pipelined call instead of working around the gap with + `whenResolved()`. + - **`MessageBuilder.writeTo` and `writePackedTo` were neither type-checked nor tested.** Both are frozen Stable API with **zero call sites in the tree**, and Zig does not analyse an uninstantiated generic body — so making them concrete diff --git a/docs/api-snapshot-experimental.txt b/docs/api-snapshot-experimental.txt index 2a347e0..ff8a5f2 100644 --- a/docs/api-snapshot-experimental.txt +++ b/docs/api-snapshot-experimental.txt @@ -977,6 +977,7 @@ capnpc-zig.rpc.peer.Peer.enableRuntimeThreadChecks: fn (*rpc.peer.mod.Peer, bool capnpc-zig.rpc.peer.Peer.ensureExportAt: fn (*rpc.peer.mod.Peer, u32, rpc.peer.mod.Export) error{CapTableFull,OutOfMemory}!bool capnpc-zig.rpc.peer.Peer.entropy: field ?rpc.peer.mod.EntropySource = null capnpc-zig.rpc.peer.Peer.exports: field hash_map.HashMap(u32,rpc.peer.state.ExportEntry(rpc.peer.mod.Export),hash_map.AutoContext(u32),80) +capnpc-zig.rpc.peer.Peer.failed_answers: field hash_map.HashMap(u32,rpc.peer.state.FailedAnswer,hash_map.AutoContext(u32),80) capnpc-zig.rpc.peer.Peer.finished_early_answers: field hash_map.HashMap(u32,bool,hash_map.AutoContext(u32),80) capnpc-zig.rpc.peer.Peer.forgetImportRefsForHost: fn (*rpc.peer.mod.Peer, u32, u32) anyerror!void capnpc-zig.rpc.peer.Peer.forwarded_questions: field hash_map.HashMap(u32,u32,hash_map.AutoContext(u32),80) @@ -1400,6 +1401,9 @@ capnpc-zig.rpc.peer.resolve: struct capnpc-zig.rpc.peer.seedEntropyCsprng: fn (Io) error{Canceled,EntropyUnavailable}!Random.ChaCha capnpc-zig.rpc.peer.shutdown_reason: const *const [18:0]u8 capnpc-zig.rpc.peer.state.ExportEntry: fn (comptime type) type +capnpc-zig.rpc.peer.state.FailedAnswer.ex_type: field rpc.wire.protocol.ExceptionType +capnpc-zig.rpc.peer.state.FailedAnswer.reason: field []u8 +capnpc-zig.rpc.peer.state.FailedAnswer: struct capnpc-zig.rpc.peer.state.JoinKeyPart.join_id: field u32 capnpc-zig.rpc.peer.state.JoinKeyPart.part_count: field u16 capnpc-zig.rpc.peer.state.JoinKeyPart.part_num: field u16 diff --git a/src/rpc/peer/call/peer_call_orchestration.zig b/src/rpc/peer/call/peer_call_orchestration.zig index 35569c1..45e1f70 100644 --- a/src/rpc/peer/call/peer_call_orchestration.zig +++ b/src/rpc/peer/call/peer_call_orchestration.zig @@ -286,9 +286,11 @@ pub fn handleCallPromisedTargetForPeer( promised: protocol.PromisedAnswer, resolve_promised_answer: *const fn (*PeerType, protocol.PromisedAnswer) anyerror!cap_table.ResolvedCap, has_unresolved_promise_export: *const fn (*PeerType, u32) bool, + lookup_failed_answer: *const fn (*PeerType, u32) ?peer_call_targets.FailedAnswerView, queue_promised_call: *const fn (*PeerType, u32, []const u8, InboundCapsType) anyerror!void, queue_promise_export_call: *const fn (*PeerType, u32, []const u8, InboundCapsType) anyerror!void, send_return_exception: *const fn (*PeerType, u32, []const u8) anyerror!void, + send_return_exception_typed: *const fn (*PeerType, u32, []const u8, protocol.ExceptionType) anyerror!void, handle_resolved_call: *const fn (*PeerType, protocol.Call, *const InboundCapsType, cap_table.ResolvedCap) anyerror!void, release_inbound_caps: *const fn (*PeerType, *InboundCapsType) anyerror!void, report_nonfatal_error: *const fn (*PeerType, anyerror) void, @@ -311,6 +313,7 @@ pub fn handleCallPromisedTargetForPeer( promised, resolve_promised_answer, has_unresolved_promise_export, + lookup_failed_answer, ); switch (target_plan) { @@ -335,6 +338,16 @@ pub fn handleCallPromisedTargetForPeer( try send_return_exception(peer, call.question_id, @errorName(err)); return; }, + .fail_broken_answer => |failed| { + // Pipelined on an answer that already returned an exception. The + // failure drain (`failQueuedPromisedCalls`) only reaches calls + // queued BEFORE the exception Return; this one arrived after, so + // fail it the same way — a copy of the answer's own exception, + // preserving the retryability signal. + release_caps = true; + try send_return_exception_typed(peer, call.question_id, failed.reason, failed.ex_type); + return; + }, .handle_resolved => |resolved| { release_caps = true; try handle_resolved_call(peer, call, &inbound_caps, resolved); @@ -347,9 +360,11 @@ pub fn handleCallPromisedTargetForPeerFn( comptime InboundCapsType: type, comptime resolve_promised_answer: *const fn (*PeerType, protocol.PromisedAnswer) anyerror!cap_table.ResolvedCap, comptime has_unresolved_promise_export: *const fn (*PeerType, u32) bool, + comptime lookup_failed_answer: *const fn (*PeerType, u32) ?peer_call_targets.FailedAnswerView, comptime queue_promised_call: *const fn (*PeerType, u32, []const u8, InboundCapsType) anyerror!void, comptime queue_promise_export_call: *const fn (*PeerType, u32, []const u8, InboundCapsType) anyerror!void, comptime send_return_exception: *const fn (*PeerType, u32, []const u8) anyerror!void, + comptime send_return_exception_typed: *const fn (*PeerType, u32, []const u8, protocol.ExceptionType) anyerror!void, comptime handle_resolved_call: *const fn (*PeerType, protocol.Call, *const InboundCapsType, cap_table.ResolvedCap) anyerror!void, comptime release_inbound_caps: *const fn (*PeerType, *InboundCapsType) anyerror!void, comptime report_nonfatal_error: *const fn (*PeerType, anyerror) void, @@ -365,9 +380,11 @@ pub fn handleCallPromisedTargetForPeerFn( promised, resolve_promised_answer, has_unresolved_promise_export, + lookup_failed_answer, queue_promised_call, queue_promise_export_call, send_return_exception, + send_return_exception_typed, handle_resolved_call, release_inbound_caps, report_nonfatal_error, diff --git a/src/rpc/peer/call/peer_call_targets.zig b/src/rpc/peer/call/peer_call_targets.zig index 20641c1..5c369dc 100644 --- a/src/rpc/peer/call/peer_call_targets.zig +++ b/src/rpc/peer/call/peer_call_targets.zig @@ -26,11 +26,22 @@ pub fn planImportedTarget( return if (has_handler) .call_handler else .missing_export_handler; } +/// Borrowed view of a recorded failed answer's exception (see +/// `state.FailedAnswer`): valid until the record is removed at Finish. +pub const FailedAnswerView = struct { + reason: []const u8, + ex_type: protocol.ExceptionType, +}; + pub const PromisedTargetPlan = union(enum) { queue_promised_call, queue_export_promise: u32, handle_resolved: cap_table.ResolvedCap, send_exception: anyerror, + /// The target answer already returned an exception: answer the call with + /// a copy of that exception. Queueing would wedge it forever — a failed + /// answer never replays its queue. + fail_broken_answer: FailedAnswerView, }; pub fn planPromisedTarget( @@ -39,9 +50,19 @@ pub fn planPromisedTarget( promised: protocol.PromisedAnswer, resolve_promised_answer: *const fn (*PeerType, protocol.PromisedAnswer) anyerror!cap_table.ResolvedCap, has_unresolved_promise_export: *const fn (*PeerType, u32) bool, + lookup_failed_answer: *const fn (*PeerType, u32) ?FailedAnswerView, ) PromisedTargetPlan { const resolved = resolve_promised_answer(peer, promised) catch |err| { - if (err == error.PromiseUnresolved) return .queue_promised_call; + if (err == error.PromiseUnresolved) { + // `resolved_answers` records results Returns only. An answer that + // already FAILED misses there identically to one still pending — + // distinguish via the failed-answer record, or the call queues + // against a Return that will never come. + if (lookup_failed_answer(peer, promised.question_id)) |failed| { + return .{ .fail_broken_answer = failed }; + } + return .queue_promised_call; + } return .{ .send_exception = err }; }; @@ -108,10 +129,11 @@ test "peer_call_targets imported target planning covers all branches" { ); } -test "peer_call_targets promised target planning handles unresolved, exception, queue-export and resolved" { +test "peer_call_targets promised target planning handles unresolved, failed-answer, exception, queue-export and resolved" { const FakePeer = struct { mode: enum { unresolved, + failed_answer, failure, exported_unresolved, exported_resolved, @@ -123,7 +145,7 @@ test "peer_call_targets promised target planning handles unresolved, exception, fn resolvePromisedAnswer(peer: *FakePeer, promised: protocol.PromisedAnswer) !cap_table.ResolvedCap { _ = promised; return switch (peer.mode) { - .unresolved => error.PromiseUnresolved, + .unresolved, .failed_answer => error.PromiseUnresolved, .failure => error.TestExpectedError, .exported_unresolved, .exported_resolved => .{ .exported = .{ .id = 9 } }, .imported_resolved => .{ .imported = .{ .id = 11 } }, @@ -133,6 +155,11 @@ test "peer_call_targets promised target planning handles unresolved, exception, fn hasUnresolvedPromiseExport(peer: *FakePeer, export_id: u32) bool { return peer.mode == .exported_unresolved and export_id == 9; } + + fn lookupFailedAnswer(peer: *FakePeer, question_id: u32) ?FailedAnswerView { + if (peer.mode != .failed_answer or question_id != 1) return null; + return .{ .reason = "broken", .ex_type = .overloaded }; + } }; // Neither the planner nor the fake hooks above inspect the transform, so @@ -153,10 +180,33 @@ test "peer_call_targets promised target planning handles unresolved, exception, promised, Hooks.resolvePromisedAnswer, Hooks.hasUnresolvedPromiseExport, + Hooks.lookupFailedAnswer, ); try std.testing.expectEqual(PromisedTargetPlan.queue_promised_call, plan); } + { + // Same PromiseUnresolved from the resolver, but the answer is on + // record as FAILED: the plan must carry the recorded exception, not + // queue the call behind a Return that will never come. + var peer = FakePeer{ .mode = .failed_answer }; + const plan = planPromisedTarget( + FakePeer, + &peer, + promised, + Hooks.resolvePromisedAnswer, + Hooks.hasUnresolvedPromiseExport, + Hooks.lookupFailedAnswer, + ); + switch (plan) { + .fail_broken_answer => |failed| { + try std.testing.expectEqualStrings("broken", failed.reason); + try std.testing.expectEqual(protocol.ExceptionType.overloaded, failed.ex_type); + }, + else => return error.TestExpectedEqual, + } + } + { var peer = FakePeer{ .mode = .failure }; const plan = planPromisedTarget( @@ -165,6 +215,7 @@ test "peer_call_targets promised target planning handles unresolved, exception, promised, Hooks.resolvePromisedAnswer, Hooks.hasUnresolvedPromiseExport, + Hooks.lookupFailedAnswer, ); switch (plan) { .send_exception => |err| try std.testing.expectEqual(error.TestExpectedError, err), @@ -180,6 +231,7 @@ test "peer_call_targets promised target planning handles unresolved, exception, promised, Hooks.resolvePromisedAnswer, Hooks.hasUnresolvedPromiseExport, + Hooks.lookupFailedAnswer, ); switch (plan) { .queue_export_promise => |export_id| try std.testing.expectEqual(@as(u32, 9), export_id), @@ -195,6 +247,7 @@ test "peer_call_targets promised target planning handles unresolved, exception, promised, Hooks.resolvePromisedAnswer, Hooks.hasUnresolvedPromiseExport, + Hooks.lookupFailedAnswer, ); switch (plan) { .handle_resolved => |cap| switch (cap) { @@ -213,6 +266,7 @@ test "peer_call_targets promised target planning handles unresolved, exception, promised, Hooks.resolvePromisedAnswer, Hooks.hasUnresolvedPromiseExport, + Hooks.lookupFailedAnswer, ); switch (plan) { .handle_resolved => |cap| switch (cap) { diff --git a/src/rpc/peer/mod.zig b/src/rpc/peer/mod.zig index 777068e..80b5309 100644 --- a/src/rpc/peer/mod.zig +++ b/src/rpc/peer/mod.zig @@ -173,6 +173,7 @@ const PersistenceState = struct { const ExportEntry = state.ExportEntry(Export); const ResolvedAnswer = state.ResolvedAnswer; +const FailedAnswer = state.FailedAnswer; const PendingCall = state.PendingCall; const ProvideTarget = state.ProvideTarget; @@ -519,6 +520,7 @@ fn msToNs(ms: u64) i64 { /// | `questions` | question ID | `Question` | Outstanding outbound calls awaiting a Return. Removed when the Return arrives. | /// | `question_param_export_refs` | question ID | export IDs (list) | Wire refs the question's Call params took on local exports (one per emitted senderHosted/senderPromise descriptor). Spent when the Return arrives with `releaseParamCaps = true` (the rpc.capnp default), dropped when it is false (the remote sends explicit Releases), freed unspent when the question dies without a wire Return. | /// | `resolved_answers` | question ID | `ResolvedAnswer` | Cached Return frames for answered questions (used to resolve PromisedAnswer references). Holds one answer-held reference per results export so pipeline targets survive an early Release. Removed on Finish, releasing those references. | +/// | `failed_answers` | question ID | `FailedAnswer` | The exception already returned for a failed answer (results-only `resolved_answers` never records it). Lets a call pipelined on that answer arriving after the Return be failed with a copy of the same exception instead of queueing forever. Removed on Finish. | /// | `pending_promises` | question ID | `ArrayList(PendingCall)` | Calls targeting a PromisedAnswer whose Return has not yet arrived. Replayed once the answer resolves. | /// | `pending_export_promises` | export ID | `ArrayList(PendingCall)` | Calls targeting a promise export not yet resolved. Replayed on `resolvePromiseExportToExport`. | /// | `forwarded_questions` | original answer ID | forwarded question ID | Maps an inbound call's answer ID to the question ID of the forwarded outbound call. | @@ -1410,6 +1412,15 @@ pub const Peer = struct { /// resolved answer, it commits first to drain queued promised calls, then /// immediately removes the recorded answer through normal Finish cleanup. finished_early_answers: std.AutoHashMap(u32, bool), + /// Inbound answers that already returned an EXCEPTION, kept until Finish + /// (the mirror of `resolved_answers`, which records results only). A call + /// pipelined on such an answer that arrives AFTER the exception Return + /// would otherwise queue in `pending_promises` forever — the failed + /// answer can never replay it; this record answers it with a copy of the + /// same exception instead. Bounded by max_active_inbound_questions, + /// best-effort under pressure (a skipped record degrades to the old + /// queue-forever behavior for that answer, never a crash). + failed_answers: std.AutoHashMap(u32, FailedAnswer), // -- Promise queueing --------------------------------------------------- @@ -1726,6 +1737,7 @@ pub const Peer = struct { .active_inbound_questions = std.AutoHashMap(u32, void).init(allocator), .resolving_answers = std.AutoHashMap(u32, void).init(allocator), .finished_early_answers = std.AutoHashMap(u32, bool).init(allocator), + .failed_answers = std.AutoHashMap(u32, FailedAnswer).init(allocator), .pending_promises = std.AutoHashMap(u32, std.ArrayList(PendingCall)).init(allocator), .pending_export_promises = std.AutoHashMap(u32, std.ArrayList(PendingCall)).init(allocator), .forwarded_questions = std.AutoHashMap(u32, u32).init(allocator), @@ -2111,6 +2123,11 @@ pub const Peer = struct { self.active_inbound_questions.deinit(); self.resolving_answers.deinit(); self.finished_early_answers.deinit(); + { + var f_it = self.failed_answers.valueIterator(); + while (f_it.next()) |failed| self.allocator.free(failed.reason); + } + self.failed_answers.deinit(); { var p_it = self.persistent_exports.valueIterator(); while (p_it.next()) |st| self.allocator.destroy(st.*); @@ -5215,6 +5232,9 @@ pub const Peer = struct { reason: []const u8, ex_type: protocol.ExceptionType, ) !void { + // Capture before delivery consumes the loopback marker (see + // sendReturnResults). + const is_loopback = self.loopback_questions.contains(answer_id); try peer_return_dispatch.sendReturnExceptionForPeer( Peer, self, @@ -5224,7 +5244,48 @@ pub const Peer = struct { clearSendResultsRouting, sendReturnFrameWithLoopback, ); - _ = self.finished_early_answers.remove(answer_id); + const finished_early = self.finished_early_answers.remove(answer_id); + // Record AFTER a successful send, mirroring resolved_answers' gates: + // never for loopback answers (no Finish will clear the record and the + // id would be poisoned for legal reuse) and never for finished-early + // answers (the Finish already arrived — same poisoned-reuse hazard, + // and no compliant pipelined call can follow it). This is what every + // exception Return funnels through, so calls failed by the queued + // drain below get their own record too — a late call pipelined on a + // FAILED pipelined call still finds the failure, transitively. + if (!is_loopback and !finished_early) { + self.recordFailedAnswer(answer_id, reason, ex_type); + } + } + + /// Best-effort record of an exception Return for `answer_id` (see the + /// `failed_answers` field doc). Keeps the FIRST exception if somehow + /// recorded twice; bounded like `finished_early_answers`; a skip under + /// budget or OOM degrades to the pre-record behavior for that answer + /// (late pipelined calls queue until their own Finish), never a crash. + fn recordFailedAnswer( + self: *Peer, + answer_id: u32, + reason: []const u8, + ex_type: protocol.ExceptionType, + ) void { + if (self.failed_answers.contains(answer_id)) return; + if (self.failed_answers.count() >= self.limits.max_active_inbound_questions) return; + const owned = self.allocator.dupe(u8, reason) catch |err| { + return self.reportNonfatalError(err); + }; + self.failed_answers.put(answer_id, .{ .reason = owned, .ex_type = ex_type }) catch |err| { + self.allocator.free(owned); + self.reportNonfatalError(err); + }; + } + + /// Planner hook (`planPromisedTarget`): the recorded exception for an + /// already-failed inbound answer, if any. Borrowed view — valid until the + /// record is removed at Finish. + fn lookupFailedAnswer(self: *Peer, answer_id: u32) ?peer_call_targets.FailedAnswerView { + const failed = self.failed_answers.get(answer_id) orelse return null; + return .{ .reason = failed.reason, .ex_type = failed.ex_type }; } /// Fail and drain every pipelined call queued against `answer_id` (and, @@ -7922,6 +7983,11 @@ pub const Peer = struct { try self.detachProvisionForFinish(qid); const was_active = self.active_inbound_questions.remove(qid); const was_resolving = self.resolving_answers.contains(qid); + // The failed-answer record lives exactly as long as resolved_answers + // entries do: until the remote finishes the question. + if (self.failed_answers.fetchRemove(qid)) |failed| { + self.allocator.free(failed.value.reason); + } self.clearPendingJoinResultAnswer(qid); try self.clearPendingJoinRelay(qid, true, finish_msg.release_result_caps); // Cancellation race: a Finish for an in-flight inbound call (still @@ -9546,8 +9612,9 @@ pub const Peer = struct { } /// True when `question_id` is already consumed by an inbound Call or - /// Bootstrap answer (active, resolved, resolving, finished-early, forwarded, - /// or queued for a promised target). This is the shared answer namespace; + /// Bootstrap answer (active, resolved, failed, resolving, finished-early, + /// forwarded, or queued for a promised target). This is the shared answer + /// namespace; /// it deliberately excludes the provide/join question tables so their /// handlers can report their own specific errors for same-type collisions. /// @@ -9560,6 +9627,7 @@ pub const Peer = struct { /// remote-forceable premature export release. fn inboundAnswerQuestionIdInUse(self: *Peer, question_id: u32) !bool { return self.resolved_answers.contains(question_id) or + self.failed_answers.contains(question_id) or self.active_inbound_questions.contains(question_id) or self.resolving_answers.contains(question_id) or self.finished_early_answers.contains(question_id) or @@ -9648,9 +9716,11 @@ pub const Peer = struct { cap_table.InboundCapTable, Peer.resolvePromisedAnswer, peer_call_targets.hasUnresolvedPromiseExportForPeerFn(Peer), + Peer.lookupFailedAnswer, Peer.queuePromisedCall, Peer.queuePromiseExportCall, Peer.sendReturnException, + Peer.sendReturnExceptionTyped, Peer.handleResolvedCall, Peer.releaseInboundCaps, peer_return_dispatch.reportNonfatalErrorForPeerFn(Peer), diff --git a/src/rpc/peer/state.zig b/src/rpc/peer/state.zig index ae24c12..b4ece84 100644 --- a/src/rpc/peer/state.zig +++ b/src/rpc/peer/state.zig @@ -2,6 +2,7 @@ const std = @import("std"); const builtin = @import("builtin"); const state_types = @import("./peer_state_types.zig"); const cap_table = @import("../caps/table.zig"); +const protocol = @import("../wire/protocol.zig"); /// Tunable hard limits for peer-owned protocol state. /// @@ -120,6 +121,19 @@ pub const ResolvedAnswer = struct { frame: []u8, }; +/// The exception a Return already delivered for an inbound answer, kept until +/// that answer's Finish. Only results Returns are recorded in +/// `resolved_answers`, so without this record a call pipelined on a FAILED +/// answer that arrives after its exception Return would queue in +/// `pending_promises` forever — the failed answer can never replay it. The +/// record lets the peer answer such a call with a copy of the same exception, +/// the broken-pipeline behavior of the C++ reference. +pub const FailedAnswer = struct { + /// Owned copy of the exception reason. + reason: []u8, + ex_type: protocol.ExceptionType, +}; + pub const PendingCall = struct { frame: []u8, caps: cap_table.InboundCapTable, diff --git a/tests/e2e/cpp/l3_vatc_client.cpp b/tests/e2e/cpp/l3_vatc_client.cpp index 15fa89c..6b9f387 100644 --- a/tests/e2e/cpp/l3_vatc_client.cpp +++ b/tests/e2e/cpp/l3_vatc_client.cpp @@ -606,14 +606,21 @@ int runDriver(kj::StringPtr host, kj::StringPtr port, kj::StringPtr scenario) { } tap.ok(true, "A received the introduced third-party cap from B"); - // First use forces the lazy Accept; with the lift the host must SERVE - // it. whenResolved() settles on the Accept's own Return (the - // unknown-token pattern), so a refusal would surface here as a thrown - // exception. + // First use forces the lazy Accept, which the lift now SERVES. + // + // The probe is deliberately a PIPELINED call, not whenResolved(): rpc.c++ + // sends this Call on the Accept question's promisedAnswer without waiting + // for the Accept's own Return, so it exercises the broken-pipeline rule as + // well as the serve. That matters historically — this exact wait HUNG + // FOREVER when the host dropped calls pipelined on an already-failed + // answer, and the fix that closed it keeps its cross-impl teeth here in the + // answered direction. The FAILED-answer direction of that rule no longer + // has a cross-impl cell (the lift turned this scenario's refusal into a + // success); it is covered by the unit tests that shipped with the fix. bool threw = false; kj::String desc = kj::str(""); KJ_IF_SOME(e, kj::runCatchingExceptions([&]() { - accepted.whenResolved().wait(ws); + accepted.getNumberRequest().send().wait(ws); })) { threw = true; desc = kj::str(e.getDescription()); @@ -626,7 +633,11 @@ int runDriver(kj::StringPtr host, kj::StringPtr port, kj::StringPtr scenario) { // 43, not host-Carol's 42. uint32_t n = accepted.getNumberRequest().send().wait(ws).getN(); tap.ok(n == 43, "call on the accepted cap returned 43 (B's local cap)"); - tap.ok(localNum.calls == 1, "B's local capability was invoked exactly once"); + // TWO invocations, not one: the probe above is itself a real pipelined + // call, not a whenResolved(). Both must land on B's cap — the first + // proves a call pipelined on the Accept question is answered, the second + // that the settled capability still routes to the same place. + tap.ok(localNum.calls == 2, "B's local capability was invoked by both the pipelined probe and the settled call"); // Release ceremony (the `happy` pattern): drop A's only ref to the // accepted cap and turn the kj loop so the queued Release actually diff --git a/tests/rpc/peer/rpc_peer_from_peer_zig_test.zig b/tests/rpc/peer/rpc_peer_from_peer_zig_test.zig index bd2c4e3..676d427 100644 --- a/tests/rpc/peer/rpc_peer_from_peer_zig_test.zig +++ b/tests/rpc/peer/rpc_peer_from_peer_zig_test.zig @@ -3917,6 +3917,122 @@ test "bootstrap return is recorded for promisedAnswer pipelined calls" { try std.testing.expectEqual(protocol.ReturnTag.exception, ret.tag); } +// Exception Returns are never recorded in resolved_answers, so before the +// failed_answers record a call pipelined on an already-FAILED answer parked in +// pending_promises forever — the exactly-one-Return-per-call invariant broke +// and a compliant caller hung. Found on the wire by the cross-impl L3 lane +// (pipelined-provide scenarios): the C++ recipient pipelines a Call on its +// Accept question, the Accept is refused, and the Call never got a Return. +test "call pipelined on an already-failed answer gets a copy of that exception Return" { + const allocator = std.testing.allocator; + + const Capture = struct { + allocator: std.mem.Allocator, + frames: std.ArrayList([]u8), + + fn onFrame(ctx_ptr: *anyopaque, frame: []const u8) anyerror!void { + const ctx: *@This() = castCtx(*@This(), ctx_ptr); + const copy = try ctx.allocator.dupe(u8, frame); + try ctx.frames.append(ctx.allocator, copy); + } + }; + const Handlers = struct { + // Every call on the bootstrap cap fails, with a NON-default exception + // type so the copy assertion below can tell "the recorded exception + // was replayed" apart from "some fresh .failed exception was built". + fn onCall(ctx_ptr: *anyopaque, peer: *Peer, call: protocol.Call, caps: *const cap_table.InboundCapTable) anyerror!void { + _ = ctx_ptr; + _ = caps; + try peer.sendReturnExceptionTyped(call.question_id, "boom", .overloaded); + } + }; + + var peer = Peer.initDetached(allocator); + defer peer.deinit(); + + var capture = Capture{ + .allocator = allocator, + .frames = std.ArrayList([]u8).empty, + }; + defer { + for (capture.frames.items) |frame| allocator.free(frame); + capture.frames.deinit(allocator); + } + peer.setSendFrameOverride(&capture, Capture.onFrame); + + _ = try peer.setBootstrap(.{ .ctx = &capture, .on_call = Handlers.onCall }); + + const bootstrap_question_id: u32 = 41; + { + var bootstrap_builder = protocol.MessageBuilder.init(allocator); + defer bootstrap_builder.deinit(); + try bootstrap_builder.buildBootstrap(bootstrap_question_id); + const bootstrap_frame = try bootstrap_builder.finish(); + defer allocator.free(bootstrap_frame); + try peer.handleFrame(bootstrap_frame); + } + + // The parent call fails synchronously: its exception Return is on the + // wire BEFORE the next frame is even parsed. + const failed_question_id: u32 = 42; + { + var call_builder = protocol.MessageBuilder.init(allocator); + defer call_builder.deinit(); + var call = try call_builder.beginCall(failed_question_id, 0xABCD, 7); + try call.setTargetPromisedAnswer(bootstrap_question_id); + _ = try call.initCapTableTyped(0); + const call_frame = try call_builder.finish(); + defer allocator.free(call_frame); + try peer.handleFrame(call_frame); + } + try std.testing.expect(peer.failed_answers.contains(failed_question_id)); + + // The pipelined child arrives AFTER that Return — exactly the ordering + // the queued-call drain in sendReturnException can never see. + const late_question_id: u32 = 43; + { + var call_builder = protocol.MessageBuilder.init(allocator); + defer call_builder.deinit(); + var call = try call_builder.beginCall(late_question_id, 0xABCD, 8); + try call.setTargetPromisedAnswer(failed_question_id); + _ = try call.initCapTableTyped(0); + const call_frame = try call_builder.finish(); + defer allocator.free(call_frame); + try peer.handleFrame(call_frame); + } + + // Not queued behind a Return that will never come... + try std.testing.expect(!peer.pending_promises.contains(failed_question_id)); + + // ...but answered, with a COPY of the parent's exception: same reason, + // same type (the retryability signal survives the replay). + try std.testing.expectEqual(@as(usize, 3), capture.frames.items.len); + { + var ret_msg = try protocol.DecodedMessage.init(allocator, capture.frames.items[2]); + defer ret_msg.deinit(); + try std.testing.expectEqual(protocol.MessageTag.@"return", ret_msg.tag); + const ret = try ret_msg.asReturn(); + try std.testing.expectEqual(late_question_id, ret.answer_id); + try std.testing.expectEqual(protocol.ReturnTag.exception, ret.tag); + const ex = ret.exception orelse return error.MissingException; + try std.testing.expectEqualStrings("boom", ex.reason); + try std.testing.expectEqual(protocol.ExceptionType.overloaded, ex.kind()); + } + + // The records live exactly as long as resolved_answers entries: each + // question's Finish clears its own, leak-free under testing allocator. + try std.testing.expect(peer.failed_answers.contains(late_question_id)); + for ([_]u32{ failed_question_id, late_question_id }) |qid| { + var finish_builder = protocol.MessageBuilder.init(allocator); + defer finish_builder.deinit(); + try finish_builder.buildFinish(qid, true, false); + const finish_frame = try finish_builder.finish(); + defer allocator.free(finish_frame); + try peer.handleFrame(finish_frame); + } + try std.testing.expectEqual(@as(usize, 0), peer.failed_answers.count()); +} + test "bootstrap promisedAnswer call still resolves after bootstrap export release" { const allocator = std.testing.allocator; diff --git a/tests/rpc/peer/rpc_three_party_handoff_vatc_test.zig b/tests/rpc/peer/rpc_three_party_handoff_vatc_test.zig index 64c493a..7476799 100644 --- a/tests/rpc/peer/rpc_three_party_handoff_vatc_test.zig +++ b/tests/rpc/peer/rpc_three_party_handoff_vatc_test.zig @@ -2986,6 +2986,85 @@ test "L17 V2-M6: wire refs draining between Provide and Accept cannot kill the s try harness.expectNoProvideState(&host.peer); } +// The recipient does not wait for the Accept's Return before using the +// capability — it PIPELINES a Call on the Accept question (rpc.c++ always +// does; the e2e pipelined-provide scenarios hit exactly this). When the +// Accept is refused, that Call arrives AFTER the refusal already went out, so +// the queued-call drain in sendReturnException ran before there was anything +// to drain. Without the failed_answers record the Call parked in +// pending_promises forever and the C++ recipient hung on a Return that never +// came. Spec rule: every Call gets exactly one Return; a call pipelined on a +// failed answer gets (a copy of) that answer's exception. +test "a Call pipelined on a refused Accept gets the refusal exception, not silence" { + const allocator = std.testing.allocator; + + var host: FrameHost = undefined; + try host.init(allocator, .{}); + defer host.deinitAll(); + + var vat: ImportOwnerVat = undefined; + try vat.init(allocator, &host.index); + defer vat.deinitAll(); + + // A refusal that SURVIVES the receiverHosted lift. This test originally + // staged a receiverHosted target, which the lift now serves; the + // vanished-import shape is the refusal that remains — a site-2 `.promised` + // target whose import dies before the Accept, since site 2 takes no + // Provide-time pin. What is under test is the broken-pipeline rule, not + // which particular refusal produced it. + var deferring = DeferringService{}; + const token = try host.mintToken(allocator, "pipelined-on-refusal"); + defer allocator.free(token); + // The probes live HERE, not inside the staging helper: the staged call's + // Return is dropped in flight, so its question is still outstanding when + // `vat.deinitAll()` cancels it THROUGH this ctx pointer. Helper locals + // would be dead stack by then — a segfault on amd64 under ReleaseSafe. + var bprobe = CapImportProbe{}; + var call_probe = TolerantCall{}; + const service_import = try stageSite2Provision(&vat, &host, &deferring, &bprobe, &call_probe, token, 90); + + try vat.owner.releaseImport(vat.carol_import, 1); + try harness.expectNoImport(&vat.owner, vat.carol_import); + + try host.injectAccept(allocator, 900, token, null); + try std.testing.expectEqual(@as(?protocol.ReturnTag, .exception), host.capture.returnFor(900)); + + // The pipelined Call lands after the refusal: Call{qid=901, + // target=promisedAnswer{900}} — the exact frame order the C++ recipient + // produces (Accept, then Call, with the Return in between on our side). + { + var call_builder = protocol.MessageBuilder.init(allocator); + defer call_builder.deinit(); + var call = try call_builder.beginCall(901, NUMBER_INTERFACE_ID, GET_NUMBER_METHOD_ID); + try call.setTargetPromisedAnswer(900); + _ = try call.initCapTableTyped(0); + const call_frame = try call_builder.finish(); + defer allocator.free(call_frame); + try host.peer.handleFrame(call_frame); + } + + // THE PIN: the pipelined call is ANSWERED — with a copy of the Accept's + // own refusal, not parked in pending_promises waiting forever. + try std.testing.expectEqual(@as(?protocol.ReturnTag, .exception), host.capture.returnFor(901)); + try std.testing.expectEqualStrings(target_unavailable_reason, host.capture.reasonFor(901).?); + try std.testing.expect(!host.peer.pending_promises.contains(900)); + + // Failed closed all the way down: no proxy, and Carol was never reached + // through the refused pipeline either. + try harness.expectNoCrossPeerProxyLinks(&vat.owner); + try harness.expectNoCrossPeerProxyLinks(&host.peer); + try std.testing.expectEqual(@as(u32, 0), vat.carol.get_number_calls); + + { + const frame = try buildFinishFrame(allocator, 90); + defer allocator.free(frame); + try vat.owner.handleFrame(frame); + } + try harness.expectNoProvideState(&vat.owner); + try harness.expectNoProvideState(&host.peer); + try vat.remote.releaseImport(service_import, 1); +} + /// A Return whose single result cap is one the ANSWERING peer only IMPORTS, /// emitted origin-tagged as `receiverHosted` — the same encode path /// `sendReturnProvidedTarget` takes for a `.local` receiverHosted target.