lodestone_client/

subscription.rs

use std::collections::BTreeMap;
use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::io;
use std::sync::Arc;
use std::time::Duration;

use http::{Uri, uri::PathAndQuery};
use hyper_util::rt::TokioIo;
use lodestone_wire::{
    Digest, LODESTONE_API_SPEC_SHA256, LODESTONE_API_SUBSCRIPTION_ACCEPTED_FORMAT_VERSION,
    LODESTONE_API_SUBSCRIPTION_EVENT_FORMAT_VERSION, LODESTONE_API_SUBSCRIPTION_GAP_FORMAT_VERSION,
    LODESTONE_API_SUBSCRIPTION_HEARTBEAT_FORMAT_VERSION, LODESTONE_API_SUBSCRIPTION_START_FORMAT_VERSION,
    LODESTONE_API_SUBSCRIPTION_TERMINAL_FORMAT_VERSION, Resource, SubscriptionAccepted, SubscriptionCursorEntry,
    SubscriptionEvent, SubscriptionFrame, SubscriptionRoute, SubscriptionStart, SubscriptionStartPosition,
};
use rustls::pki_types::CertificateDer;
use tokio::net::TcpStream;
use tokio_rustls::TlsConnector;
use tonic::Code;
use tonic::metadata::MetadataValue;
use tonic::transport::{Channel, Endpoint};
use tower::service_fn;

use crate::config::parse_u64;
use crate::transport::authorization_value;
use crate::{CallOptions, Client, ClientError};

const SUBSCRIPTION_METHOD: &str = "/lodestone.api.FilterSubscription/Subscribe";
const MAX_SUBSCRIPTION_MESSAGE_BYTES: usize = 2 * 1024 * 1024;

#[derive(Clone, PartialEq, prost::Message)]
struct GrpcSubscriptionRequest {
    #[prost(bytes = "vec", tag = "1")]
    json_request: Vec<u8>,
}

#[derive(Clone, PartialEq, prost::Message)]
struct GrpcSubscriptionFrame {
    #[prost(bytes = "vec", tag = "1")]
    json_frame: Vec<u8>,
}

#[derive(Clone, Debug, PartialEq)]
pub struct SubscriptionCheckpoint {
    pub index: Resource,
    pub routes: Vec<SubscriptionRoute>,
    pub cut: Vec<SubscriptionCursorEntry>,
}

pub struct Subscription {
    client: Client,
    template: SubscriptionStart,
    options: CallOptions,
    stream: tonic::Streaming<GrpcSubscriptionFrame>,
    checkpoint: SubscriptionCheckpoint,
    pending: Option<SubscriptionFrame>,
    reconnects: u8,
    ended: bool,
    authorization_revision: Option<u64>,
    last_events: BTreeMap<CursorKey, SubscriptionEvent>,
}

impl Client {
    pub async fn subscribe(
        &self,
        start: SubscriptionStart,
        options: CallOptions,
    ) -> Result<Subscription, SubscriptionError> {
        if self.profile_verification().is_none() {
            return Err(SubscriptionError::ProfileNotVerified);
        }
        validate_start(&start, self.connection_profile().as_wire().schema_digest.clone())?;
        let (stream, accepted) = open_stream(self, &start, &options).await?;
        let checkpoint = checkpoint_from_accepted(&start.index, &accepted)?;
        Ok(Subscription {
            client: self.clone(),
            template: start,
            options,
            stream,
            checkpoint,
            pending: Some(SubscriptionFrame::Accepted { body: accepted }),
            reconnects: 0,
            ended: false,
            authorization_revision: None,
            last_events: BTreeMap::new(),
        })
    }
}

impl Subscription {
    #[must_use]
    pub const fn checkpoint(&self) -> &SubscriptionCheckpoint {
        &self.checkpoint
    }

