From cade98e711cf4616f3fdad4026bc09f0792b24a9 Mon Sep 17 00:00:00 2001 From: Scott Fleckenstein Date: Thu, 30 Jul 2026 20:46:58 -0800 Subject: [PATCH] rpc: answer calls pipelined on an already-failed answer with that exception MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Call targeting the promisedAnswer of an answer that already returned an exception queued in pending_promises forever: resolved_answers records results Returns only, so a failed answer missed there identically to a still-pending one, and the queued-call drain in sendReturnException runs before a pipelined call sent in the same burst can arrive. Found on the wire by the cross-impl L3 lane (pipelined-provide scenarios): the C++ recipient pipelines its Call on the Accept question, the host refuses the Accept fail-closed, and the Call never got a Return — the recipient hung on the exactly-one-Return-per-call guarantee. Record exception Returns in a failed_answers map (owned reason + type, kept until Finish — the resolved_answers lifecycle), written in sendReturnExceptionNoDrain, the funnel every exception Return passes through, so drained pipeline children self-record and transitive late arrivals are covered. planPromisedTarget consults the record exactly where it would otherwise queue, and the call is answered immediately with a COPY of the recorded exception, preserving the retryability signal — the broken-pipeline behavior of the C++ reference. failed_answers also joins the inbound answer-id namespace so a stale record can never be replayed against a reused id, and the record is skipped for loopback and finished-early answers (no Finish would ever clear it). Ablation-proven tests pin both shapes: the refused cross-peer Accept with a pipelined Call (the e2e wire trace, in the vatc suite) and the plain two-party failed answer (from_peer suite), each asserting reason AND type survive the copy and that Finish drains the record. The e2e C++ driver now observes the refusal THROUGH a pipelined call instead of working around the gap with whenResolved(); the full lane passes against the vendored C++ reference (18/18), full suite 914/914, Stable API surface unchanged (experimental snapshot +4 lines). Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 26 ++++ docs/api-snapshot-experimental.txt | 4 + src/rpc/peer/call/peer_call_orchestration.zig | 17 +++ src/rpc/peer/call/peer_call_targets.zig | 60 ++++++++- src/rpc/peer/mod.zig | 76 +++++++++++- src/rpc/peer/state.zig | 14 +++ tests/e2e/cpp/l3_vatc_client.cpp | 14 ++- .../rpc/peer/rpc_peer_from_peer_zig_test.zig | 116 ++++++++++++++++++ .../rpc_three_party_handoff_vatc_test.zig | 104 ++++++++++++++++ 9 files changed, 419 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b4c913..bcac84f 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 8213053..69506da 100644 --- a/docs/api-snapshot-experimental.txt +++ b/docs/api-snapshot-experimental.txt @@ -969,6 +969,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) @@ -1391,6 +1392,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 13ab646..dcfc4a9 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; @@ -502,6 +503,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. | @@ -1393,6 +1395,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 --------------------------------------------------- @@ -1709,6 +1720,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), @@ -2094,6 +2106,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.*); @@ -5171,6 +5188,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, @@ -5180,7 +5200,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, @@ -7705,6 +7766,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 @@ -9326,8 +9392,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. /// @@ -9340,6 +9407,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 @@ -9428,9 +9496,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 0d98174..96a408d 100644 --- a/tests/e2e/cpp/l3_vatc_client.cpp +++ b/tests/e2e/cpp/l3_vatc_client.cpp @@ -593,16 +593,18 @@ 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; the host must refuse it fail-closed. - // Observe the Accept's own Return via whenResolved() (the unknown-token - // pattern) rather than a pipelined call: a call pipelined on the refused - // Accept question is a separate surface (empirically the host never - // answers it — a known gap), while whenResolved() settles on the Accept - // exception itself. + // Observe the refusal THROUGH A PIPELINED CALL: rpc.c++ sends the Call on + // the Accept question's promisedAnswer without waiting for the Accept's + // Return, so the host must answer that Call with (a copy of) the Accept's + // own exception — the broken-pipeline rule. This wait HUNG FOREVER when + // the host dropped calls pipelined on an already-failed answer (they + // parked in its pending-promise queue with no Return ever coming); this + // probe is the cross-impl regression proof for that fix. bool threw = false; bool named = 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()); 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 93bbb14..530ed29 100644 --- a/tests/rpc/peer/rpc_three_party_handoff_vatc_test.zig +++ b/tests/rpc/peer/rpc_three_party_handoff_vatc_test.zig @@ -2742,6 +2742,110 @@ test "L10: a cross-peer Accept whose provide target resolved to an IMPORT fails try harness.expectNoImport(&vat.owner, vat.carol_import); } +// 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 "L10: 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(); + + // Identical arrangement to the first L10 test: a Provide whose stored + // `.local` target decodes to receiverHosted, so the un-embargoed Accept + // is refused by name in serveProvisionOnPeer. + var introducer = Introducer{}; + _ = try vat.owner.setBootstrap(.{ .ctx = &introducer, .on_call = Introducer.onCall }); + var iprobe = CapImportProbe{}; + _ = try vat.remote.sendBootstrap(&iprobe, CapImportProbe.onReturn); + const introducer_import = iprobe.import_id orelse return error.IntroducerBootstrapFailed; + var pprobe = CapImportProbe{}; + _ = try vat.remote.sendCall(introducer_import, 0x1234_5678_9abc_def0, 0, &pprobe, null, CapImportProbe.onReturn); + const promise_import = pprobe.import_id orelse return error.PromiseNotImported; + const promise_export = introducer.promise_export_id orelse return error.PromiseNotMinted; + try vat.owner.resolvePromiseExportToImport(promise_export, vat.carol_import); + + const token = try host.mintToken(allocator, "l10-pipelined-on-refusal"); + defer allocator.free(token); + { + var token_msg = try message.Message.initUnvalidated(allocator, token); + defer token_msg.deinit(); + const recipient = try token_msg.getRootAnyPointer(); + const frame = try buildProvideFrame(allocator, 90, .{ + .tag = .importedCap, + .imported_cap = promise_export, + .promised_answer = null, + }, recipient); + defer allocator.free(frame); + try vat.owner.handleFrame(frame); + } + + 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(receiver_hosted_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); + + // A compliant recipient Finishes both questions; each Finish clears its + // own failure record (leak-free teardown is the other half, as always). + try std.testing.expectEqual(@as(usize, 2), host.peer.failed_answers.count()); + for ([_]u32{ 900, 901 }) |qid| { + const frame = try buildFinishFrame(allocator, qid); + defer allocator.free(frame); + try host.peer.handleFrame(frame); + } + try std.testing.expectEqual(@as(usize, 0), host.peer.failed_answers.count()); + + // Standard L10 teardown: the provision drains on the owner's Finish and + // every wire reference goes home. + { + const frame = try buildFinishFrame(allocator, 90); + defer allocator.free(frame); + try vat.owner.handleFrame(frame); + } + try std.testing.expectEqual(@as(usize, 0), host.index.by_key.count()); + try harness.expectNoProvideState(&vat.owner); + try harness.expectNoProvideState(&host.peer); + try vat.remote.releaseImport(promise_import, 1); + try vat.remote.releaseImport(introducer_import, 1); + try vat.owner.releaseImport(vat.carol_import, 1); + try harness.expectNoImport(&vat.owner, vat.carol_import); +} + /// 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.