wyrd-client 0.3.0

//! Concrete unary Bifrost ingest transport.

use std::time::Duration;

use crate::WyrdClient;
use crate::auth::AuthError;
use crate::error::{WyrdClientError, from_grpc_status};
use crate::transport::GrpcConnection;
use async_trait::async_trait;
use bytes::Bytes;
use uuid::{Uuid, Version};
use wyrd_spec::error::WyrdError;
use wyrd_spec::request_id::RequestId;
use wyrd_spec::trace::TraceContext;
use wyrd_tonic::tonic::{Code, Request, Status, metadata::MetadataValue};
use wyrd_tonic::wyrd::v1::bifrost_ingest_service_client::BifrostIngestServiceClient;
use wyrd_tonic::wyrd::v1::{InsertBatchRequest, ProducerTraceContext};

use crate::bifrost::sink::IngestTransport;
use wyrd_queue::{ClientByteGuard, DurableBatchAck, SealedBatch, SinkError};

/// Maximum number of reconnect retries for one unacknowledged batch.
#[cfg(any(test, feature = "test-support"))]
pub const MAX_FRAME_RETRIES: u32 = 8;

const RETRY_BACKOFF_MS: [u64; 3] = [100, 1_000, 5_000];

/// Bifrost transport retry configuration.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BifrostTransportConfig {
    /// Number of reconnect attempts after the initial batch attempt.
    pub max_frame_retries: u32,
}

impl Default for BifrostTransportConfig {
    fn default() -> Self {
        Self {
            max_frame_retries: 3,
        }
    }
}

impl BifrostTransportConfig {
    /// Build configuration with a bounded retry budget.
    ///
    /// # Arguments
    /// * `max_frame_retries` - Reconnect attempts after the initial batch
    ///   attempt; clamped to the transport's ceiling.
    #[must_use]
    #[cfg(any(test, feature = "test-support"))]
    pub fn with_max_frame_retries(max_frame_retries: u32) -> Self {
        Self {
            max_frame_retries: max_frame_retries.min(MAX_FRAME_RETRIES),
        }
    }
}

/// Authenticated unary gRPC producer for Bifrost batches.
#[derive(Clone)]
pub struct BifrostGrpcTransport {
    connection: GrpcConnection,
    config: BifrostTransportConfig,
}

impl std::fmt::Debug for BifrostGrpcTransport {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BifrostGrpcTransport")
            .field("connection", &self.connection)
            .field("config", &self.config)
            .finish()
    }
}

impl BifrostGrpcTransport {
    /// Connect through the shared Wyrd client authentication path.
    ///
    /// # Arguments
    /// * `client` - The authenticated client whose gRPC channel and token cache
    ///   the transport shares.
    ///
    /// # Errors
    ///
    /// Returns `WyrdClientError::TransportDown` when the gRPC endpoint URI is
    /// invalid or the dial fails on all attempts.
    pub async fn connect(client: &WyrdClient) -> Result<Self, WyrdClientError> {
        Self::connect_with_config(client, BifrostTransportConfig::default()).await
    }

    /// Connect with an explicit bounded retry budget.
    ///
    /// # Arguments
    /// * `client` - The authenticated client whose gRPC channel and token cache
    ///   the transport shares.
    /// * `config` - The bounded retry budget for each batch.
    ///
    /// # Errors
    ///
    /// Returns `WyrdClientError::TransportDown` when the gRPC endpoint URI is
    /// invalid or the dial fails on all attempts.
    pub async fn connect_with_config(
        client: &WyrdClient,
        config: BifrostTransportConfig,
    ) -> Result<Self, WyrdClientError> {
        let connection =
            GrpcConnection::connect(&client.grpc_config, std::sync::Arc::clone(&client.auth))
                .await?;
        Ok(Self::new(connection, config))
    }

    /// Build a transport over an already connected, authenticated channel.
    ///
    /// # Arguments
    /// * `connection` - An already connected, authenticated gRPC channel.
    /// * `config` - The bounded retry budget for each batch.
    #[must_use]
    pub fn new(connection: GrpcConnection, config: BifrostTransportConfig) -> Self {
        Self { connection, config }
    }

    /// Returns the longest one batch can spend inside this transport.
    ///
    /// Every permitted attempt — the initial send, each configured retry, and
    /// the one resend after a forced credential refresh — may run to the
    /// channel's per-call deadline, and every retry first waits its backoff.
    /// Authentication may exchange a token twice: once for the first bearer
    /// and once for the forced refresh. [`IngestTransport::insert_batch`]
    /// enforces this bound over the whole operation, so gate waits and token
    /// exchanges spend the same budget rather than extending it.
    fn attempt_budget(&self) -> Duration {
        let attempts = self.config.max_frame_retries.saturating_add(2);
        let backoff = (0..self.config.max_frame_retries)
            .map(retry_delay)
            .fold(Duration::ZERO, Duration::saturating_add);
        self.connection
            .call_timeout()
            .saturating_mul(attempts)
            .saturating_add(backoff)
            .saturating_add(self.connection.auth().exchange_timeout().saturating_mul(2))
    }

    /// Returns the shortest send deadline a caller may put around
    /// [`IngestTransport::insert_batch`] without cancelling this transport.
    ///
    /// The transport resolves every batch within its attempt budget; one more
    /// per-call deadline of slack lets that result, terminal or retained
    /// ambiguity, reach the caller before the caller's own deadline cancels
    /// the transport mid-budget.
    #[must_use]
    pub(crate) fn send_deadline(&self) -> Duration {
        self.attempt_budget()
            .saturating_add(self.connection.call_timeout())
    }