    pub async fn next_frame(&mut self) -> Result<Option<SubscriptionFrame>, SubscriptionError> {
        if self.ended {
            return Ok(None);
        }
        if let Some(frame) = self.pending.take() {
            return Ok(Some(frame));
        }
        loop {
            let deadline = self.options.local_deadline().map_err(SubscriptionError::Client)?;
            let next = tokio::select! {
                result = tokio::time::timeout_at(deadline, self.stream.message()) => {
                    match result {
                        Ok(result) => result.map_err(SubscriptionError::Status),
                        Err(_) => Err(SubscriptionError::FrameDeadline),
                    }
                }
                () = self.options.cancellation().cancelled() => return Err(SubscriptionError::Cancelled),
            };
            let reconnect = match next {
                Ok(Some(message)) => {
                    let frame = decode_frame(message.json_frame)?;
                    self.validate_and_advance(&frame)?;
                    return Ok(Some(frame));
                }
                Err(SubscriptionError::FrameDeadline) => {
                    self.ended = true;
                    return Err(SubscriptionError::FrameDeadline);
                }
                Ok(None) => true,
                Err(SubscriptionError::Status(status)) if reconnectable_status(&status) => true,
                Err(SubscriptionError::Status(status)) => {
                    self.ended = true;
                    return Err(SubscriptionError::Status(status));
                }
                Err(error) => return Err(error),
            };
            if reconnect {
                if self.reconnects >= self.client.pool_config().maximum_subscription_reconnects {
                    self.ended = true;
                    return Err(SubscriptionError::ReconnectExhausted);
                }
                self.reconnects = self.reconnects.saturating_add(1);
                let backoff = Duration::from_millis(25_u64.saturating_mul(1_u64 << self.reconnects.min(3)));
                tokio::select! {
                    result = tokio::time::timeout_at(deadline, tokio::time::sleep(backoff)) => {
                        if result.is_err() {
                            self.ended = true;
                            return Err(SubscriptionError::FrameDeadline);
                        }
                    }
                    () = self.options.cancellation().cancelled() => {
                        return Err(SubscriptionError::Cancelled);
                    }
                }
                let resume = self.resume_start()?;
                let (stream, accepted) = open_stream(&self.client, &resume, &self.options).await?;
                let checkpoint = checkpoint_from_accepted(&resume.index, &accepted)?;
                if checkpoint.routes != self.checkpoint.routes || checkpoint.cut != self.checkpoint.cut {
                    self.ended = true;
                    return Err(SubscriptionError::CheckpointMismatch);
                }
                self.template = resume;
                self.stream = stream;
            }
        }
    }

    fn resume_start(&self) -> Result<SubscriptionStart, SubscriptionError> {
        let mut start = self.template.clone();
        start.request_id = random_request_id()?;
        start.reconnect_attempt = u32::from(self.reconnects);
        start.start = SubscriptionStartPosition::Resume {
            routes: self.checkpoint.routes.clone(),
            cut: self.checkpoint.cut.clone(),
        };
        Ok(start)
    }

    fn validate_and_advance(&mut self, frame: &SubscriptionFrame) -> Result<(), SubscriptionError> {
        match frame {
            SubscriptionFrame::Accepted { .. } => Err(SubscriptionError::UnexpectedAccepted),
            SubscriptionFrame::Event { body } => {
                validate_event(body, &self.checkpoint.index)?;
                let cursor = cursor_from_event(body);
                let key = CursorKey::from_cursor(&cursor);
                if let Some(previous) = cursor_for_key(&self.checkpoint.cut, &key) {
                    let prior = parse_u64(&previous.sequence.value).ok_or(SubscriptionError::InvalidFrame)?;
                    let next = parse_u64(&cursor.sequence.value).ok_or(SubscriptionError::InvalidFrame)?;
                    if next < prior {
                        return Err(SubscriptionError::CursorRegression);
                    }
                    if next == prior {
                        if self.last_events.get(&key) == Some(body) {
                            return Ok(());
                        }
                        return Err(SubscriptionError::DuplicateIdentityMismatch);
                    }
                }
                upsert_cursor(&mut self.checkpoint.cut, cursor)?;
                self.last_events.insert(key, body.clone());
                Ok(())
            }
            SubscriptionFrame::Heartbeat { body } => {
                if body.format_version != u32::from(LODESTONE_API_SUBSCRIPTION_HEARTBEAT_FORMAT_VERSION) {
                    return Err(SubscriptionError::InvalidFrame);
                }
                let revision = parse_u64(&body.authorization_revision.value).ok_or(SubscriptionError::InvalidFrame)?;
                if parse_u64(&body.route_revision.value).is_none()
                    || parse_u64(&body.authorization_valid_until_millis.value).is_none()
                {
                    return Err(SubscriptionError::InvalidFrame);
                }
                if self.authorization_revision.is_some_and(|previous| revision < previous) {
                    return Err(SubscriptionError::AuthorizationRegression);
                }
                merge_cut(&mut self.checkpoint.cut, &body.confirmed_cut, &self.checkpoint.index)?;
                self.authorization_revision = Some(revision);
                Ok(())
            }
            SubscriptionFrame::Gap { body } => {
                if body.format_version != u32::from(LODESTONE_API_SUBSCRIPTION_GAP_FORMAT_VERSION) || !body.terminal {
                    return Err(SubscriptionError::InvalidFrame);
                }
                for cursor in body
                    .affected
                    .iter()
                    .chain(&body.last_confirmed)
                    .chain(&body.earliest_available)
                {
                    if cursor.index != self.checkpoint.index {
                        return Err(SubscriptionError::InvalidFrame);
                    }
                }
                self.ended = true;
                Ok(())
            }
            SubscriptionFrame::Terminal { body } => {
                if body.format_version != u32::from(LODESTONE_API_SUBSCRIPTION_TERMINAL_FORMAT_VERSION) {
                    return Err(SubscriptionError::InvalidFrame);
                }
                self.ended = true;
                Ok(())
            }
        }
    }
}

