agentd/intel/
failover.rs

1// SPDX-License-Identifier: Apache-2.0
2//! The failover decision — sticky-primary with bounded sweep. RFC 0018 §3.3.
3//!
4//! The client's `complete()` is wrapped: try the **active** endpoint; on a
5//! FAILOVER-CLASS error (connect refused/reset, timeout, HTTP 5xx, 429 after the
6//! endpoint's retry, or a circuit-open skip) advance to the next *available*
7//! endpoint in list order; a *non*-failover error (401/403 auth, 4xx request,
8//! malformed body) is returned immediately (it is the same on every endpoint).
9//! On success, snap `active` back to the lowest-index healthy endpoint
10//! (sticky-primary) so a fallback is temporary by construction.
11//!
12//! §3.4: the wire/adapter/JSON path is UNCHANGED — this is the only net-new
13//! control flow. Each `complete_once` still dials fresh (RFC 0006 §7); the only
14//! state kept between calls is the cheap per-endpoint health/breaker record.
15
16use std::time::Duration;
17
18use super::client::IntelError;
19use super::endpoints::EndpointList;
20use super::health::{BreakerTransition, ErrKind};
21use crate::wire::intel::{Request, Response};
22
23/// How a single endpoint's outcome is classified for failover (RFC 0018 §3.3).
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum FailoverClass {
26    /// Try the next endpoint (connect refused/reset, timeout, 5xx, 429).
27    Failover(ErrKind),
28    /// Do not fail over — fatal/observation, identical on every endpoint
29    /// (auth 401/403, 4xx, malformed body).
30    Fatal,
31}
32
33/// Classify an [`IntelError`] for the failover sweep (RFC 0018 §3.3 table). This
34/// EXTENDS RFC 0006 §3 / RFC 0007 §3.6 — it does not redefine the per-call retry,
35/// which has already run inside `complete_once` before we get here.
36pub fn classify(err: &IntelError) -> FailoverClass {
37    match err {
38        // Transport-layer failures are always failover-class: the endpoint is
39        // down/moving/wedged — a sibling may be fine.
40        IntelError::Transport(e) => {
41            use std::io::ErrorKind::*;
42            let kind = match e.kind() {
43                ConnectionRefused => ErrKind::Refused,
44                ConnectionReset | ConnectionAborted | BrokenPipe => ErrKind::Reset,
45                TimedOut | WouldBlock => ErrKind::Timeout,
46                _ => ErrKind::Refused, // NotFound (DNS), other I/O → treat as down
47            };
48            FailoverClass::Failover(kind)
49        }
50        // HTTP status: 5xx and 429 are failover-class; 401/403 (auth) and other
51        // 4xx are fatal — a bad request/credential is bad on every endpoint.
52        IntelError::Http(code, _) => match *code {
53            500 | 502 | 503 | 504 => FailoverClass::Failover(ErrKind::Http5xx),
54            429 => FailoverClass::Failover(ErrKind::Http429),
55            // any other 5xx is still upstream-transient
56            c if (500..600).contains(&c) => FailoverClass::Failover(ErrKind::Http5xx),
57            _ => FailoverClass::Fatal, // 401/403/4xx
58        },
59        // A malformed body is a bad response everywhere → observation/abort.
60        IntelError::Parse(_) => FailoverClass::Fatal,
61        // An unsupported transport is a config error, not a transient outage.
62        IntelError::Unsupported(_) => FailoverClass::Fatal,
63        // All-endpoints-down is already terminal — not re-classified.
64        IntelError::AllEndpointsDown(_) => FailoverClass::Fatal,
65    }
66}
67
68/// Is this a fatal **auth** failure (401/403)? §6 distinguishes it from all-down:
69/// an auth failure on every endpoint is a misconfig (exit 4 immediately), NOT a
70/// backoff-loop — masking a credential error as a transient outage would hide it.
71pub fn is_auth(err: &IntelError) -> bool {
72    matches!(err, IntelError::Http(401 | 403, _))
73}
74
75/// HTTP statuses a **same-endpoint** retry may clear (RFC 0006 §7): a 429
76/// rate-limit or an upstream 5xx blip. Mirrors the failover-class split in
77/// [`classify`] so the two never drift — a non-transient 4xx (bad request, auth)
78/// is a caller error, identical on a re-dial, and must surface immediately.
79pub fn is_transient_status(code: u16) -> bool {
80    code == 429 || (500..600).contains(&code)
81}
82
83/// The result of one failover sweep, plus the side-channel of breaker/active
84/// transitions the caller surfaces as metrics/events (§4.3/§8) and the
85/// `agentd://intelligence` emission (§4.4).
86pub struct SweepResult {
87    pub outcome: Result<Response, IntelError>,
88    /// `(from, to)` if a failover advanced the endpoint within the sweep.
89    pub failover: Option<(usize, usize)>,
90    /// Breaker transitions observed, as `(endpoint_index, transition)`.
91    pub breaker_changes: Vec<(usize, BreakerTransition)>,
92    /// The new active index if it changed (failover or snap-back).
93    pub active_change: Option<usize>,
94    /// The endpoint that ultimately served the request (on success).
95    pub served_by: Option<usize>,
96}
97
98/// Drive one bounded failover sweep for a single logical `complete` (RFC 0018
99/// §3.3). Visits at most `eps.len()` distinct endpoints (each at most once).
100pub fn complete_resilient(
101    list: &mut EndpointList,
102    req: &Request,
103    timeout: Duration,
104    trace_id: Option<&str>,
105) -> SweepResult {
106    let order = list.attempt_order();
107    let cfg = *list.breaker_config();
108    let mut breaker_changes = Vec::new();
109    let mut failover = None;
110    let mut last_err: Option<IntelError> = None;
111    let mut prev_idx: Option<usize> = None;
112
113    // All endpoints down (every breaker OPEN-and-cooling): the documented §6
114    // terminal — surfaced by the caller as exit-4 (once) / backoff (loop).
115    if order.is_empty() {
116        return SweepResult {
117            outcome: Err(IntelError::AllEndpointsDown(None)),
118            failover: None,
119            breaker_changes,
120            active_change: None,
121            served_by: None,
122        };
123    }
124
125    for idx in order {
126        // A second-or-later attempt in the sweep IS a failover advance (§8 event).
127        if let Some(prev) = prev_idx
128            && prev != idx
129        {
130            failover = Some((prev, idx));
131        }
132        prev_idx = Some(idx);
133
134        match list.ep(idx).complete_once(req, timeout, trace_id) {
135            Ok((resp, latency)) => {
136                if let Some(t) = list.ep(idx).health.record_success(latency) {
137                    breaker_changes.push((idx, t));
138                }
139                let mut active_change = list.set_active(idx);
140                // Snap back to the lowest-index healthy endpoint (sticky-primary).
141                if let Some(snapped) = list.prefer_lowest_healthy() {
142                    active_change = Some(snapped);
143                }
144                return SweepResult {
145                    outcome: Ok(resp),
146                    failover,
147                    breaker_changes,
148                    active_change,
149                    served_by: Some(idx),
150                };
151            }
152            Err(e) => match classify(&e) {
153                FailoverClass::Failover(kind) => {
154                    if let Some(t) = list.ep(idx).health.record_failure(kind, &cfg) {
155                        breaker_changes.push((idx, t));
156                    }
157                    last_err = Some(e);
158                    continue; // advance to the next available endpoint
159                }
160                FailoverClass::Fatal => {
161                    // Auth/4xx/malformed: same on every endpoint → return now.
162                    return SweepResult {
163                        outcome: Err(e),
164                        failover,
165                        breaker_changes,
166                        active_change: None,
167                        served_by: None,
168                    };
169                }
170            },
171        }
172    }
173
174    // Every available endpoint failed over → all down (§6).
175    SweepResult {
176        outcome: Err(IntelError::AllEndpointsDown(last_err.map(Box::new))),
177        failover,
178        breaker_changes,
179        active_change: None,
180        served_by: None,
181    }
182}
183
184#[cfg(test)]
185mod tests {
186    use super::*;
187    use std::io;
188
189    fn io_err(kind: io::ErrorKind) -> IntelError {
190        IntelError::Transport(io::Error::new(kind, "x"))
191    }
192
193    #[test]
194    fn transport_errors_are_failover_class() {
195        assert!(matches!(
196            classify(&io_err(io::ErrorKind::ConnectionRefused)),
197            FailoverClass::Failover(ErrKind::Refused)
198        ));
199        assert!(matches!(
200            classify(&io_err(io::ErrorKind::TimedOut)),
201            FailoverClass::Failover(ErrKind::Timeout)
202        ));
203        assert!(matches!(
204            classify(&io_err(io::ErrorKind::ConnectionReset)),
205            FailoverClass::Failover(ErrKind::Reset)
206        ));
207    }
208
209    #[test]
210    fn http_5xx_and_429_failover_but_4xx_does_not() {
211        assert!(matches!(
212            classify(&IntelError::Http(503, "x".into())),
213            FailoverClass::Failover(ErrKind::Http5xx)
214        ));
215        assert!(matches!(
216            classify(&IntelError::Http(429, "x".into())),
217            FailoverClass::Failover(ErrKind::Http429)
218        ));
219        // auth / request error → fatal, NOT failover
220        assert_eq!(
221            classify(&IntelError::Http(401, "x".into())),
222            FailoverClass::Fatal
223        );
224        assert_eq!(
225            classify(&IntelError::Http(403, "x".into())),
226            FailoverClass::Fatal
227        );
228        assert_eq!(
229            classify(&IntelError::Http(400, "x".into())),
230            FailoverClass::Fatal
231        );
232        assert_eq!(
233            classify(&IntelError::Http(404, "x".into())),
234            FailoverClass::Fatal
235        );
236    }
237
238    #[test]
239    fn malformed_body_is_fatal_not_failover() {
240        assert_eq!(
241            classify(&IntelError::Parse("bad json".into())),
242            FailoverClass::Fatal
243        );
244    }
245
246    #[test]
247    fn auth_detection_distinguishes_from_all_down() {
248        assert!(is_auth(&IntelError::Http(401, "x".into())));
249        assert!(is_auth(&IntelError::Http(403, "x".into())));
250        assert!(!is_auth(&IntelError::Http(503, "x".into())));
251        assert!(!is_auth(&io_err(io::ErrorKind::ConnectionRefused)));
252    }
253
254    #[test]
255    fn transient_status_matches_the_failover_class_split() {
256        // The same-endpoint retry set (429 + all 5xx) must mirror `classify`'s
257        // failover-class HTTP codes so the two never drift.
258        for c in [429, 500, 502, 503, 504, 599] {
259            assert!(is_transient_status(c), "{c} should be transient");
260        }
261        for c in [200, 400, 401, 403, 404, 418] {
262            assert!(!is_transient_status(c), "{c} should NOT be transient");
263        }
264    }
265
266    // --- Sweep integration tests over real TCP endpoints -------------------
267    // A tiny single-shot HTTP server returns a fixed status (+ a canned
268    // OpenAI-compatible body for 200) so the sweep dials a *real* endpoint via
269    // `complete_once`. A closed/never-bound port gives a connect failure.
270
271    use std::io::{Read, Write};
272    use std::net::TcpListener;
273
274    /// Bind `127.0.0.1:0`, serve one request returning `status`, and return the
275    /// `http://127.0.0.1:<port>` URI. The thread self-terminates after one conn.
276    fn serve_status(status: u16) -> String {
277        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
278        let port = listener.local_addr().unwrap().port();
279        std::thread::spawn(move || {
280            if let Ok((mut s, _)) = listener.accept() {
281                let mut buf = [0u8; 2048];
282                let _ = s.read(&mut buf); // drain the request
283                let body = if status == 200 {
284                    r#"{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}"#
285                } else {
286                    r#"{"error":{"message":"boom"}}"#
287                };
288                let resp = format!(
289                    "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
290                    body.len()
291                );
292                let _ = s.write_all(resp.as_bytes());
293                let _ = s.flush();
294            }
295        });
296        format!("http://127.0.0.1:{port}")
297    }
298
299    /// Bind `127.0.0.1:0` and serve one response per element of `statuses` on
300    /// successive connections — so a *same-endpoint* retry (which re-dials) walks
301    /// into the next status in the list. Returns the URI.
302    fn serve_sequence(statuses: Vec<u16>) -> String {
303        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
304        let port = listener.local_addr().unwrap().port();
305        std::thread::spawn(move || {
306            for status in statuses {
307                let Ok((mut s, _)) = listener.accept() else {
308                    break;
309                };
310                let mut buf = [0u8; 2048];
311                let _ = s.read(&mut buf);
312                let body = if status == 200 {
313                    r#"{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}"#
314                } else {
315                    r#"{"error":{"message":"boom"}}"#
316                };
317                let resp = format!(
318                    "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
319                    body.len()
320                );
321                let _ = s.write_all(resp.as_bytes());
322                let _ = s.flush();
323            }
324        });
325        format!("http://127.0.0.1:{port}")
326    }
327
328    /// A bound-then-dropped listener yields a port nothing listens on → connect
329    /// refused (a failover-class transport error).
330    fn dead_endpoint() -> String {
331        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
332        let port = listener.local_addr().unwrap().port();
333        drop(listener);
334        format!("http://127.0.0.1:{port}")
335    }
336
337    fn req() -> Request {
338        Request {
339            model: "m".into(),
340            messages: vec![crate::wire::intel::Message::user("hi")],
341            tools: Vec::new(),
342            max_tokens: 16,
343            temperature: Some(0.0),
344        }
345    }
346
347    fn list_of(uris: &[String]) -> EndpointList {
348        EndpointList::parse_with_env(&uris.join(","), None, &|_| None).unwrap()
349    }
350
351    #[test]
352    fn connect_failure_advances_to_next_healthy_endpoint() {
353        let good = serve_status(200);
354        let mut list = list_of(&[dead_endpoint(), good]);
355        let r = complete_resilient(&mut list, &req(), Duration::from_secs(2), None);
356        assert!(r.outcome.is_ok(), "sweep failed over to the healthy ep");
357        assert_eq!(r.served_by, Some(1));
358        // a failover advance was recorded (0 → 1)
359        assert_eq!(r.failover, Some((0, 1)));
360    }
361
362    #[test]
363    fn http_5xx_advances_to_next_endpoint() {
364        let bad = serve_status(503);
365        let good = serve_status(200);
366        let mut list = list_of(&[bad, good]);
367        let r = complete_resilient(&mut list, &req(), Duration::from_secs(2), None);
368        assert!(r.outcome.is_ok());
369        assert_eq!(r.served_by, Some(1));
370    }
371
372    #[test]
373    fn http_4xx_does_not_failover() {
374        let bad = serve_status(400);
375        let good = serve_status(200);
376        let mut list = list_of(&[bad, good]);
377        let r = complete_resilient(&mut list, &req(), Duration::from_secs(2), None);
378        // a 4xx is fatal — the sweep returns it WITHOUT trying endpoint 1.
379        assert!(matches!(r.outcome, Err(IntelError::Http(400, _))));
380        assert_eq!(r.served_by, None);
381        assert_eq!(r.failover, None);
382    }
383
384    #[test]
385    fn auth_401_does_not_failover() {
386        let bad = serve_status(401);
387        let good = serve_status(200);
388        let mut list = list_of(&[bad, good]);
389        let r = complete_resilient(&mut list, &req(), Duration::from_secs(2), None);
390        assert!(matches!(r.outcome, Err(IntelError::Http(401, _))));
391        assert!(is_auth(&r.outcome.unwrap_err()));
392    }
393
394    #[test]
395    fn circuit_broken_endpoint_is_skipped() {
396        let good = serve_status(200);
397        let mut list = list_of(&[dead_endpoint(), good]);
398        let cfg = *list.breaker_config();
399        // open endpoint 0's breaker up front (threshold 3)
400        for _ in 0..3 {
401            list.ep(0).health.record_failure(ErrKind::Refused, &cfg);
402        }
403        // the sweep skips the broken endpoint 0 entirely → serves on 1, no
404        // failover advance recorded (0 was never dialed).
405        let r = complete_resilient(&mut list, &req(), Duration::from_secs(2), None);
406        assert!(r.outcome.is_ok());
407        assert_eq!(r.served_by, Some(1));
408        assert_eq!(r.failover, None, "broken ep was skipped, not failed-over");
409    }
410
411    #[test]
412    fn all_endpoints_down_yields_all_endpoints_down_error() {
413        let mut list = list_of(&[dead_endpoint(), dead_endpoint()]);
414        // One sweep over two dead endpoints exhausts the list (each failed-over)
415        // → the documented §6 terminal, mapped to exit 4 on `once`.
416        let r = complete_resilient(&mut list, &req(), Duration::from_secs(2), None);
417        assert!(matches!(r.outcome, Err(IntelError::AllEndpointsDown(_))));
418        // After enough sweeps the breakers open and `all_down()` (breaker-state)
419        // also reports true — at which point `attempt_order()` is empty.
420        for _ in 0..3 {
421            let _ = complete_resilient(&mut list, &req(), Duration::from_secs(2), None);
422        }
423        assert!(list.all_down());
424        assert!(list.attempt_order().is_empty());
425    }
426
427    #[test]
428    fn transient_5xx_is_retried_on_the_same_endpoint() {
429        // One 503 then a 200 on a SINGLE endpoint: complete_once's same-endpoint
430        // retry rides out the blip and succeeds in place — no failover, no exit 4.
431        // (This is the once-mode gap the retry closes: with no higher-level loop,
432        // the bare 503 used to propagate straight to exit 4.)
433        let ep = serve_sequence(vec![503, 200]);
434        let mut list = list_of(&[ep]);
435        let r = complete_resilient(&mut list, &req(), Duration::from_secs(2), None);
436        assert!(r.outcome.is_ok(), "same-endpoint retry cleared the 503");
437        assert_eq!(r.served_by, Some(0));
438        assert_eq!(r.failover, None, "handled in place, not failed over");
439    }
440
441    #[test]
442    fn transient_429_is_retried_then_succeeds() {
443        let ep = serve_sequence(vec![429, 200]);
444        let mut list = list_of(&[ep]);
445        let r = complete_resilient(&mut list, &req(), Duration::from_secs(2), None);
446        assert!(r.outcome.is_ok(), "429 rate-limit blip was retried");
447        assert_eq!(r.served_by, Some(0));
448    }
449
450    #[test]
451    fn non_transient_4xx_is_not_retried() {
452        // 400 then 200 on one endpoint: because a 4xx is NOT retried, the first
453        // (400) response surfaces immediately and the 200 is never consumed. If
454        // the retry wrongly fired on 4xx, this would spuriously succeed.
455        let ep = serve_sequence(vec![400, 200]);
456        let mut list = list_of(&[ep]);
457        let r = complete_resilient(&mut list, &req(), Duration::from_secs(2), None);
458        assert!(
459            matches!(r.outcome, Err(IntelError::Http(400, _))),
460            "4xx must surface on the first dial, not be retried"
461        );
462    }
463}
agentd/intel/failover.rs

agentd/intel/
failover.rs