    /// Sends one authenticated ingest attempt and verifies its ACK identity.
    ///
    /// The attempt carries `request_id` when supplied, otherwise a fresh one.
    ///
    /// # Errors
    ///
    /// Returns a terminal attempt error when credentials or metadata cannot be
    /// attached or the ACK names another batch, and the classified gRPC
    /// refusal otherwise.
    async fn send_once(
        &self,
        request: InsertBatchRequest,
        request_id: Option<&RequestId>,
    ) -> Result<(), AttemptError> {
        let expected_batch_id = request.wyrd_batch_id.clone();
        let mut request = Request::new(request);
        self.add_auth_metadata(&mut request, request_id)
            .await
            .map_err(AttemptError::terminal)?;
        // The server owns the configured ingest wire ceiling and refuses a
        // larger frame with its typed payload-too-large error; tonic's default
        // encoding size is unbounded, so the client sends what it was given.
        let mut client = BifrostIngestServiceClient::new(self.connection.channel());
        let response = client
            .insert_batch(request)
            .await
            .map_err(AttemptError::from_status)?
            .into_inner();
        if response.wyrd_batch_id == expected_batch_id {
            Ok(())
        } else {
            Err(AttemptError::terminal(WyrdError::Internal {
                message: "bifrost ACK does not match the submitted batch".to_owned(),
                details: serde_json::json!({
                    "expected_batch_id": hex::encode(&expected_batch_id),
                    "actual_batch_id": hex::encode(&response.wyrd_batch_id),
                }),
            }))
        }
    }

    /// Retries an internally owned frame without copying its payload bytes.
    ///
    /// Transport loss and stable ingest-busy retry within the bounded budget.
    /// An `Unauthenticated` refusal forces one credential refresh through the
    /// shared [`crate::auth::AuthMiddleware`] and resends the same batch once,
    /// so an expired or revoked cached token recovers without a second refresh
    /// owner; a repeated refusal is terminal. Every attempt carries the same
    /// `request_id` and producer trace contexts when supplied.
    ///
    /// # Errors
    ///
    /// Returns [`SinkError::Retryable`] when the attempts end on an ambiguous
    /// or busy refusal, or after any earlier attempt was ambiguous, because
    /// the server may then hold the batch and only a resend of the same
    /// identity can tell. Returns [`SinkError::Terminal`] for a definite
    /// refusal with no earlier ambiguity, including a failed forced refresh.
    async fn send_owned_bytes(
        &self,
        table: &str,
        batch_id: [u8; 16],
        request_id: Option<&RequestId>,
        arrow_ipc: Bytes,
        producer_trace_contexts: &[ProducerTraceContext],
    ) -> Result<(), SinkError> {
        let mut retry_number = 0_u32;
        let mut refreshed = false;
        let mut uncertain = false;
        let settle = |error: WyrdError, retryable: bool| {
            if retryable {
                SinkError::Retryable(error)
            } else {
                SinkError::Terminal(error)
            }
        };
        loop {
            let request = InsertBatchRequest {
                table: table.to_owned(),
                arrow_ipc: arrow_ipc.clone(),
                wyrd_batch_id: Bytes::copy_from_slice(&batch_id),
                producer_trace_contexts: producer_trace_contexts.to_vec(),
            };
            match self.send_once(request, request_id).await {
                Ok(()) => return Ok(()),
                Err(error) if error.unauthenticated && !refreshed => {
                    refreshed = true;
                    self.connection
                        .auth()
                        .force_refresh()
                        .await
                        .map_err(|error| settle(auth_error_to_wyrd(error), uncertain))?;
                }
                Err(error) if error.retryable && retry_number < self.config.max_frame_retries => {
                    uncertain = true;
                    tokio::time::sleep(retry_delay(retry_number)).await;
                    retry_number = retry_number.saturating_add(1);
                }
                Err(error) => return Err(settle(error.error, error.retryable || uncertain)),
            }
        }
    }

    /// Attaches the bearer token, `wyrd-request-id`, and the caller's W3C
    /// trace context to one attempt.
    ///
    /// A supplied `request_id` is forwarded unchanged through the shared
    /// [`crate::auth::AuthMiddleware::request_id`]; otherwise one is minted.
    /// The trace context is the shared transport's outbound context, sent as
    /// `traceparent` and, when non-empty, `tracestate` metadata.
    ///
    /// # Errors
    ///
    /// Returns the mapped authentication failure when no bearer is available,
    /// and [`WyrdError::Internal`] when a value cannot be gRPC metadata.
    async fn add_auth_metadata<T>(
        &self,
        request: &mut Request<T>,
        request_id: Option<&RequestId>,
    ) -> Result<(), WyrdError> {
        let bearer = self
            .connection
            .auth()
            .bearer()
            .await
            .map_err(auth_error_to_wyrd)?;
        let access_token = format!("Bearer {}", bearer.expose());
        let access_token =
            MetadataValue::try_from(access_token.as_str()).map_err(|_| WyrdError::Internal {
                message: "bifrost access token cannot be represented as gRPC metadata".to_owned(),
                details: serde_json::json!({}),
            })?;
        request
            .metadata_mut()
            .insert("x-wyrd-access-token", access_token);

        let request_id = self
            .connection
            .auth()
            .request_id(request_id.map(RequestId::as_str));
        let request_id =
            MetadataValue::try_from(request_id.as_str()).map_err(|_| WyrdError::Internal {
                message: "bifrost request ID cannot be represented as gRPC metadata".to_owned(),
                details: serde_json::json!({}),
            })?;
        request.metadata_mut().insert("wyrd-request-id", request_id);
        for (name, value) in crate::transport::trace::headers() {
            let value =
                MetadataValue::try_from(value.as_str()).map_err(|_| WyrdError::Internal {
                    message: "trace context cannot be represented as gRPC metadata".to_owned(),
                    details: serde_json::json!({}),
                })?;
            request.metadata_mut().insert(name, value);
        }
        Ok(())
    }
}