async fn open_stream(
    client: &Client,
    start: &SubscriptionStart,
    options: &CallOptions,
) -> Result<(tonic::Streaming<GrpcSubscriptionFrame>, SubscriptionAccepted), SubscriptionError> {
    let payload = serde_json::to_vec(start).map_err(|_| SubscriptionError::RequestCodec)?;
    if payload.is_empty() || payload.len() > MAX_SUBSCRIPTION_MESSAGE_BYTES {
        return Err(SubscriptionError::RequestTooLarge);
    }
    let deadline = options.local_deadline().map_err(SubscriptionError::Client)?;
    let credential = tokio::select! {
        result = tokio::time::timeout_at(deadline, client.credential()) => {
            result.map_err(|_| SubscriptionError::OpenDeadline)?
                .map_err(SubscriptionError::Client)?
        }
        () = options.cancellation().cancelled() => return Err(SubscriptionError::Cancelled),
    };
    let authorization = authorization_value(&credential).map_err(SubscriptionError::Client)?;
    let channel = connect_channel(client, options, deadline).await?;
    let mut grpc = tonic::client::Grpc::new(channel)
        .max_decoding_message_size(MAX_SUBSCRIPTION_MESSAGE_BYTES)
        .max_encoding_message_size(MAX_SUBSCRIPTION_MESSAGE_BYTES);
    let mut request = tonic::Request::new(GrpcSubscriptionRequest { json_request: payload });
    let authorization =
        std::str::from_utf8(authorization.as_slice()).map_err(|_| SubscriptionError::InvalidMetadata)?;
    request.metadata_mut().insert(
        "authorization",
        MetadataValue::try_from(authorization).map_err(|_| SubscriptionError::InvalidMetadata)?,
    );
    request.metadata_mut().insert(
        "x-lodestone-collection-id",
        MetadataValue::try_from(
            client
                .connection_profile()
                .as_wire()
                .expected_collection_id
                .value
                .as_str(),
        )
        .map_err(|_| SubscriptionError::InvalidMetadata)?,
    );
    request.metadata_mut().insert(
        "x-lodestone-index-name",
        MetadataValue::try_from(start.index.name.as_str()).map_err(|_| SubscriptionError::InvalidMetadata)?,
    );
    // INVARIANT: Tower's buffered channel reserves capacity through poll_ready;
    // invoking the RPC without this reservation can panic instead of refusing.
    tokio::select! {
        result = tokio::time::timeout_at(deadline, grpc.ready()) => {
            result.map_err(|_| SubscriptionError::OpenDeadline)?
                .map_err(|_| SubscriptionError::Connect)?;
        }
        () = options.cancellation().cancelled() => return Err(SubscriptionError::Cancelled),
    }
    let response = tokio::select! {
        result = tokio::time::timeout_at(
            deadline,
            grpc.server_streaming(
                request,
                PathAndQuery::from_static(SUBSCRIPTION_METHOD),
                tonic_prost::ProstCodec::default(),
            ),
        ) => result.map_err(|_| SubscriptionError::OpenDeadline)?
            .map_err(SubscriptionError::Status)?,
        () = options.cancellation().cancelled() => return Err(SubscriptionError::Cancelled),
    };
    let mut stream = response.into_inner();
    let first: GrpcSubscriptionFrame = tokio::select! {
        result = tokio::time::timeout_at(deadline, stream.message()) => {
            result.map_err(|_| SubscriptionError::OpenDeadline)?
                .map_err(SubscriptionError::Status)?
        }
        () = options.cancellation().cancelled() => return Err(SubscriptionError::Cancelled),
    }
    .ok_or(SubscriptionError::AcceptedAbsent)?;
    let frame = decode_frame(first.json_frame)?;
    let SubscriptionFrame::Accepted { body } = frame else {
        return Err(SubscriptionError::AcceptedAbsent);
    };
    Ok((stream, body))
}

