v0.2.9-pre.009

crates/ksp-onchain-transport-lib/src/grpc_stream.rs

@@ -0,0 +1,584 @@
+// file: crates/ksp-onchain-transport-lib/src/grpc_stream.rs
+// version: 1
+
+use tonic_prost::prost::Message; // rust-rules: trait-import
+
+const AUTO_SUBSCRIBE_PING_ID: i32 = 1;
+const PATH_SUBSCRIBE: &str = "/geyser.Geyser/Subscribe";
+
+/// Safe lifecycle state of one standard Yellowstone bidirectional subscribe session.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub enum YellowstoneGrpcSubscribeState {
+    /// The bidirectional stream is active and accepts request mutations.
+    Active,
+    /// KSP has started a bounded graceful half-close.
+    Closing,
+    /// The stream ended normally, including a server half-close.
+    Closed,
+    /// The stream terminated because a transport, protocol, decoding or backpressure error occurred.
+    Failed,
+}
+
+/// Standard Solana Yellowstone bidirectional `Subscribe` session layered on one KSP-owned physical gRPC channel.
+///
+/// Request mutations and decoded updates use bounded Tokio channels. The raw Tonic stream and upstream protobuf messages remain private to Transport.
+pub struct SolanaYellowstoneGrpcSubscribeSession {
+    endpoint_name: std::string::String,
+    provider: crate::YellowstoneGrpcProviderName,
+    cluster: crate::YellowstoneGrpcClusterName,
+    request_tx: std::option::Option<tokio::sync::mpsc::Sender<yellowstone_grpc_proto::geyser::SubscribeRequest>>,
+    update_rx: tokio::sync::mpsc::Receiver<ksp_core_lib::Result<crate::YellowstoneSubscribeUpdate>>,
+    shutdown_tx: tokio::sync::watch::Sender<std::option::Option<tokio::time::Instant>>,
+    state_rx: tokio::sync::watch::Receiver<SubscribeActorState>,
+    task: tokio::task::JoinHandle<()>,
+    close_timeout: std::time::Duration,
+    max_outbound_message_size_bytes: usize,
+}
+
+impl SolanaYellowstoneGrpcSubscribeSession {
+    /// Returns the safe logical endpoint name.
+    #[must_use]
+    pub fn endpoint_name(&self) -> &str {
+        return self.endpoint_name.as_str();
+    }
+
+    /// Returns the open provider descriptor without exposing endpoint credentials.
+    #[must_use]
+    pub const fn provider(&self) -> &crate::YellowstoneGrpcProviderName {
+        return &self.provider;
+    }
+
+    /// Returns the open cluster descriptor.
+    #[must_use]
+    pub const fn cluster(&self) -> &crate::YellowstoneGrpcClusterName {
+        return &self.cluster;
+    }
+
+    /// Returns the current safe stream lifecycle state.
+    #[must_use]
+    pub fn state(&self) -> crate::YellowstoneGrpcSubscribeState {
+        return self.state_rx.borrow().public_state();
+    }
+
+    /// Queues one complete standard Yellowstone request mutation without waiting for network dispatch.
+    ///
+    /// The mutation is rejected synchronously when local validation fails, the encoded request exceeds the configured outbound bound, the bounded request
+    /// queue is full, or the session is no longer active. Accepted mutations preserve bounded-channel admission order.
+    pub fn try_update(&self, request: &crate::YellowstoneSubscribeRequest) -> ksp_core_lib::Result<()> {
+        if self.state() != crate::YellowstoneGrpcSubscribeState::Active {
+            return std::result::Result::Err(subscribe_session_error(
+                crate::ERROR_CODE_GRPC_SESSION_CLOSED,
+                "Yellowstone subscribe session is not active",
+                self.endpoint_name.as_str(),
+                self.provider.as_str(),
+                self.cluster.as_str(),
+            ));
+        }
+        let wire = match crate::yellowstone_subscribe_request_to_wire(request) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        if wire.encoded_len() > self.max_outbound_message_size_bytes {
+            return std::result::Result::Err(subscribe_session_error(
+                crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW,
+                "Yellowstone subscribe request exceeds the configured outbound message bound",
+                self.endpoint_name.as_str(),
+                self.provider.as_str(),
+                self.cluster.as_str(),
+            ));
+        }
+        let sender = match self.request_tx.as_ref() {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => {
+                return std::result::Result::Err(subscribe_session_error(
+                    crate::ERROR_CODE_GRPC_SESSION_CLOSED,
+                    "Yellowstone subscribe request channel is closed",
+                    self.endpoint_name.as_str(),
+                    self.provider.as_str(),
+                    self.cluster.as_str(),
+                ));
+            },
+        };
+        return match sender.try_send(wire) {
+            std::result::Result::Ok(()) => std::result::Result::Ok(()),
+            std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => std::result::Result::Err(subscribe_session_error(
+                crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW,
+                "Yellowstone subscribe request queue is full",
+                self.endpoint_name.as_str(),
+                self.provider.as_str(),
+                self.cluster.as_str(),
+            )),
+            std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => std::result::Result::Err(subscribe_session_error(
+                crate::ERROR_CODE_GRPC_SESSION_CLOSED,
+                "Yellowstone subscribe request queue is closed",
+                self.endpoint_name.as_str(),
+                self.provider.as_str(),
+                self.cluster.as_str(),
+            )),
+        };
+    }
+
+    /// Receives the next decoded standard Yellowstone update.
+    ///
+    /// A normal server half-close returns `Ok(None)`. Remote statuses, malformed updates and overflow terminate the session and surface a safe KSP error.
+    pub async fn next_update(&mut self) -> ksp_core_lib::Result<std::option::Option<crate::YellowstoneSubscribeUpdate>> {
+        return match self.update_rx.recv().await {
+            std::option::Option::Some(std::result::Result::Ok(update)) => std::result::Result::Ok(std::option::Option::Some(update)),
+            std::option::Option::Some(std::result::Result::Err(error)) => {
+                self.request_tx.take();
+                std::result::Result::Err(error)
+            },
+            std::option::Option::None => {
+                self.request_tx.take();
+                match *self.state_rx.borrow() {
+                    SubscribeActorState::Closed => std::result::Result::Ok(std::option::Option::None),
+                    SubscribeActorState::Failed(code) => std::result::Result::Err(subscribe_session_error(
+                        code,
+                        terminal_message(code),
+                        self.endpoint_name.as_str(),
+                        self.provider.as_str(),
+                        self.cluster.as_str(),
+                    )),
+                    SubscribeActorState::Active | SubscribeActorState::Closing => std::result::Result::Err(subscribe_session_error(
+                        crate::ERROR_CODE_GRPC_SESSION_CLOSED,
+                        "Yellowstone subscribe update channel ended before a terminal state was published",
+                        self.endpoint_name.as_str(),
+                        self.provider.as_str(),
+                        self.cluster.as_str(),
+                    )),
+                }
+            },
+        };
+    }
+
+    /// Gracefully half-closes the client request side and waits up to the configured close timeout for the server side to finish.
+    pub async fn close(mut self) -> ksp_core_lib::Result<()> {
+        let terminal_before_close = *self.state_rx.borrow();
+        self.request_tx.take();
+        if terminal_before_close == SubscribeActorState::Active {
+            let deadline = tokio::time::Instant::now() + self.close_timeout;
+            self.shutdown_tx.send_replace(std::option::Option::Some(deadline));
+        }
+        let deadline = tokio::time::Instant::now() + self.close_timeout;
+        match tokio::time::timeout_at(deadline, &mut self.task).await {
+            std::result::Result::Ok(join_result) => {
+                if join_result.is_err() {
+                    return std::result::Result::Err(subscribe_session_error(
+                        crate::ERROR_CODE_GRPC_SESSION_CLOSED,
+                        "Yellowstone subscribe actor terminated unexpectedly",
+                        self.endpoint_name.as_str(),
+                        self.provider.as_str(),
+                        self.cluster.as_str(),
+                    ));
+                }
+            },
+            std::result::Result::Err(_) => {
+                self.task.abort();
+                let _ = (&mut self.task).await;
+                return std::result::Result::Err(subscribe_session_error(
+                    crate::ERROR_CODE_TIMEOUT,
+                    "Yellowstone subscribe graceful shutdown exceeded the configured close timeout",
+                    self.endpoint_name.as_str(),
+                    self.provider.as_str(),
+                    self.cluster.as_str(),
+                ));
+            },
+        }
+        return match *self.state_rx.borrow() {
+            SubscribeActorState::Closed => std::result::Result::Ok(()),
+            SubscribeActorState::Failed(code) => std::result::Result::Err(subscribe_session_error(
+                code,
+                terminal_message(code),
+                self.endpoint_name.as_str(),
+                self.provider.as_str(),
+                self.cluster.as_str(),
+            )),
+            SubscribeActorState::Active | SubscribeActorState::Closing => std::result::Result::Err(subscribe_session_error(
+                crate::ERROR_CODE_GRPC_SESSION_CLOSED,
+                "Yellowstone subscribe actor ended without a terminal lifecycle state",
+                self.endpoint_name.as_str(),
+                self.provider.as_str(),
+                self.cluster.as_str(),
+            )),
+        };
+    }
+}
+
+impl std::fmt::Debug for SolanaYellowstoneGrpcSubscribeSession {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("SolanaYellowstoneGrpcSubscribeSession")
+            .field("endpoint_name", &self.endpoint_name)
+            .field("provider", &self.provider)
+            .field("cluster", &self.cluster)
+            .field("state", &self.state())
+            .field("request_queue_capacity", &self.request_tx.as_ref().map(|sender| return sender.capacity()))
+            .field("update_queue_capacity", &self.update_rx.capacity())
+            .finish();
+    }
+}
+
+impl Drop for SolanaYellowstoneGrpcSubscribeSession {
+    fn drop(&mut self) {
+        self.request_tx.take();
+        if self.state() == crate::YellowstoneGrpcSubscribeState::Active {
+            self.shutdown_tx.send_replace(std::option::Option::Some(tokio::time::Instant::now() + self.close_timeout));
+        }
+    }
+}
+
+/// Opens one standard Yellowstone `Subscribe` stream from a validated KSP channel.
+pub(crate) async fn open_yellowstone_subscribe_session(
+    channel: tonic::transport::Channel,
+    metadata: std::vec::Vec<crate::YellowstoneGrpcMetadataEntry>,
+    settings: crate::YellowstoneGrpcSessionSettings,
+    endpoint_name: std::string::String,
+    provider: crate::YellowstoneGrpcProviderName,
+    cluster: crate::YellowstoneGrpcClusterName,
+    initial_request: crate::YellowstoneSubscribeRequest,
+) -> ksp_core_lib::Result<crate::SolanaYellowstoneGrpcSubscribeSession> {
+    let initial_wire = match crate::yellowstone_subscribe_request_to_wire(&initial_request) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    if initial_wire.encoded_len() > settings.max_outbound_message_size_bytes() {
+        return std::result::Result::Err(subscribe_session_error(
+            crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW,
+            "initial Yellowstone subscribe request exceeds the configured outbound message bound",
+            endpoint_name.as_str(),
+            provider.as_str(),
+            cluster.as_str(),
+        ));
+    }
+    let (request_tx, request_rx) = tokio::sync::mpsc::channel(settings.request_channel_capacity());
+    if request_tx.try_send(initial_wire).is_err() {
+        return std::result::Result::Err(subscribe_session_error(
+            crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW,
+            "initial Yellowstone subscribe request cannot enter the bounded request queue",
+            endpoint_name.as_str(),
+            provider.as_str(),
+            cluster.as_str(),
+        ));
+    }
+    let path = match PATH_SUBSCRIBE.parse::<http::uri::PathAndQuery>() {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(subscribe_session_error(
+                crate::ERROR_CODE_GRPC_CHANNEL_FAILED,
+                "internal Yellowstone Subscribe method path is invalid",
+                endpoint_name.as_str(),
+                provider.as_str(),
+                cluster.as_str(),
+            ));
+        },
+    };
+    let mut grpc = tonic::client::Grpc::new(channel)
+        .max_decoding_message_size(settings.max_inbound_message_size_bytes())
+        .max_encoding_message_size(settings.max_outbound_message_size_bytes());
+    let open_future = async {
+        if grpc.ready().await.is_err() {
+            return std::result::Result::Err(subscribe_session_error(
+                crate::ERROR_CODE_GRPC_CHANNEL_FAILED,
+                "Yellowstone gRPC channel is not ready for Subscribe dispatch",
+                endpoint_name.as_str(),
+                provider.as_str(),
+                cluster.as_str(),
+            ));
+        }
+        let mut request = tonic::Request::new(MpscStream::new(request_rx));
+        for entry in &metadata {
+            if let std::result::Result::Err(error) = entry.append_to(request.metadata_mut()) {
+                return std::result::Result::Err(error);
+            }
+        }
+        let response = match grpc
+            .streaming(
+                request,
+                path,
+                tonic_prost::ProstCodec::<yellowstone_grpc_proto::geyser::SubscribeRequest, yellowstone_grpc_proto::geyser::SubscribeUpdate>::default(),
+            )
+            .await
+        {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(status) => {
+                return std::result::Result::Err(stream_status_error("SubscribeOpen", status, endpoint_name.as_str(), provider.as_str(), cluster.as_str()));
+            },
+        };
+        return std::result::Result::Ok(response.into_inner());
+    };
+    let incoming = match tokio::time::timeout(settings.connect_timeout(), open_future).await {
+        std::result::Result::Ok(std::result::Result::Ok(value)) => value,
+        std::result::Result::Ok(std::result::Result::Err(error)) => return std::result::Result::Err(error),
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(subscribe_session_error(
+                crate::ERROR_CODE_TIMEOUT,
+                "Yellowstone Subscribe stream opening exceeded the configured connection timeout",
+                endpoint_name.as_str(),
+                provider.as_str(),
+                cluster.as_str(),
+            ));
+        },
+    };
+    let (update_tx, update_rx) = tokio::sync::mpsc::channel(settings.update_channel_capacity());
+    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(std::option::Option::None::<tokio::time::Instant>);
+    let (state_tx, state_rx) = tokio::sync::watch::channel(SubscribeActorState::Active);
+    let actor_request_tx = request_tx.clone();
+    let actor_endpoint_name = endpoint_name.clone();
+    let actor_provider = provider.clone();
+    let actor_cluster = cluster.clone();
+    let close_timeout = settings.close_timeout();
+    let max_outbound_message_size_bytes = settings.max_outbound_message_size_bytes();
+    let task = tokio::spawn(run_subscribe_actor(
+        actor_endpoint_name,
+        actor_provider,
+        actor_cluster,
+        incoming,
+        actor_request_tx,
+        update_tx,
+        shutdown_rx,
+        state_tx,
+        max_outbound_message_size_bytes,
+    ));
+    return std::result::Result::Ok(crate::SolanaYellowstoneGrpcSubscribeSession {
+        endpoint_name,
+        provider,
+        cluster,
+        request_tx: std::option::Option::Some(request_tx),
+        update_rx,
+        shutdown_tx,
+        state_rx,
+        task,
+        close_timeout,
+        max_outbound_message_size_bytes,
+    });
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum SubscribeActorState {
+    Active,
+    Closing,
+    Closed,
+    Failed(ksp_core_lib::ErrorCode),
+}
+
+impl SubscribeActorState {
+    const fn public_state(self) -> crate::YellowstoneGrpcSubscribeState {
+        return match self {
+            Self::Active => crate::YellowstoneGrpcSubscribeState::Active,
+            Self::Closing => crate::YellowstoneGrpcSubscribeState::Closing,
+            Self::Closed => crate::YellowstoneGrpcSubscribeState::Closed,
+            Self::Failed(_) => crate::YellowstoneGrpcSubscribeState::Failed,
+        };
+    }
+}
+
+struct MpscStream<T> {
+    receiver: tokio::sync::mpsc::Receiver<T>,
+}
+
+impl<T> MpscStream<T> {
+    const fn new(receiver: tokio::sync::mpsc::Receiver<T>) -> Self {
+        return Self { receiver };
+    }
+}
+
+impl<T> futures_util::Stream for MpscStream<T> {
+    type Item = T;
+
+    fn poll_next(self: std::pin::Pin<&mut Self>, context: &mut std::task::Context<'_>) -> std::task::Poll<std::option::Option<Self::Item>> {
+        return self.get_mut().receiver.poll_recv(context);
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn run_subscribe_actor(
+    endpoint_name: std::string::String,
+    provider: crate::YellowstoneGrpcProviderName,
+    cluster: crate::YellowstoneGrpcClusterName,
+    mut incoming: tonic::Streaming<yellowstone_grpc_proto::geyser::SubscribeUpdate>,
+    request_tx: tokio::sync::mpsc::Sender<yellowstone_grpc_proto::geyser::SubscribeRequest>,
+    update_tx: tokio::sync::mpsc::Sender<ksp_core_lib::Result<crate::YellowstoneSubscribeUpdate>>,
+    mut shutdown_rx: tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
+    state_tx: tokio::sync::watch::Sender<SubscribeActorState>,
+    max_outbound_message_size_bytes: usize,
+) {
+    let mut request_tx = std::option::Option::Some(request_tx);
+    loop {
+        tokio::select! {
+            shutdown_changed = shutdown_rx.changed() => {
+                let deadline = shutdown_deadline(&shutdown_rx, shutdown_changed);
+                state_tx.send_replace(SubscribeActorState::Closing);
+                request_tx.take();
+                finish_client_half_close(&endpoint_name, &provider, &cluster, &mut incoming, deadline, &state_tx).await;
+                return;
+            }
+            message = incoming.message() => {
+                match message {
+                    std::result::Result::Ok(std::option::Option::Some(wire)) => {
+                        let update = match crate::yellowstone_subscribe_update_from_wire(wire) {
+                            std::result::Result::Ok(value) => value,
+                            std::result::Result::Err(error) => {
+                                let _ = update_tx.try_send(std::result::Result::Err(error));
+                                state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_INVALID_RESPONSE));
+                                request_tx.take();
+                                return;
+                            },
+                        };
+                        if matches!(&update, crate::YellowstoneSubscribeUpdate::Ping(_)) {
+                            let ping_wire = ping_request_wire();
+                            if ping_wire.encoded_len() > max_outbound_message_size_bytes {
+                                state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW));
+                                request_tx.take();
+                                return;
+                            }
+                            let sender = match request_tx.as_ref() {
+                                std::option::Option::Some(value) => value,
+                                std::option::Option::None => {
+                                    state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_SESSION_CLOSED));
+                                    return;
+                                },
+                            };
+                            if sender.try_send(ping_wire).is_err() {
+                                state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW));
+                                request_tx.take();
+                                return;
+                            }
+                        }
+                        match update_tx.try_send(std::result::Result::Ok(update)) {
+                            std::result::Result::Ok(()) => {},
+                            std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
+                                ksp_logging_lib::warn!(
+                                    target: crate::TRACING_TARGET,
+                                    endpoint_name = endpoint_name.as_str(),
+                                    provider = provider.as_str(),
+                                    cluster = cluster.as_str(),
+                                    "Yellowstone subscribe update queue overflowed"
+                                );
+                                state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW));
+                                request_tx.take();
+                                return;
+                            },
+                            std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
+                                state_tx.send_replace(SubscribeActorState::Closed);
+                                request_tx.take();
+                                return;
+                            },
+                        }
+                    },
+                    std::result::Result::Ok(std::option::Option::None) => {
+                        state_tx.send_replace(SubscribeActorState::Closed);
+                        request_tx.take();
+                        return;
+                    },
+                    std::result::Result::Err(status) => {
+                        let error = stream_status_error("Subscribe", status, endpoint_name.as_str(), provider.as_str(), cluster.as_str());
+                        let _ = update_tx.try_send(std::result::Result::Err(error));
+                        state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_STATUS));
+                        request_tx.take();
+                        return;
+                    },
+                }
+            }
+        }
+    }
+}
+
+async fn finish_client_half_close(
+    endpoint_name: &str,
+    provider: &crate::YellowstoneGrpcProviderName,
+    cluster: &crate::YellowstoneGrpcClusterName,
+    incoming: &mut tonic::Streaming<yellowstone_grpc_proto::geyser::SubscribeUpdate>,
+    deadline: tokio::time::Instant,
+    state_tx: &tokio::sync::watch::Sender<SubscribeActorState>,
+) {
+    loop {
+        let message = tokio::time::timeout_at(deadline, incoming.message()).await;
+        match message {
+            std::result::Result::Ok(std::result::Result::Ok(std::option::Option::Some(_))) => {},
+            std::result::Result::Ok(std::result::Result::Ok(std::option::Option::None)) => {
+                state_tx.send_replace(SubscribeActorState::Closed);
+                return;
+            },
+            std::result::Result::Ok(std::result::Result::Err(status)) => {
+                let code = status.code().to_string();
+                ksp_logging_lib::warn!(
+                    target: crate::TRACING_TARGET,
+                    endpoint_name,
+                    provider = provider.as_str(),
+                    cluster = cluster.as_str(),
+                    grpc_code = code.as_str(),
+                    "Yellowstone subscribe endpoint returned a status during graceful shutdown"
+                );
+                state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_STATUS));
+                return;
+            },
+            std::result::Result::Err(_) => {
+                state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_TIMEOUT));
+                return;
+            },
+        }
+    }
+}
+
+fn shutdown_deadline(
+    shutdown_rx: &tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
+    shutdown_changed: std::result::Result<(), tokio::sync::watch::error::RecvError>,
+) -> tokio::time::Instant {
+    return match shutdown_changed {
+        std::result::Result::Ok(()) => match *shutdown_rx.borrow() {
+            std::option::Option::Some(deadline) => deadline,
+            std::option::Option::None => tokio::time::Instant::now(),
+        },
+        std::result::Result::Err(_) => tokio::time::Instant::now(),
+    };
+}
+
+fn ping_request_wire() -> yellowstone_grpc_proto::geyser::SubscribeRequest {
+    return yellowstone_grpc_proto::geyser::SubscribeRequest {
+        ping: std::option::Option::Some(yellowstone_grpc_proto::geyser::SubscribeRequestPing { id: AUTO_SUBSCRIBE_PING_ID }),
+        ..std::default::Default::default()
+    };
+}
+
+fn stream_status_error(operation: &'static str, status: tonic::Status, endpoint_name: &str, provider: &str, cluster: &str) -> ksp_core_lib::Error {
+    let code = status.code().to_string();
+    ksp_logging_lib::warn!(
+        target: crate::TRACING_TARGET,
+        endpoint_name,
+        provider,
+        cluster,
+        grpc_operation = operation,
+        grpc_code = code.as_str(),
+        "Yellowstone subscribe endpoint returned a gRPC status"
+    );
+    return subscribe_session_error(crate::ERROR_CODE_GRPC_STATUS, "Yellowstone subscribe endpoint returned a gRPC status", endpoint_name, provider, cluster)
+        .with_context("grpc_operation", operation)
+        .with_context("grpc_code", code);
+}
+
+fn subscribe_session_error(code: ksp_core_lib::ErrorCode, message: &'static str, endpoint_name: &str, provider: &str, cluster: &str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(code, message)
+        .with_context("endpoint_name", endpoint_name)
+        .with_context("provider", provider)
+        .with_context("cluster", cluster);
+}
+
+fn terminal_message(code: ksp_core_lib::ErrorCode) -> &'static str {
+    if code == crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW {
+        return "Yellowstone subscribe session terminated because a bounded queue overflowed";
+    }
+    if code == crate::ERROR_CODE_GRPC_STATUS {
+        return "Yellowstone subscribe session terminated after a remote gRPC status";
+    }
+    if code == crate::ERROR_CODE_INVALID_RESPONSE {
+        return "Yellowstone subscribe session terminated after an invalid update";
+    }
+    if code == crate::ERROR_CODE_TIMEOUT {
+        return "Yellowstone subscribe session exceeded a configured deadline";
+    }
+    return "Yellowstone subscribe session is closed";
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/grpc_stream.rs"]
+mod tests;