#[async_trait]
impl IngestTransport<ClientByteGuard> for BifrostGrpcTransport {
    /// Borrows a sealed owner and shares its frame with gRPC without copying.
    ///
    /// The queue's `Arc<Vec<u8>>` shell becomes an owned [`Bytes`] transport
    /// frame through `Bytes::from_owner` without copying. The queue keeps its
    /// guard and stable UUIDv7 while this cancellable borrowed attempt runs.
    ///
    /// # Errors
    ///
    /// Returns [`SinkError::Terminal`] for frame validation and every definite
    /// server refusal. Returns [`SinkError::Retryable`] with the last ambiguous
    /// unavailability or stable `WYRD_VALA_429_INGEST_BUSY` once this
    /// transport's bounded retry budget is exhausted with the same UUID and
    /// bytes, and when the whole operation, including authentication waits
    /// and refresh, outlives that budget: the outcome is then unknown, so the
    /// queue retains the batch and resends the same identity later rather
    /// than counting a possibly durable batch as lost. Cancellation before an
    /// ACK likewise leaves the queue's borrowed UUID, bytes, and permit intact
    /// for retry resolution.
    async fn insert_batch(
        &self,
        batch: &SealedBatch<ClientByteGuard>,
    ) -> Result<DurableBatchAck, SinkError> {
        if let Err(error) =
            validate_table(&batch.table).and_then(|()| validate_batch_id(batch.batch_id))
        {
            return Err(SinkError::Terminal(error));
        }
        let bytes = Bytes::from_owner(batch.frame.shared_bytes());
        let producers: Vec<_> = batch
            .producer_traces
            .iter()
            .filter_map(|trace| producer_trace_context(trace.context()))
            .collect();
        let send = self.send_owned_bytes(
            &batch.table,
            batch.batch_id,
            batch.request_id.as_ref(),
            bytes,
            &producers,
        );
        tokio::time::timeout(self.attempt_budget(), send)
            .await
            .unwrap_or_else(|_| {
                Err(SinkError::Retryable(transport_unavailable(
                    "attempt budget exhausted before ACK",
                )))
            })
            .map(|()| DurableBatchAck {
                batch_id: batch.batch_id,
                rows: batch.rows,
            })
    }
}

#[derive(Debug)]
struct AttemptError {
    /// Typed stable error returned to the caller when retries stop.
    error: WyrdError,
    /// Whether this attempt is ambiguous or carries stable ingest-busy identity.
    retryable: bool,
    /// Whether the server refused the attempt's credential, which earns one
    /// forced refresh before the refusal becomes terminal.
    unauthenticated: bool,
}

impl AttemptError {
    /// Wraps a failure that must settle without another transport attempt.
    fn terminal(error: WyrdError) -> Self {
        Self {
            error,
            retryable: false,
            unauthenticated: false,
        }
    }

    /// Classifies a gRPC failure by stable reason before considering transport loss.
    fn from_status(status: Status) -> Self {
        let stable = from_grpc_status(&status);
        if stable.code() == "WYRD_VALA_429_INGEST_BUSY" {
            Self {
                error: stable,
                retryable: true,
                unauthenticated: false,
            }
        } else if retryable_status(&status) {
            Self {
                error: transport_unavailable("unary ingest transport failed before ACK"),
                retryable: true,
                unauthenticated: false,
            }
        } else {
            Self {
                unauthenticated: status.code() == Code::Unauthenticated,
                ..Self::terminal(stable)
            }
        }
    }
}

/// Refuses a batch that names no table before it reaches the transport.
///
/// Frame size is deliberately not checked here: the ingest wire ceiling is an
/// operator-configured server setting, and the server refuses a larger frame
/// with its typed payload-too-large error.
///
/// # Errors
///
/// Returns [`WyrdError::Validation`] when `table` is empty.
fn validate_table(table: &str) -> Result<(), WyrdError> {
    if table.is_empty() {
        return Err(WyrdError::Validation {
            message: "bifrost batch table must not be empty".to_owned(),
            details: serde_json::json!({ "field": "table" }),
        });
    }
    Ok(())
}

/// Refuses a queue-owned frame identity that is not a `UUIDv7`.
///
/// The identity is supplied by the producer that sealed the frame, so the
/// transport verifies it rather than trusting the queue's caller.
///
/// # Errors
///
/// Returns [`WyrdError::Validation`] when `batch_id` is not a `UUIDv7`.
fn validate_batch_id(batch_id: [u8; 16]) -> Result<(), WyrdError> {
    if Uuid::from_bytes(batch_id).get_version() != Some(Version::SortRand) {
        return Err(WyrdError::Validation {
            message: "bifrost batch_id must be UUIDv7".to_owned(),
            details: serde_json::json!({ "field": "wyrd_batch_id" }),
        });
    }
    Ok(())
}

/// Returns whether a status represents ambiguous transport loss before ACK.
fn retryable_status(status: &Status) -> bool {
    matches!(
        status.code(),
        Code::Cancelled | Code::DeadlineExceeded | Code::Unavailable | Code::Unknown
    )
}

fn retry_delay(retry_number: u32) -> Duration {
    Duration::from_millis(
        RETRY_BACKOFF_MS
            .get(retry_number as usize)
            .copied()
            .unwrap_or(5_000),
    )
}