async fn connect_channel(
    client: &Client,
    options: &CallOptions,
    deadline: tokio::time::Instant,
) -> Result<Channel, SubscriptionError> {
    let config = client.transport_config();
    let mut roots = rustls::RootCertStore::empty();
    for certificate in config.trust.certificates_der() {
        roots
            .add(CertificateDer::from(certificate.clone()))
            .map_err(|_| SubscriptionError::InvalidTls)?;
    }
    let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
    let mut tls = rustls::ClientConfig::builder_with_provider(provider)
        .with_protocol_versions(&[&rustls::version::TLS13])
        .map_err(|_| SubscriptionError::InvalidTls)?
        .with_root_certificates(roots)
        .with_no_client_auth();
    tls.alpn_protocols = vec![b"h2".to_vec()];
    let connector = TlsConnector::from(Arc::new(tls));
    let host = config.stream_endpoint.host().to_owned();
    let port = config.stream_endpoint.port();
    let server_name = rustls::pki_types::ServerName::try_from(config.identity.tls_server_name().to_owned())
        .map_err(|_| SubscriptionError::InvalidTls)?;
    let connect = service_fn(move |_uri: Uri| {
        let connector = connector.clone();
        let host = host.clone();
        let server_name = server_name.clone();
        async move {
            let tcp = TcpStream::connect((host.as_str(), port)).await?;
            let tls = connector.connect(server_name, tcp).await.map_err(io::Error::other)?;
            if tls.get_ref().1.alpn_protocol() != Some(b"h2".as_slice()) {
                return Err(io::Error::other("Core subscription TLS did not negotiate h2"));
            }
            Ok::<_, io::Error>(TokioIo::new(tls))
        }
    });
    let transport_uri = format!(
        "http://{}:{}",
        config.stream_endpoint.host(),
        config.stream_endpoint.port()
    );
    let origin: Uri = config
        .stream_endpoint
        .as_str()
        .parse()
        .map_err(|_| SubscriptionError::InvalidEndpoint)?;
    let endpoint = Endpoint::from_shared(transport_uri)
        .map_err(|_| SubscriptionError::InvalidEndpoint)?
        .connect_timeout(config.pool.connect_timeout)
        .origin(origin);
    tokio::select! {
        result = tokio::time::timeout_at(deadline, endpoint.connect_with_connector(connect)) => {
            result.map_err(|_| SubscriptionError::OpenDeadline)?
                .map_err(|_| SubscriptionError::Connect)
        }
        () = options.cancellation().cancelled() => Err(SubscriptionError::Cancelled),
    }
}

fn validate_start(start: &SubscriptionStart, schema: Digest) -> Result<(), SubscriptionError> {
    if start.format_version != u32::from(LODESTONE_API_SUBSCRIPTION_START_FORMAT_VERSION)
        || start.request_id.is_empty()
        || start.request_id.len() > 128
        || start.request_id.bytes().any(|byte| byte.is_ascii_control())
        || start.index.kind != "index"
        || start.index.name.is_empty()
        || start.index.incarnation.validate().is_err()
        || start.schema_digest != schema
        || start.schema_digest.value != LODESTONE_API_SPEC_SHA256
        || parse_u64(&start.heartbeat_millis.value).is_none_or(|value| value == 0)
    {
        return Err(SubscriptionError::InvalidStart);
    }
    Ok(())
}

