@@ -8,6 +8,17 @@
/// server can't hang job completion forever.
const MAX_DRAIN_PASSES: usize = 8;
/// Classification of a single log-batch upload attempt.
enum SendOutcome {
    /// Server accepted the batch.
    Ok,
    /// Temporary failure (5xx / timeout / network) — retry later, don't lose it.
    Transient,
    /// Server rejected the batch and will keep rejecting it (permanent 4xx) —
    /// drop it; re-queueing would loop forever.
    Permanent,
}
#[derive(Clone)]
pub struct LogReporter {
    client: reqwest::Client,
@@ -53,10 +64,19 @@
            std::mem::take(&mut *buf)
        };
        if !self.send_lines(&lines).await {
            // Put the un-sent lines back ahead of anything appended meanwhile,
            // preserving overall order.
        match self.send_lines(&lines).await {
            SendOutcome::Ok => {}
            SendOutcome::Transient => {
                // Server may recover — put the un-sent lines back ahead of
                // anything appended meanwhile (preserving order) so a later
                // flush retries them (anvil-cli#27).
                let mut buf = self.buffer.lock().await;
                buf.splice(0..0, lines);
            }
            SendOutcome::Permanent => {
                // The server rejected this batch and will keep rejecting it;
                // re-queueing would loop forever (anvil-cli#28). Drop it — the
                // error was already logged by send_lines.
            }
            let mut buf = self.buffer.lock().await;
            buf.splice(0..0, lines);
        }
    }
@@ -74,8 +94,9 @@
        }
    }
    /// Send one batch, classifying the outcome so `flush` knows whether to
    /// re-queue (transient) or drop (permanent).
    /// Send one batch, returning whether the server accepted it.
    async fn send_lines(&self, lines: &[String]) -> bool {
    async fn send_lines(&self, lines: &[String]) -> SendOutcome {
        for attempt in 0..=MAX_RETRIES {
            let result = self
                .client
@@ -87,10 +108,24 @@
                .await;
            match result {
                Ok(resp) if resp.status().is_success() => return true,
                Ok(resp) if resp.status().is_success() => return SendOutcome::Ok,
                Ok(resp) => {
                    let status = resp.status();
                    // A permanent client error (4xx) won't succeed on retry —
                    // don't spin the retry loop or re-queue it (anvil-cli#28).
                    // 408/429 are the retryable exceptions.
                    if status.is_client_error()
                        && status != reqwest::StatusCode::REQUEST_TIMEOUT
                        && status != reqwest::StatusCode::TOO_MANY_REQUESTS
                    {
                        eprintln!(
                            "log upload rejected (status {status}) — dropping {} line(s), not retrying",
                            lines.len()
                        );
                        return SendOutcome::Permanent;
                    }
                    if attempt < MAX_RETRIES {
                        eprintln!("log upload failed (status {}), retrying...", resp.status());
                        eprintln!("log upload failed (status {status}), retrying...");
                        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
                    } else {
                        eprintln!(
@@ -112,7 +147,7 @@
                }
            }
        }
        SendOutcome::Transient
        false
    }
    /// Start a background flush timer. Cancel by aborting the returned handle.
@@ -214,5 +249,60 @@
        let got = received_lines(&server).await;
        assert_eq!(got, vec!["error: could not compile".to_string()]);
    }
    // A permanent server rejection (e.g. 422) must NOT be retried or re-queued —
    // otherwise flush/drain/the periodic timer spin on it forever, wedging the
    // runner (anvil-cli#28, observed on carl). The batch is sent once and dropped.
    #[tokio::test]
    async fn permanent_4xx_is_dropped_without_retry_or_loop() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(422))
            .mount(&server)
            .await;
        let reporter = LogReporter::new(&server.uri(), "job3", "tok");
        reporter.append("noisy build line").await;
        reporter.flush().await;
        // drain() must terminate immediately (buffer was dropped, not re-queued).
        reporter.drain().await;
        // Permanent 4xx => exactly one POST: no in-send retries, no re-queue loop.
        let reqs = server.received_requests().await.unwrap();
        assert_eq!(
            reqs.len(),
            1,
            "permanent 422 should be sent once and dropped, got {} requests",
            reqs.len()
        );
    }
    // A 429 (rate limited) is transient, not permanent — it must be retried,
    // not dropped like other 4xx.
    #[tokio::test]
    async fn rate_limit_429_is_treated_as_transient() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(429))
            .up_to_n_times((MAX_RETRIES + 1) as u64)
            .with_priority(1)
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .with_priority(2)
            .mount(&server)
            .await;
        let reporter = LogReporter::new(&server.uri(), "job4", "tok");
        reporter.append("line").await;
        reporter.flush().await;
        reporter.drain().await;
        assert!(
            received_lines(&server).await.contains(&"line".to_string()),
            "429 should be retried until it succeeds, not dropped"
        );
    }
}