fn transport_unavailable(stage: &str) -> WyrdError {
    WyrdError::ServiceUnavailable {
        message: "bifrost ingest transport unavailable before batch ACK".to_owned(),
        details: serde_json::json!({ "stage": stage }),
    }
}

fn auth_error_to_wyrd(error: AuthError) -> WyrdError {
    match error {
        AuthError::Server(error) => error,
        AuthError::Client(error) => {
            tracing::warn!(error = %error, "bifrost gRPC authentication transport unavailable");
            transport_unavailable("authentication")
        }
    }
}

/// Encode one producer's span context as the binary ids the ingest wire
/// carries, or `None` when either hex id does not decode.
fn producer_trace_context(trace: &TraceContext) -> Option<ProducerTraceContext> {
    Some(ProducerTraceContext {
        trace_id: hex::decode(&trace.trace_id).ok()?,
        span_id: hex::decode(&trace.span_id).ok()?,
        trace_flags: u32::from(trace.trace_flags),
        trace_state: trace.trace_state.clone(),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::bifrost::BifrostClientError;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

    use arrow::array::Int64Array;
    use arrow::datatypes::{DataType, Field, Schema};
    use arrow::record_batch::RecordBatch;
    use wyrd_queue::{ClientByteBudget, OwnedIpcBytes, ProducerTrace, QueueConfig, WyrdQueueError};

    use crate::auth::AuthMiddleware;
    use crate::config::ClientConfig;
    use crate::transport::HttpTransport;
    use crate::transport::credential::{AccessTokenSource, MintedAccessToken, ResolvedCredential};
    use secrecy::SecretString;
    use tokio::sync::Mutex;
    use wyrd_spec::auth::SecretBearer;
    use wyrd_spec::vala::error::BifrostError;
    use wyrd_tonic::error::wyrd_error_to_status;
    use wyrd_tonic::tonic::Response;
    use wyrd_tonic::tonic::transport::Server;
    use wyrd_tonic::tonic::transport::server::TcpIncoming;
    use wyrd_tonic::wyrd::v1::InsertBatchResponse;
    use wyrd_tonic::wyrd::v1::bifrost_ingest_service_server::{
        BifrostIngestService, BifrostIngestServiceServer,
    };

    /// Renders `error` through the server's own gRPC problem producer.
    fn refusal(error: BifrostError) -> Status {
        wyrd_error_to_status(WyrdError::from(error), None)
    }

    /// Serves `service` on an ephemeral loopback port and returns its address
    /// with the server task.
    ///
    /// The bound listener is handed to the server rather than dropped and
    /// rebound, so the port accepts connections before this returns and a
    /// client configured with no connect retries cannot race server startup.
    ///
    /// # Panics
    ///
    /// Panics when the loopback port cannot be bound or its address read.
    async fn serve_ingest<S: BifrostIngestService>(
        service: S,
    ) -> (
        std::net::SocketAddr,
        tokio::task::JoinHandle<Result<(), wyrd_tonic::tonic::transport::Error>>,
    ) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("test port binds");
        let address = listener.local_addr().expect("test address resolves");
        let server = tokio::spawn(
            Server::builder()
                .add_service(BifrostIngestServiceServer::new(service))
                .serve_with_incoming(TcpIncoming::from(listener)),
        );
        (address, server)
    }

    /// Scripted real gRPC service that returns busy, ACK, then permanent capacity.
    #[derive(Clone, Default)]
    struct ScriptedIngest {
        /// Ordered stable batch identities observed at the RPC boundary.
        batch_ids: Arc<Mutex<Vec<Vec<u8>>>>,
        /// Ordered `wyrd-request-id` metadata observed at the RPC boundary.
        request_ids: Arc<Mutex<Vec<String>>>,
        /// Ordered producer trace contexts each attempt carried.
        producers: Arc<Mutex<Vec<Vec<ProducerTraceContext>>>>,
        /// Selects the next scripted response without serializing request recording.
        attempts: Arc<AtomicUsize>,
        /// Answers every attempt with stable busy, standing in for a sink that
        /// never recovers.
        always_busy: Arc<AtomicBool>,
    }

    #[wyrd_tonic::tonic::async_trait]
    impl BifrostIngestService for ScriptedIngest {
        /// Returns stable busy once, acknowledges its retry, then refuses permanently.
        ///
        /// # Errors
        ///
        /// Returns scripted stable busy and WAL-capacity statuses so the real
        /// client transport proves its retry and terminal classification.
        async fn insert_batch(
            &self,
            request: Request<InsertBatchRequest>,
        ) -> Result<Response<InsertBatchResponse>, Status> {
            let request_id = request
                .metadata()
                .get("wyrd-request-id")
                .and_then(|value| value.to_str().ok())
                .unwrap_or_default()
                .to_owned();
            self.request_ids.lock().await.push(request_id);
            let request = request.into_inner();
            self.producers
                .lock()
                .await
                .push(request.producer_trace_contexts.clone());
            self.batch_ids
                .lock()
                .await
                .push(request.wyrd_batch_id.to_vec());
            let attempt = self.attempts.fetch_add(1, Ordering::AcqRel);
            match if self.always_busy.load(Ordering::Acquire) {
                0
            } else {
                attempt
            } {
                0 => {
                    let mut status = refusal(BifrostError::IngestBusy {
                        table: "events".to_owned(),
                    });
                    status
                        .metadata_mut()
                        .insert("retry-after-ms", MetadataValue::from_static("1000"));
                    Err(status)
                }
                1 => Ok(Response::new(InsertBatchResponse {
                    wyrd_batch_id: request.wyrd_batch_id,
                })),
                _ => Err(refusal(BifrostError::WalDiskFull)),
            }
        }
    }

    /// Seals one queue-owned batch so the test drives the real transport door.
    ///
    /// The transport only accepts a [`SealedBatch`]; minting the identity and
    /// the byte guard here is what a producer does before it hands the batch to
    /// the sink.
    fn sealed_batch(budget: &ClientByteBudget, bytes: Vec<u8>) -> SealedBatch<ClientByteGuard> {
        let guard = budget
            .reserve(bytes.len())
            .expect("test frame fits the budget");
        SealedBatch {
            table: "events".to_owned(),
            batch_id: Uuid::now_v7().into_bytes(),
            frame: OwnedIpcBytes::new(bytes, guard),
            rows: 1,
            request_id: None,
            producer_traces: Vec::new(),
        }
    }

    #[test]
    fn invalid_batch_id_is_rejected_before_transport() {
        let error = validate_batch_id([0; 16]).expect_err("non-v7 ID must fail");
        assert_eq!(error.code(), "WYRD_SPEC_400_VALIDATION");
    }

    #[test]
    fn only_transport_loss_statuses_are_retryable() {
        assert!(retryable_status(&Status::new(Code::Unavailable, "down")));
        assert!(!retryable_status(&Status::new(
            Code::InvalidArgument,
            "bad batch"
        )));
    }

    /// Stable busy retries while every other capacity status terminalizes, and
    /// a sink that stays busy settles terminally after exactly the configured
    /// attempts under one UUID, request ID, and set of producer contexts.
    #[tokio::test]
    async fn ingest_busy_retries_same_batch_and_permanent_capacity_is_terminal() {
        let service = ScriptedIngest::default();
        let (address, server) = serve_ingest(service.clone()).await;

        let config = ClientConfig {
            grpc: crate::transport::GrpcConfig {
                endpoint: format!("http://{address}"),
                connect_retries: 0,
                ..crate::transport::GrpcConfig::default()
            },
            ..ClientConfig::default()
        };
        let auth = AuthMiddleware::new(
            &config,
            ResolvedCredential::BearerToken(SecretString::from("test-token")),
        )
        .expect("test auth builds");
        let http = HttpTransport::new(&config.http, auth.clone()).expect("test HTTP layer builds");
        let client = WyrdClient::from_parts(auth, http, config.grpc);
        let transport = BifrostGrpcTransport::connect_with_config(
            &client,
            BifrostTransportConfig::with_max_frame_retries(3),
        )
        .await
        .expect("real test gRPC transport connects");
        let budget = ClientByteBudget::new(QueueConfig::DEFAULT_CLIENT_BYTE_LIMIT);
        let request_id = RequestId::now_v7();
        let producer = TraceContext {
            trace_id: "0af7651916cd43dd8448eb211c80319c".to_owned(),
            span_id: "b7ad6b7169203331".to_owned(),
            trace_flags: 1,
            trace_state: "wyrd=producer".to_owned(),
        };
        let retried = SealedBatch {
            request_id: Some(request_id.clone()),
            producer_traces: vec![Arc::new(
                ProducerTrace::charged(&budget, producer).expect("context bytes fit"),
            )],
            ..sealed_batch(&budget, vec![1, 2, 3])
        };
        let batch_id = retried.batch_id;
        transport
            .insert_batch(&retried)
            .await
            .expect("stable busy retries to concrete ACK");
        let observed = service.batch_ids.lock().await.clone();
        assert_eq!(observed, vec![batch_id.to_vec(), batch_id.to_vec()]);
        assert_eq!(
            *service.request_ids.lock().await,
            vec![request_id.to_string(), request_id.to_string()],
            "every retry carries the batch's originating request"
        );
        let sent = ProducerTraceContext {
            trace_id: hex::decode("0af7651916cd43dd8448eb211c80319c").expect("hex"),
            span_id: hex::decode("b7ad6b7169203331").expect("hex"),
            trace_flags: 1,
            trace_state: "wyrd=producer".to_owned(),
        };
        assert_eq!(
            *service.producers.lock().await,
            vec![vec![sent.clone()], vec![sent]],
            "every retry carries the batch's sealed producer contexts"
        );

        let terminal = transport
            .insert_batch(&sealed_batch(&budget, vec![4]))
            .await
            .expect_err("permanent capacity terminalizes without retry");
        assert_ne!(terminal.error().code(), "WYRD_VALA_429_INGEST_BUSY");
        assert!(
            matches!(terminal, SinkError::Terminal(_)),
            "permanent capacity must not ask the queue to retry: {terminal:?}"
        );
        assert_eq!(service.attempts.load(Ordering::Acquire), 3);

        service.always_busy.store(true, Ordering::Release);
        let exhausting = BifrostGrpcTransport::connect_with_config(
            &client,
            BifrostTransportConfig::with_max_frame_retries(1),
        )
        .await
        .expect("real test gRPC transport connects");
        let exhausted = SealedBatch {
            request_id: Some(request_id.clone()),
            ..sealed_batch(&budget, vec![5])
        };
        let error = exhausting
            .insert_batch(&exhausted)
            .await
            .expect_err("a sink that stays busy exhausts the retry budget");
        assert!(
            matches!(&error, SinkError::Retryable(error) if error.code() == "WYRD_VALA_429_INGEST_BUSY"),
            "exhaustion asks the queue to retain the batch with the last refusal: {error:?}"
        );
        assert_eq!(
            service.attempts.load(Ordering::Acquire),
            5,
            "one retry only"
        );
        assert_eq!(
            service.batch_ids.lock().await[3..],
            [exhausted.batch_id.to_vec(), exhausted.batch_id.to_vec()]
        );
        assert_eq!(
            service.request_ids.lock().await[3..],
            [request_id.to_string(), request_id.to_string()]
        );
        drop((retried, exhausted));
        assert_eq!(
            budget.used_bytes(),
            0,
            "settled batches release their bytes"
        );
        server.abort();

        // Pin the classifier independently so malformed capacity identities
        // cannot become transport retries if the scripted server changes.
        let busy = refusal(BifrostError::IngestBusy {
            table: "events".to_owned(),
        });
        assert!(AttemptError::from_status(busy).retryable);

        for permanent in [
            BifrostError::PayloadTooLarge { bytes: 2, limit: 1 },
            BifrostError::WalDiskFull,
        ] {
            let status = refusal(permanent);
            assert_eq!(status.code(), Code::ResourceExhausted);
            assert!(!AttemptError::from_status(status).retryable);
        }
    }

    /// Real gRPC service that counts each attempt and never answers it, so the
    /// client's per-call deadline ends every call.
    #[derive(Clone, Default)]
    struct HeldIngest {
        /// Attempts that reached the RPC boundary.
        attempts: Arc<AtomicUsize>,
    }

    #[wyrd_tonic::tonic::async_trait]
    impl BifrostIngestService for HeldIngest {
        /// Counts the attempt and holds it open until the client's deadline
        /// cancels it.
        ///
        /// # Errors
        ///
        /// Never returns; the client abandons the call at its deadline.
        async fn insert_batch(
            &self,
            _request: Request<InsertBatchRequest>,
        ) -> Result<Response<InsertBatchResponse>, Status> {
            self.attempts.fetch_add(1, Ordering::AcqRel);
            std::future::pending().await
        }
    }

    /// Sends one one-row batch through a real facade authenticated by
    /// `credential` against `service`, whose producer asks for a 1 ms send
    /// deadline, then shuts the facade down. `http_timeout_ms` bounds a
    /// renewable credential's mint.
    ///
    /// Every send ends without an acknowledgement, so the outcome is
    /// ambiguous: the queue must retain the oldest batch with its bytes and
    /// retry slot, report no loss, and refuse to report shutdown as drained.
    ///
    /// Returns the RPC attempts `attempts` counted when the flush returned, the
    /// loss observer's reports, and the facade's auth owner.
    ///
    /// # Panics
    ///
    /// Panics when the service, facade, or batch cannot be built, when a
    /// flush returns an unexpected error, the ambiguous batch releases its
    /// ownership, or shutdown reports the unacknowledged batch as drained.
    async fn retain_held_batch<S: BifrostIngestService>(
        service: S,
        attempts: &AtomicUsize,
        credential: ResolvedCredential,
        http_timeout_ms: u64,
    ) -> (usize, Vec<u64>, Arc<AuthMiddleware>) {
        let (address, server) = serve_ingest(service).await;

        let config = ClientConfig {
            grpc: crate::transport::GrpcConfig {
                endpoint: format!("http://{address}"),
                connect_retries: 0,
                timeout_ms: 50,
                ..crate::transport::GrpcConfig::default()
            },
            http: crate::transport::HttpConfig {
                timeout_ms: http_timeout_ms,
                ..crate::transport::HttpConfig::default()
            },
            ..ClientConfig::default()
        };
        let auth = AuthMiddleware::new(&config, credential).expect("test auth builds");
        let http = HttpTransport::new(&config.http, auth.clone()).expect("test HTTP layer builds");
        let client = WyrdClient::from_parts(Arc::clone(&auth), http, config.grpc);
        let bifrost = crate::bifrost::Bifrost::assemble(
            &client,
            None,
            QueueConfig {
                linger_ms: 60_000,
                flush_timeout_ms: 1,
                ..QueueConfig::default()
            },
        )
        .await
        .expect("real test gRPC transport connects");
        let losses = Arc::new(Mutex::new(Vec::new()));
        bifrost.observe_losses({
            let losses = Arc::clone(&losses);
            move |rows| {
                losses
                    .try_lock()
                    .expect("loss observer is uncontended")
                    .push(rows);
            }
        });

        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
        let batch = RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(vec![1]))])
            .expect("test batch builds");
        bifrost
            .enqueue_batch("events", batch, None)
            .expect("batch admitted");
        match bifrost.flush().await {
            Ok(())
            | Err(BifrostClientError::Queue(
                WyrdQueueError::Sink(WyrdError::ServiceUnavailable { .. })
                | WyrdQueueError::FlushTimeout,
            )) => {}
            Err(error) => panic!("unexpected flush failure: {error}"),
        }
        let flushed = attempts.load(Ordering::Acquire);
        let metrics = bifrost.metrics();
        assert!(metrics.owned_bytes > 0, "ambiguity retains the batch bytes");
        assert!(metrics.live_batches >= 1, "ambiguity retains the batch");
        assert_eq!(
            metrics.retry_entries, 1,
            "the oldest ambiguity holds the retry slot"
        );

        bifrost
            .shutdown()
            .await
            .expect_err("an unacknowledged batch is not drained");
        server.abort();
        let losses = losses.lock().await.clone();
        (flushed, losses, auth)
    }

    /// A producer whose own send deadline is shorter than one call cannot
    /// cancel and restart the transport's attempt budget: the facade raises it,
    /// so a flushed batch held at every call's deadline receives the full
    /// configured attempt set, and the ambiguous outcome retains the batch
    /// instead of reporting a loss.
    ///
    /// The upper bound allows the retained batch's scheduled retry, which may
    /// reach the service before the count is read, but never a second whole
    /// attempt set inside the one flush.
    ///
    /// # Panics
    ///
    /// Panics when the flush covers other than one attempt set, a loss is
    /// reported, or the batch's ownership is released.
    #[tokio::test]
    async fn held_calls_exhaust_one_transport_budget_then_retain() {
        let service = HeldIngest::default();
        let attempts = Arc::clone(&service.attempts);
        let (flushed, losses, _auth) = retain_held_batch(
            service,
            &attempts,
            ResolvedCredential::BearerToken(SecretString::from("test-token")),
            crate::transport::HttpConfig::default().timeout_ms,
        )
        .await;
        let configured = BifrostTransportConfig::default().max_frame_retries as usize + 1;
        assert!(
            (configured..2 * configured).contains(&flushed),
            "one attempt set: {flushed}"
        );
        assert!(losses.is_empty(), "ambiguity is not a loss: {losses:?}");
    }

    /// Real gRPC service refusing the first minted credential as
    /// unauthenticated and holding every later attempt open until the client's
    /// per-call deadline.
    #[derive(Clone, Default)]
    struct RefreshHeldIngest {
        /// Attempts that reached the RPC boundary.
        attempts: Arc<AtomicUsize>,
    }

    #[wyrd_tonic::tonic::async_trait]
    impl BifrostIngestService for RefreshHeldIngest {
        /// Refuses `token-1` as unauthenticated and holds any other credential.
        ///
        /// # Errors
        ///
        /// Returns `Unauthenticated` for `token-1`; otherwise never returns.
        async fn insert_batch(
            &self,
            request: Request<InsertBatchRequest>,
        ) -> Result<Response<InsertBatchResponse>, Status> {
            self.attempts.fetch_add(1, Ordering::AcqRel);
            if request
                .metadata()
                .get("x-wyrd-access-token")
                .is_some_and(|token| token == "Bearer token-1")
            {
                return Err(Status::unauthenticated("stale credential"));
            }
            std::future::pending().await
        }
    }

    /// With renewable authentication, one `Unauthenticated` refusal, a refresh
    /// slower than a whole call, and every later call held to its deadline,
    /// the refresh spends the transport's one budget instead of extending it
    /// past the producer's send deadline: the flush covers at most one attempt
    /// set plus the resend, and the ambiguous outcome retains the batch
    /// instead of reporting a loss.
    ///
    /// # Panics
    ///
    /// Panics when the refusal does not force exactly one refresh, the flush
    /// exceeds one attempt set, a loss is reported, or the batch's ownership
    /// is released.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn delayed_refresh_spends_one_transport_budget_then_retains() {
        let service = RefreshHeldIngest::default();
        let attempts = Arc::clone(&service.attempts);
        let source = Arc::new(SequencedSource {
            minted: AtomicUsize::new(0),
            refresh_delay: Duration::from_millis(200),
        });
        let (flushed, losses, _auth) = retain_held_batch(
            service,
            &attempts,
            ResolvedCredential::Renewable(source.clone()),
            crate::transport::HttpConfig::default().timeout_ms,
        )
        .await;
        let attempt_set = BifrostTransportConfig::default().max_frame_retries as usize + 2;
        assert_eq!(
            source.minted.load(Ordering::Acquire),
            2,
            "one forced refresh"
        );
        assert!(
            (2..=attempt_set).contains(&flushed),
            "the refusal is resent within one attempt set: {flushed}"
        );
        assert!(losses.is_empty(), "ambiguity is not a loss: {losses:?}");
    }

    /// A renewable mint that blocks past the whole transport budget cannot
    /// hold the queue's send: the transport's deadline still fires, so the
    /// flush and shutdown complete, no RPC attempt starts, and the batch is
    /// retained rather than reported lost, all before the source is released.
    /// Shutdown reuses the still-running mint instead of starting another.
    /// Once released, the same auth owner serves that mint's token and
    /// refreshes normally.
    ///
    /// # Panics
    ///
    /// Panics when the flush or shutdown is not bounded, an RPC attempt or a
    /// second concurrent mint starts, a loss is reported, ownership is
    /// released, or the released owner cannot authenticate.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn blocking_mint_spends_one_transport_budget_then_retains() {
        let service = HeldIngest::default();
        let attempts = Arc::clone(&service.attempts);
        let (release, held) = std::sync::mpsc::sync_channel::<()>(0);
        let source = Arc::new(HeldSource {
            minted: AtomicUsize::new(0),
            held: std::sync::Mutex::new(held),
        });
        let (flushed, losses, auth) = tokio::time::timeout(
            Duration::from_secs(30),
            retain_held_batch(
                service,
                &attempts,
                ResolvedCredential::Renewable(source.clone()),
                50,
            ),
        )
        .await
        .expect("a blocked mint cannot hold the flush and shutdown");
        assert_eq!(
            source.minted.load(Ordering::Acquire),
            1,
            "shutdown reuses the one running mint"
        );
        assert_eq!(flushed, 0, "no RPC attempt starts");
        assert!(losses.is_empty(), "ambiguity is not a loss: {losses:?}");

        drop(release);
        auth.bearer()
            .await
            .expect("the released mint serves the next caller");
        assert_eq!(source.minted.load(Ordering::Acquire), 1);
        auth.force_refresh()
            .await
            .expect("a later renewable refresh completes");
        assert_eq!(source.minted.load(Ordering::Acquire), 2);
    }

    /// Renewable source whose every mint blocks until its release sender is
    /// dropped.
    struct HeldSource {
        /// Number of mints started.
        minted: AtomicUsize,
        /// Receiver that unblocks every mint once the test drops its sender.
        held: std::sync::Mutex<std::sync::mpsc::Receiver<()>>,
    }

    impl AccessTokenSource for HeldSource {
        /// Stable identity of the test producer.
        fn identity(&self) -> &'static str {
            "held-test-source"
        }

        /// Counts the mint, blocks until released, then mints a long-lived
        /// token.
        ///
        /// # Errors
        ///
        /// Never fails.
        ///
        /// # Panics
        ///
        /// Panics when the release receiver's lock is poisoned.
        fn mint(&self) -> Result<MintedAccessToken, crate::error::WyrdClientError> {
            self.minted.fetch_add(1, Ordering::AcqRel);
            let _released = self.held.lock().expect("release receiver locks").recv();
            Ok(MintedAccessToken {
                access_token: SecretBearer::new("held-token".to_owned()),
                expires_at: chrono::Utc::now() + chrono::Duration::seconds(900),
            })
        }
    }

    /// Renewable source minting `token-1`, `token-2`, … and counting mints.
    struct SequencedSource {
        /// Number of tokens minted so far.
        minted: AtomicUsize,
        /// How long every mint after the first takes, standing in for a slow
        /// refresh.
        refresh_delay: Duration,
    }

    impl AccessTokenSource for SequencedSource {
        /// Stable identity of the test producer.
        fn identity(&self) -> &'static str {
            "sequenced-test-source"
        }

        /// Mints the next sequenced token with a long lifetime, blocking for
        /// the refresh delay after the first.
        ///
        /// # Errors
        ///
        /// Never fails.
        fn mint(&self) -> Result<MintedAccessToken, crate::error::WyrdClientError> {
            let ordinal = self.minted.fetch_add(1, Ordering::AcqRel) + 1;
            if ordinal > 1 {
                std::thread::sleep(self.refresh_delay);
            }
            Ok(MintedAccessToken {
                access_token: SecretBearer::new(format!("token-{ordinal}")),
                expires_at: chrono::Utc::now() + chrono::Duration::seconds(900),
            })
        }
    }

    /// Ingest service that accepts only the refreshed `token-2` credential.
    #[derive(Clone, Default)]
    struct RefreshGatedIngest {
        /// Credentials observed in attempt order.
        tokens: Arc<Mutex<Vec<String>>>,
    }

    #[wyrd_tonic::tonic::async_trait]
    impl BifrostIngestService for RefreshGatedIngest {
        /// Refuses every credential but `token-2` as unauthenticated.
        ///
        /// # Errors
        ///
        /// Returns `Unauthenticated` for a stale credential.
        async fn insert_batch(
            &self,
            request: Request<InsertBatchRequest>,
        ) -> Result<Response<InsertBatchResponse>, Status> {
            let token = request
                .metadata()
                .get("x-wyrd-access-token")
                .and_then(|value| value.to_str().ok())
                .unwrap_or_default()
                .to_owned();
            self.tokens.lock().await.push(token.clone());
            if token == "Bearer token-2" {
                Ok(Response::new(InsertBatchResponse {
                    wyrd_batch_id: request.into_inner().wyrd_batch_id,
                }))
            } else {
                Err(Status::unauthenticated("stale credential"))
            }
        }
    }

    /// An unauthenticated refusal forces exactly one shared refresh, resends
    /// the same batch with the replacement token, and never loops.
    ///
    /// # Panics
    ///
    /// Panics when the refused batch is not resent once with a freshly minted
    /// renewable token.
    #[tokio::test]
    async fn unauthenticated_refusal_forces_one_shared_refresh_and_resends() {
        let service = RefreshGatedIngest::default();
        let (address, server) = serve_ingest(service.clone()).await;

        let exchange = wiremock::MockServer::start().await;
        for token in ["token-1", "token-2"] {
            wiremock::Mock::given(wiremock::matchers::method("POST"))
                .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
                    serde_json::json!({
                        "access_token": token,
                        "token_type": "Bearer",
                        "expires_in": 900,
                    }),
                ))
                .up_to_n_times(1)
                .mount(&exchange)
                .await;
        }
        let config = ClientConfig {
            grpc: crate::transport::GrpcConfig {
                endpoint: format!("http://{address}"),
                connect_retries: 0,
                ..crate::transport::GrpcConfig::default()
            },
            ..ClientConfig::default()
        };
        let source = Arc::new(SequencedSource {
            minted: AtomicUsize::new(0),
            refresh_delay: Duration::ZERO,
        });
        let auth = AuthMiddleware::new(&config, ResolvedCredential::Renewable(source.clone()))
            .expect("test auth builds");
        let http = HttpTransport::new(&config.http, auth.clone()).expect("test HTTP layer builds");
        let client = WyrdClient::from_parts(auth, http, config.grpc);
        let transport = BifrostGrpcTransport::connect(&client)
            .await
            .expect("real test gRPC transport connects");
        let budget = ClientByteBudget::new(QueueConfig::DEFAULT_CLIENT_BYTE_LIMIT);
        transport
            .insert_batch(&sealed_batch(&budget, vec![1]))
            .await
            .expect("refreshed credential acknowledges");
        assert_eq!(source.minted.load(Ordering::Acquire), 2);
        assert_eq!(
            *service.tokens.lock().await,
            vec!["Bearer token-1".to_owned(), "Bearer token-2".to_owned()]
        );
        server.abort();
    }
}