fn checkpoint_from_accepted(
    index: &Resource,
    accepted: &SubscriptionAccepted,
) -> Result<SubscriptionCheckpoint, SubscriptionError> {
    if accepted.format_version != u32::from(LODESTONE_API_SUBSCRIPTION_ACCEPTED_FORMAT_VERSION)
        || accepted.schema_digest.value != LODESTONE_API_SPEC_SHA256
        || accepted.routes.is_empty()
    {
        return Err(SubscriptionError::InvalidFrame);
    }
    for route in &accepted.routes {
        if &route.index != index || route.shard_ids.is_empty() {
            return Err(SubscriptionError::InvalidFrame);
        }
    }
    let mut cut = Vec::new();
    merge_cut(&mut cut, &accepted.accepted_cut, index)?;
    Ok(SubscriptionCheckpoint {
        index: index.clone(),
        routes: accepted.routes.clone(),
        cut,
    })
}

fn decode_frame(bytes: Vec<u8>) -> Result<SubscriptionFrame, SubscriptionError> {
    if bytes.is_empty() || bytes.len() > MAX_SUBSCRIPTION_MESSAGE_BYTES {
        return Err(SubscriptionError::InvalidFrame);
    }
    let frame: SubscriptionFrame = serde_json::from_slice(&bytes).map_err(|_| SubscriptionError::FrameCodec)?;
    if serde_json::to_vec(&frame).map_err(|_| SubscriptionError::FrameCodec)? != bytes {
        return Err(SubscriptionError::NonCanonicalFrame);
    }
    Ok(frame)
}

fn validate_event(event: &SubscriptionEvent, index: &Resource) -> Result<(), SubscriptionError> {
    if event.format_version != u32::from(LODESTONE_API_SUBSCRIPTION_EVENT_FORMAT_VERSION)
        || &event.index != index
        || event.event_id.validate().is_err()
        || event.recovery_era.validate().is_err()
        || event.writer_fence.recovery_era != event.recovery_era
        || parse_u64(&event.routing_revision.value).is_none()
        || parse_u64(&event.shard_id.value).is_none()
        || parse_u64(&event.sequence.value).is_none()
    {
        return Err(SubscriptionError::InvalidFrame);
    }
    Ok(())
}

fn cursor_from_event(event: &SubscriptionEvent) -> SubscriptionCursorEntry {
    SubscriptionCursorEntry {
        index: event.index.clone(),
        routing_revision: event.routing_revision.clone(),
        recovery_era: event.recovery_era.clone(),
        shard_id: event.shard_id.clone(),
        sequence: event.sequence.clone(),
    }
}

fn merge_cut(
    current: &mut Vec<SubscriptionCursorEntry>,
    observed: &[SubscriptionCursorEntry],
    index: &Resource,
) -> Result<(), SubscriptionError> {
    for cursor in observed {
        if &cursor.index != index {
            return Err(SubscriptionError::InvalidFrame);
        }
        upsert_cursor(current, cursor.clone())?;
    }
    Ok(())
}

fn upsert_cursor(
    current: &mut Vec<SubscriptionCursorEntry>,
    observed: SubscriptionCursorEntry,
) -> Result<(), SubscriptionError> {
    let key = CursorKey::from_cursor(&observed);
    let next = parse_u64(&observed.sequence.value).ok_or(SubscriptionError::InvalidFrame)?;
    if let Some(existing) = current.iter_mut().find(|cursor| CursorKey::from_cursor(cursor) == key) {
        let prior = parse_u64(&existing.sequence.value).ok_or(SubscriptionError::InvalidFrame)?;
        if next < prior {
            return Err(SubscriptionError::CursorRegression);
        }
        *existing = observed;
    } else {
        current.push(observed);
    }
    current.sort_by_key(CursorKey::from_cursor);
    Ok(())
}

fn cursor_for_key<'a>(cut: &'a [SubscriptionCursorEntry], key: &CursorKey) -> Option<&'a SubscriptionCursorEntry> {
    cut.iter().find(|cursor| &CursorKey::from_cursor(cursor) == key)
}

