trusty_console/webhook/relay.rs
1//! Sending a spooled delivery to its target, and deciding what counts as an
2//! acknowledgement.
3//!
4//! Why: ADR-0034 §2 permits deleting a spool entry only when "the target has
5//! acknowledged the frame on the UDS response". The easy wrong answer is to
6//! treat a successful `connect` — or any response at all — as success, which
7//! moves the silent loss down one layer instead of removing it.
8//!
9//! What: [`UdsRelay::deliver`] builds a `trusty_common::webhook_relay::RelayFrame`
10//! carrying the raw body byte-exact plus the provenance record, sends it through
11//! `trusty_common::uds::send_framed_request` (which verifies the socket's `0700`
12//! directory and `0600` mode before writing), and classifies the answer into
13//! [`RelayOutcome`]. The wire types themselves live in `trusty-common` because
14//! step 4's receivers are `trusty-review` and `trusty-analyze`, which cannot
15//! depend on the console.
16//!
17//! #5182 bound those listeners and added the spawn that precedes the dial: a
18//! relay first asks [`super::spawn::TargetSupervisor`] to make sure something is
19//! serving the socket, because ADR-0034 §1 requires the target to exist without
20//! running resident. A supervisor failure is still [`RelayOutcome::Unreachable`]
21//! — a first-class durable state, the entry stays pending and its attempt count
22//! grows, not an error to swallow.
23//!
24//! Test: `webhook/tests.rs` — `relay_*` cases run against a test-double
25//! `UnixListener` bound through `bind_hardened`.
26
27use std::path::{Path, PathBuf};
28use std::time::Duration;
29
30use trusty_common::uds::{UdsRpcError, send_framed_request};
31use trusty_common::webhook_relay::{RelayFrame, RelayResponse};
32
33use super::spawn::SharedSupervisor;
34use super::spool::SpoolEntry;
35
36/// Default budget for one relay round trip.
37///
38/// GitHub's own delivery timeout is 10 s and the relay runs inside the request
39/// that must beat it, so this leaves headroom for the spool write and the
40/// response. It is also the grace period the retry sweep gives a freshly
41/// spooled entry before considering it its own to relay — see
42/// [`super::BackoffPolicy`].
43pub const DEFAULT_RELAY_TIMEOUT: Duration = Duration::from_secs(5);
44
45/// What one relay attempt established.
46///
47/// Why: three states, not two. "Reached the target and it refused" and "never
48/// reached the target" call for different operator action, and neither is an
49/// acknowledgement. Only [`RelayOutcome::Acked`] permits deleting the entry.
50/// What: `Acked` carries nothing; the other two carry a human-readable reason
51/// written into the entry's `last_error`.
52/// Test: `relay_*` cases in `webhook/tests.rs`.
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub enum RelayOutcome {
55 /// The target answered with an explicit `"ack": true`.
56 Acked,
57 /// The target answered, but not with an acknowledgement — a JSON-RPC
58 /// error, `"ack": false`, or a result with no `ack` field at all.
59 Refused {
60 /// Why the target's answer was not an ack.
61 reason: String,
62 },
63 /// The target could not be reached, did not answer, or answered
64 /// unintelligibly.
65 Unreachable {
66 /// Transport-level reason.
67 reason: String,
68 },
69}
70
71impl RelayOutcome {
72 /// True only for [`RelayOutcome::Acked`].
73 ///
74 /// The single predicate the deletion path is allowed to consult, so no call
75 /// site can spell "not unreachable" and delete a refused entry.
76 pub fn is_acked(&self) -> bool {
77 matches!(self, RelayOutcome::Acked)
78 }
79
80 /// The reason string, or `"acknowledged"`.
81 pub fn reason(&self) -> &str {
82 match self {
83 RelayOutcome::Acked => "acknowledged",
84 RelayOutcome::Refused { reason } | RelayOutcome::Unreachable { reason } => reason,
85 }
86 }
87}
88
89/// Console's UDS client for one target.
90///
91/// Why: the dial half of ADR-0034's relay. Since #5182 it can also START the
92/// target: with a supervisor attached it calls `ensure_running` before writing
93/// the frame, which is what lets the target stay non-resident. Without one it
94/// dials whatever is already at the path, which is what the tests and any
95/// externally-managed deployment want.
96/// What: a socket path, a timeout, and an optional supervisor. Cheap to clone;
97/// opens a fresh connection per delivery, matching every other UDS client in
98/// this workspace.
99/// Test: `relay_*` cases in `webhook/tests.rs`.
100#[derive(Debug, Clone)]
101pub struct UdsRelay {
102 socket: PathBuf,
103 timeout: Duration,
104 source: String,
105 supervisor: Option<SharedSupervisor>,
106}
107
108impl UdsRelay {
109 /// Target `socket` with [`DEFAULT_RELAY_TIMEOUT`] and no supervision.
110 pub fn new(socket: impl Into<PathBuf>) -> Self {
111 Self {
112 socket: socket.into(),
113 timeout: DEFAULT_RELAY_TIMEOUT,
114 source: String::new(),
115 supervisor: None,
116 }
117 }
118
119 /// Start the target on demand, under `supervisor`, keyed by `source`.
120 ///
121 /// Test: `relay_with_a_supervisor_still_acks_against_a_live_target`,
122 /// `spawn_adopts_a_socket_that_is_already_served`.
123 pub fn with_supervisor(
124 mut self,
125 source: impl Into<String>,
126 supervisor: SharedSupervisor,
127 ) -> Self {
128 self.source = source.into();
129 self.supervisor = Some(supervisor);
130 self
131 }
132
133 /// Override the round-trip budget.
134 pub fn with_timeout(mut self, timeout: Duration) -> Self {
135 self.timeout = timeout;
136 self
137 }
138
139 /// Socket this relay dials.
140 pub fn socket(&self) -> &Path {
141 &self.socket
142 }
143
144 /// Round-trip budget, which is also the sweep's hands-off grace period for
145 /// an entry the request path may still be relaying.
146 pub fn timeout(&self) -> Duration {
147 self.timeout
148 }
149
150 /// Send one delivery and classify the answer.
151 ///
152 /// Why: the classification is the whole safety property. Anything other
153 /// than an explicit `"ack": true` leaves the entry pending, so a target that
154 /// accepts a connection and then crashes, or answers with an empty object,
155 /// does not cause the delivery to be dropped.
156 ///
157 /// What: dials through the hardened UDS entry point, writes one frame,
158 /// reads one frame, and defers the ack decision to
159 /// `RelayResponse::is_ack` so both halves of the contract agree on it. A
160 /// JSON-RPC error or a non-ack result is [`RelayOutcome::Refused`]; any
161 /// transport failure is [`RelayOutcome::Unreachable`].
162 ///
163 /// Never returns `Err`: every failure is a first-class outcome the caller
164 /// must record durably, and an error return invites the `let _ =` ADR-0034
165 /// §2 forbids.
166 ///
167 /// Test: `relay_acked_response_is_the_only_ack`,
168 /// `relay_treats_a_result_without_ack_as_refused`,
169 /// `relay_treats_a_jsonrpc_error_as_refused`,
170 /// `relay_reports_unreachable_when_no_listener_is_bound`,
171 /// `relay_reports_unreachable_when_the_target_hangs_up_without_answering`.
172 pub async fn deliver(&self, entry: &SpoolEntry) -> RelayOutcome {
173 let frame = RelayFrame::new(
174 &entry.delivery_id,
175 &entry.source,
176 &entry.event,
177 &entry.headers,
178 &entry.body_b64,
179 &entry.provenance,
180 entry.received_at_unix_ms,
181 entry.attempts,
182 );
183 // #5182: make sure something is serving the socket before writing to
184 // it. A supervisor failure is a transport failure — never an ack — so
185 // the entry stays pending and the sweep retries.
186 if let Some(supervisor) = &self.supervisor
187 && let Err(e) = supervisor.ensure_running(&self.source, &self.socket).await
188 {
189 return RelayOutcome::Unreachable {
190 reason: format!("could not start the {} target: {e}", self.source),
191 };
192 }
193
194 let response: Result<RelayResponse, UdsRpcError> =
195 send_framed_request(&self.socket, &frame, self.timeout).await;
196
197 match response {
198 Ok(resp) if resp.is_ack() => RelayOutcome::Acked,
199 Ok(resp) => RelayOutcome::Refused {
200 reason: refusal_reason(&resp),
201 },
202 Err(e) => RelayOutcome::Unreachable {
203 reason: format!("{e}"),
204 },
205 }
206 }
207}
208
209/// Turn a non-ack response into the string stored in the entry's `last_error`.
210///
211/// Prefers the target's own words — a JSON-RPC error message, then a result
212/// `detail` — so an operator reading the durable record sees the target's
213/// diagnosis rather than console's paraphrase of it.
214fn refusal_reason(resp: &RelayResponse) -> String {
215 if let Some(err) = &resp.error {
216 return format!(
217 "target rejected the frame: code {} — {}",
218 err.code, err.message
219 );
220 }
221 match resp.result.as_ref().and_then(|r| r.detail.clone()) {
222 Some(detail) => detail,
223 None if resp.result.is_some() => {
224 "target answered without an explicit \"ack\": true".to_string()
225 }
226 None => "target answered with neither a result nor an error".to_string(),
227 }
228}