#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
struct CursorKey {
    routing_revision: String,
    recovery_era: String,
    shard_id: String,
}

impl CursorKey {
    fn from_cursor(cursor: &SubscriptionCursorEntry) -> Self {
        Self {
            routing_revision: cursor.routing_revision.value.clone(),
            recovery_era: cursor.recovery_era.value.clone(),
            shard_id: cursor.shard_id.value.clone(),
        }
    }
}

fn random_request_id() -> Result<String, SubscriptionError> {
    let mut bytes = [0_u8; 16];
    getrandom::fill(&mut bytes).map_err(|_| SubscriptionError::RandomUnavailable)?;
    if bytes == [0; 16] {
        return Err(SubscriptionError::RandomUnavailable);
    }
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut output = String::with_capacity(34);
    output.push_str("s-");
    for byte in bytes {
        output.push(char::from(HEX[usize::from(byte >> 4)]));
        output.push(char::from(HEX[usize::from(byte & 0x0f)]));
    }
    Ok(output)
}

fn reconnectable_status(status: &tonic::Status) -> bool {
    matches!(
        status.code(),
        Code::Cancelled | Code::Unknown | Code::DeadlineExceeded | Code::Aborted | Code::Internal | Code::Unavailable
    )
}

#[derive(Debug)]
pub enum SubscriptionError {
    Client(ClientError),
    ProfileNotVerified,
    InvalidStart,
    RequestCodec,
    RequestTooLarge,
    InvalidEndpoint,
    InvalidTls,
    InvalidMetadata,
    Connect,
    OpenDeadline,
    FrameDeadline,
    Cancelled,
    Status(tonic::Status),
    AcceptedAbsent,
    UnexpectedAccepted,
    FrameCodec,
    NonCanonicalFrame,
    InvalidFrame,
    CursorRegression,
    AuthorizationRegression,
    DuplicateIdentityMismatch,
    CheckpointMismatch,
    ReconnectExhausted,
    RandomUnavailable,
}

impl Display for SubscriptionError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::Client(_) => "Core subscription client dependency failed",
            Self::ProfileNotVerified => "Core subscription is refused until discovery and binding are verified",
            Self::InvalidStart => "Core subscription start is invalid",
            Self::RequestCodec => "Core subscription start could not be encoded",
            Self::RequestTooLarge => "Core subscription start exceeds its byte bound",
            Self::InvalidEndpoint => "Core subscription endpoint is invalid",
            Self::InvalidTls => "Core subscription TLS configuration is invalid",
            Self::InvalidMetadata => "Core subscription admission metadata is invalid",
            Self::Connect => "Core subscription connection failed",
            Self::OpenDeadline => "Core subscription open deadline elapsed",
            Self::FrameDeadline => "Core subscription frame deadline elapsed",
            Self::Cancelled => "Core subscription was cancelled",
            Self::Status(_) => "Core subscription transport returned a gRPC error",
            Self::AcceptedAbsent => "Core subscription did not begin with an accepted checkpoint",
            Self::UnexpectedAccepted => "Core subscription repeated its accepted checkpoint",
            Self::FrameCodec => "Core subscription frame is malformed",
            Self::NonCanonicalFrame => "Core subscription frame encoding is noncanonical",
            Self::InvalidFrame => "Core subscription frame violates its current semantic contract",
            Self::CursorRegression => "Core subscription cursor regressed",
            Self::AuthorizationRegression => "Core subscription authorization revision regressed",
            Self::DuplicateIdentityMismatch => "Core subscription reused a cursor with different event identity",
            Self::CheckpointMismatch => "Core subscription resume accepted a different checkpoint",
            Self::ReconnectExhausted => "Core subscription reconnect budget is exhausted",
            Self::RandomUnavailable => "secure subscription request identity generation is unavailable",
        })
    }
}

impl Error for SubscriptionError {}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn authorization_and_policy_statuses_are_terminal() {
        for code in [Code::Unauthenticated, Code::PermissionDenied, Code::FailedPrecondition] {
            assert!(!reconnectable_status(&tonic::Status::new(code, "terminal")));
        }
        for code in [Code::Unavailable, Code::DeadlineExceeded, Code::Internal] {
            assert!(reconnectable_status(&tonic::Status::new(code, "transient")));
        }
    }
}