use std::sync::{Arc, Mutex};
use std::time::Duration;
use satd_events_proto::v1 as pb;
use satd_events_proto::v1::node_event_stream_client::NodeEventStreamClient;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::metadata::{Ascii, MetadataMap, MetadataValue};
use tonic::transport::{Channel, Endpoint};
use crate::compat::{self, Compat};
use crate::error::StreamError;
use crate::event::{Cursor, Event};
const CONTROL_BUFFER: usize = 16;
pub struct Categories;
impl Categories {
pub const ALL: u32 = 0;
pub const MEMPOOL: u32 = 1;
pub const CHAIN: u32 = 2;
pub const HEARTBEAT: u32 = 4;
pub const TWEAKS: u32 = 8;
pub const STATUS: u32 = 16;
}
#[derive(Debug, Clone, Default)]
pub struct SubscribeOptions {
pub categories: u32,
pub from_cursor: Option<Cursor>,
pub since_seq: Option<u64>,
pub tweak_dust_limit: Option<u64>,
pub tweaks_only: bool,
pub mempool_tweaks: bool,
pub tweak_outputs: bool,
pub tweak_unspent_only: bool,
}
impl SubscribeOptions {
#[allow(clippy::needless_update)]
fn into_request(self) -> pb::SubscribeRequest {
pb::SubscribeRequest {
categories: self.categories,
since_seq: self.since_seq,
from_cursor: self.from_cursor,
tweak_dust_limit: self.tweak_dust_limit,
tweaks_only: self.tweaks_only.then_some(true),
mempool_tweaks: self.mempool_tweaks.then_some(true),
tweak_outputs: self.tweak_outputs.then_some(true),
tweak_unspent_only: self.tweak_unspent_only.then_some(true),
..Default::default()
}
}
}
pub const MAX_SP_LABELS_PER_TARGET: usize = 16;
pub const MAX_SP_TARGETS_PER_CONNECTION: usize = 16;
#[derive(Clone)]
pub struct SilentPaymentTarget {
pub scan_secret: [u8; 32],
pub spend_pubkey: [u8; 33],
pub labels: Vec<u32>,
}
impl core::fmt::Debug for SilentPaymentTarget {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("SilentPaymentTarget")
.field("scan_secret", &"<redacted>")
.field("spend_pubkey", &hex_lower(&self.spend_pubkey))
.field("labels", &self.labels)
.finish()
}
}
impl Drop for SilentPaymentTarget {
fn drop(&mut self) {
use zeroize::Zeroize;
self.scan_secret.zeroize();
}
}
impl SilentPaymentTarget {
pub(crate) fn to_proto(&self) -> pb::SilentPaymentTarget {
pb::SilentPaymentTarget {
scan_secret: self.scan_secret.to_vec(),
spend_pubkey: self.spend_pubkey.to_vec(),
labels: self.labels.clone(),
}
}
#[cfg(feature = "bitcoin")]
pub fn scan_pubkey(&self) -> Result<[u8; 33], StreamError> {
use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
let sk = SecretKey::from_slice(&self.scan_secret)
.map_err(|_| StreamError::InvalidArgument("scan_secret is not a valid scalar".into()))?;
let secp = Secp256k1::signing_only();
Ok(PublicKey::from_secret_key(&secp, &sk).serialize())
}
#[cfg(feature = "bitcoin")]
pub fn validate(&self) -> Result<[u8; 33], StreamError> {
let id = self.scan_pubkey()?; bitcoin::secp256k1::PublicKey::from_slice(&self.spend_pubkey)
.map_err(|_| StreamError::InvalidArgument("spend_pubkey is not a valid point".into()))?;
let mut labels = self.labels.clone();
labels.sort_unstable();
labels.dedup();
if labels.len() > MAX_SP_LABELS_PER_TARGET {
return Err(StreamError::InvalidArgument(format!(
"too many silent-payment labels: {} distinct (max {MAX_SP_LABELS_PER_TARGET})",
labels.len()
)));
}
Ok(id)
}
}
fn hex_lower(bytes: &[u8]) -> String {
let mut s = String::with_capacity(bytes.len() * 2);
for b in bytes {
s.push_str(&format!("{b:02x}"));
}
s
}
pub struct EventStream {
inner: tonic::Streaming<pb::NodeEvent>,
last_cursor: Option<Cursor>,
schema_checked: bool,
}
impl EventStream {
pub async fn message(&mut self) -> Result<Option<Event>, StreamError> {
match self.inner.message().await {
Ok(Some(ev)) => {
if !self.schema_checked {
self.schema_checked = true;
if ev.schema_version != satd_events_proto::SCHEMA_VERSION {
return Err(StreamError::SchemaMismatch {
node: ev.schema_version,
sdk: satd_events_proto::SCHEMA_VERSION,
});
}
}
if let Some(c) = ev.cursor {
self.last_cursor = Some(c);
}
Ok(Some(Event::from(ev)))
}
Ok(None) => Ok(None),
Err(status) => Err(StreamError::from_status(status)),
}
}
pub fn cursor(&self) -> Option<&Cursor> {
self.last_cursor.as_ref()
}
fn new(inner: tonic::Streaming<pb::NodeEvent>) -> Self {
EventStream { inner, last_cursor: None, schema_checked: false }
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum AutoClose {
#[default]
Never,
AtDepth(u32),
}
impl AutoClose {
fn depth(self) -> u32 {
match self {
AutoClose::Never => 0,
AutoClose::AtDepth(d) => d,
}
}
}
pub struct WatchHandle {
tx: mpsc::Sender<pb::SubscribeControl>,
}
impl WatchHandle {
pub async fn send_control(&self, ctrl: pb::SubscribeControl) -> Result<(), StreamError> {
self.tx.send(ctrl).await.map_err(|_| StreamError::ControlClosed)
}
async fn send_msg(&self, msg: pb::subscribe_control::Msg) -> Result<(), StreamError> {
self.send_control(pb::SubscribeControl { msg: Some(msg) }).await
}
pub async fn add_scripts(
&self,
items: impl IntoIterator<Item = ([u8; 32], Option<u64>)>,
) -> Result<(), StreamError> {
let items: Vec<_> = items.into_iter().collect();
if items.is_empty() {
return Ok(());
}
let scripthashes = items.iter().map(|(h, _)| h.to_vec()).collect();
let min_values = if items.iter().any(|(_, f)| f.is_some()) {
items.iter().map(|(_, f)| f.unwrap_or(0)).collect()
} else {
Vec::new()
};
self.send_msg(pb::subscribe_control::Msg::AddScripts(pb::AddScripts {
scripthashes,
min_values,
}))
.await
}
pub async fn remove_scripts(
&self,
scripthashes: impl IntoIterator<Item = [u8; 32]>,
) -> Result<(), StreamError> {
let scripthashes: Vec<Vec<u8>> = scripthashes.into_iter().map(|h| h.to_vec()).collect();
if scripthashes.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::RemoveScripts(pb::RemoveScripts {
scripthashes,
}))
.await
}
pub async fn add_outpoints(
&self,
outpoints: impl IntoIterator<Item = ([u8; 32], u32)>,
) -> Result<(), StreamError> {
let outpoints: Vec<pb::Outpoint> = outpoints
.into_iter()
.map(|(txid, vout)| pb::Outpoint { txid: txid.to_vec(), vout })
.collect();
if outpoints.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::AddOutpoints(pb::AddOutpoints { outpoints }))
.await
}
pub async fn remove_outpoints(
&self,
outpoints: impl IntoIterator<Item = ([u8; 32], u32)>,
) -> Result<(), StreamError> {
let outpoints: Vec<pb::Outpoint> = outpoints
.into_iter()
.map(|(txid, vout)| pb::Outpoint { txid: txid.to_vec(), vout })
.collect();
if outpoints.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::RemoveOutpoints(pb::RemoveOutpoints {
outpoints,
}))
.await
}
pub async fn add_tx_lifecycle(
&self,
txids: impl IntoIterator<Item = [u8; 32]>,
auto_close: AutoClose,
) -> Result<(), StreamError> {
let txids: Vec<Vec<u8>> = txids.into_iter().map(|t| t.to_vec()).collect();
if txids.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::AddTransactions(pb::AddTransactions {
txids,
min_depths: Vec::new(),
auto_close_depth: auto_close.depth(),
}))
.await
}
pub async fn remove_tx_lifecycle(
&self,
txids: impl IntoIterator<Item = [u8; 32]>,
) -> Result<(), StreamError> {
let txids: Vec<Vec<u8>> = txids.into_iter().map(|t| t.to_vec()).collect();
if txids.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::RemoveTransactions(pb::RemoveTransactions {
txids,
min_depths: Vec::new(),
}))
.await
}
pub async fn add_depth_alarms(
&self,
txids: impl IntoIterator<Item = [u8; 32]>,
depths: impl IntoIterator<Item = u32>,
) -> Result<(), StreamError> {
let txids: Vec<Vec<u8>> = txids.into_iter().map(|t| t.to_vec()).collect();
let min_depths: Vec<u32> = depths.into_iter().filter(|d| *d >= 1).collect();
if txids.is_empty() || min_depths.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::AddTransactions(pb::AddTransactions {
txids,
min_depths,
auto_close_depth: 0,
}))
.await
}
pub async fn remove_depth_alarms(
&self,
txids: impl IntoIterator<Item = [u8; 32]>,
depths: impl IntoIterator<Item = u32>,
) -> Result<(), StreamError> {
let txids: Vec<Vec<u8>> = txids.into_iter().map(|t| t.to_vec()).collect();
let min_depths: Vec<u32> = depths.into_iter().filter(|d| *d >= 1).collect();
if txids.is_empty() || min_depths.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::RemoveTransactions(pb::RemoveTransactions {
txids,
min_depths,
}))
.await
}
pub async fn add_descriptor(
&self,
descriptor: impl Into<String>,
gap_limit: u32,
start: u32,
) -> Result<(), StreamError> {
self.send_msg(pb::subscribe_control::Msg::AddDescriptor(pb::AddDescriptor {
descriptor: descriptor.into(),
gap_limit,
start,
}))
.await
}
pub async fn remove_descriptor(
&self,
descriptor: impl Into<String>,
) -> Result<(), StreamError> {
self.send_msg(pb::subscribe_control::Msg::RemoveDescriptor(pb::RemoveDescriptor {
descriptor: descriptor.into(),
}))
.await
}
pub async fn add_script_prefixes(
&self,
prefixes: impl IntoIterator<Item = (Vec<u8>, u32)>,
) -> Result<(), StreamError> {
let prefixes: Vec<pb::ScriptPrefix> = prefixes
.into_iter()
.map(|(prefix, bits)| validate_prefix(prefix, bits))
.collect::<Result<_, _>>()?;
if prefixes.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::AddScriptPrefixes(pb::AddScriptPrefixes {
prefixes,
}))
.await
}
pub async fn remove_script_prefixes(
&self,
prefixes: impl IntoIterator<Item = (Vec<u8>, u32)>,
) -> Result<(), StreamError> {
let prefixes: Vec<pb::ScriptPrefix> = prefixes
.into_iter()
.map(|(prefix, bits)| validate_prefix(prefix, bits))
.collect::<Result<_, _>>()?;
if prefixes.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::RemoveScriptPrefixes(
pb::RemoveScriptPrefixes { prefixes },
))
.await
}
pub async fn add_silent_payments(
&self,
targets: impl IntoIterator<Item = SilentPaymentTarget>,
) -> Result<(), StreamError> {
let targets: Vec<SilentPaymentTarget> = targets.into_iter().collect();
#[cfg(feature = "bitcoin")]
for t in &targets {
t.validate()?;
}
let targets: Vec<pb::SilentPaymentTarget> = targets.iter().map(|t| t.to_proto()).collect();
if targets.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::AddSilentPayments(
pb::AddSilentPayments { targets },
))
.await
}
pub async fn remove_silent_payments(
&self,
scan_pubkeys: impl IntoIterator<Item = [u8; 33]>,
) -> Result<(), StreamError> {
let scan_pubkeys: Vec<Vec<u8>> = scan_pubkeys.into_iter().map(|k| k.to_vec()).collect();
if scan_pubkeys.is_empty() {
return Ok(());
}
self.send_msg(pb::subscribe_control::Msg::RemoveSilentPayments(
pb::RemoveSilentPayments { scan_pubkeys },
))
.await
}
pub async fn set_categories(&self, categories: u32) -> Result<(), StreamError> {
self.send_msg(pb::subscribe_control::Msg::SetCategories(pb::SetCategories {
categories,
}))
.await
}
pub async fn set_watch_options(&self, include_raw_tx: bool) -> Result<(), StreamError> {
self.send_msg(pb::subscribe_control::Msg::SetWatchOptions(
pb::SetWatchOptions { include_raw_tx },
))
.await
}
pub async fn set_cursor(&self, cursor: Cursor) -> Result<(), StreamError> {
self.send_msg(pb::subscribe_control::Msg::SetCursor(pb::SetCursor {
cursor: Some(cursor),
}))
.await
}
pub async fn rescan(&self, from_height: u32, to_height: u32) -> Result<(), StreamError> {
self.send_msg(pb::subscribe_control::Msg::RescanBlocks(pb::RescanBlocks {
from_height,
to_height,
}))
.await
}
}
pub(crate) const MAX_PREFIX_BITS: u32 = 32;
pub(crate) fn validate_prefix(
prefix: Vec<u8>,
bits: u32,
) -> Result<pb::ScriptPrefix, StreamError> {
if !(1..=MAX_PREFIX_BITS).contains(&bits) {
return Err(StreamError::InvalidArgument(format!(
"prefix bits {bits} out of range 1..={MAX_PREFIX_BITS}"
)));
}
let want = bits.div_ceil(8) as usize;
if prefix.len() != want {
return Err(StreamError::InvalidArgument(format!(
"prefix for {bits} bits must be {want} bytes, got {}",
prefix.len()
)));
}
Ok(pb::ScriptPrefix { prefix, bits })
}
#[cfg_attr(not(feature = "tls"), allow(dead_code))]
fn endpoint_is_https(endpoint: &str) -> bool {
endpoint
.split_once("://")
.is_some_and(|(scheme, _)| scheme.eq_ignore_ascii_case("https"))
}
#[cfg(feature = "tls")]
fn endpoint_is_encrypted(endpoint: &str) -> bool {
endpoint_is_https(endpoint)
}
#[cfg(not(feature = "tls"))]
fn endpoint_is_encrypted(_endpoint: &str) -> bool {
false
}
fn redact_endpoint_userinfo(endpoint: &str) -> String {
let Some((scheme, rest)) = endpoint.split_once("://") else {
return match endpoint.rsplit_once('@') {
Some((_, host)) => format!("***@{host}"),
None => endpoint.to_string(),
};
};
let (authority, path) = match rest.find('/') {
Some(i) => rest.split_at(i),
None => (rest, ""),
};
match authority.rsplit_once('@') {
Some((_, host)) => format!("{scheme}://***@{host}{path}"),
None => endpoint.to_string(),
}
}
#[cfg(feature = "tls")]
fn insecure_credential_reason(endpoint: &str) -> String {
redact_endpoint_userinfo(endpoint)
}
#[cfg(not(feature = "tls"))]
fn insecure_credential_reason(endpoint: &str) -> String {
format!(
"{} — this build of satd-events-client has the `tls` feature disabled, \
so no endpoint is encrypted here, https:// included",
redact_endpoint_userinfo(endpoint)
)
}
#[cfg(feature = "tls")]
#[derive(Clone, Default)]
struct TlsSettings {
ca_pem: Option<Vec<u8>>,
identity: Option<(Vec<u8>, Vec<u8>)>,
domain: Option<String>,
}
#[derive(Clone)]
pub struct StreamClientBuilder {
endpoint: String,
token: Option<String>,
allow_insecure_token: bool,
keepalive: Option<(Duration, Duration)>,
allow_old_node: bool,
#[cfg(feature = "tls")]
tls: Option<TlsSettings>,
}
impl std::fmt::Debug for StreamClientBuilder {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let mut d = f.debug_struct("StreamClientBuilder");
d.field("endpoint", &self.endpoint)
.field("token", &self.token.as_ref().map(|_| "<redacted>"))
.field("allow_insecure_token", &self.allow_insecure_token)
.field("keepalive", &self.keepalive)
.field("allow_old_node", &self.allow_old_node);
#[cfg(feature = "tls")]
d.field("tls", &self.tls.is_some());
d.finish()
}
}
impl StreamClientBuilder {
pub fn bearer_token(mut self, token: impl Into<String>) -> Self {
self.token = Some(token.into());
self.allow_insecure_token = false;
self
}
pub fn insecure_bearer_token(mut self, token: impl Into<String>) -> Self {
self.token = Some(token.into());
self.allow_insecure_token = true;
self
}
pub fn keepalive(mut self, interval_secs: u64, timeout_secs: u64) -> Self {
self.keepalive =
Some((Duration::from_secs(interval_secs), Duration::from_secs(timeout_secs)));
self
}
pub fn keepalive_default(self) -> Self {
self.keepalive(30, 20)
}
pub fn allow_old_node(mut self) -> Self {
self.allow_old_node = true;
self
}
#[cfg(feature = "tls")]
#[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
pub fn tls(mut self) -> Self {
self.tls.get_or_insert_with(TlsSettings::default);
self
}
#[cfg(feature = "tls")]
#[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
pub fn tls_ca_pem(mut self, pem: impl Into<Vec<u8>>) -> Self {
self.tls.get_or_insert_with(TlsSettings::default).ca_pem = Some(pem.into());
self
}
#[cfg(feature = "tls")]
#[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
pub fn tls_client_identity(
mut self,
cert_pem: impl Into<Vec<u8>>,
key_pem: impl Into<Vec<u8>>,
) -> Self {
self.tls.get_or_insert_with(TlsSettings::default).identity =
Some((cert_pem.into(), key_pem.into()));
self
}
#[cfg(feature = "tls")]
#[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
pub fn tls_domain(mut self, domain: impl Into<String>) -> Self {
self.tls.get_or_insert_with(TlsSettings::default).domain = Some(domain.into());
self
}
pub async fn connect(self) -> Result<StreamClient, StreamError> {
let has_token = self.token.is_some();
let auth = match self.token {
Some(t) => Some(
format!("Bearer {t}")
.parse::<MetadataValue<Ascii>>()
.map_err(|_| StreamError::InvalidToken)?,
),
None => None,
};
#[cfg(feature = "tls")]
if self.tls.is_some() && !endpoint_is_https(&self.endpoint) {
return Err(StreamError::InvalidEndpoint(
"TLS was requested (tls / tls_ca_pem / tls_client_identity / tls_domain) but \
the endpoint scheme is not https:// — refusing to connect in cleartext; use \
an https:// endpoint"
.to_string(),
));
}
if has_token && !self.allow_insecure_token && !endpoint_is_encrypted(&self.endpoint) {
return Err(StreamError::InsecureCredential(insecure_credential_reason(
&self.endpoint,
)));
}
let mut endpoint = Endpoint::from_shared(self.endpoint)
.map_err(|e| StreamError::InvalidEndpoint(e.to_string()))?;
if let Some((interval, timeout)) = self.keepalive {
endpoint = endpoint
.http2_keep_alive_interval(interval)
.keep_alive_timeout(timeout)
.keep_alive_while_idle(true);
}
#[cfg(feature = "tls")]
if let Some(tls) = self.tls {
ensure_crypto_provider();
let mut cfg = tonic::transport::ClientTlsConfig::new();
cfg = match tls.ca_pem {
Some(ca) => cfg.ca_certificate(tonic::transport::Certificate::from_pem(ca)),
None => cfg.with_webpki_roots(),
};
if let Some((cert, key)) = tls.identity {
cfg = cfg.identity(tonic::transport::Identity::from_pem(cert, key));
}
if let Some(domain) = tls.domain {
cfg = cfg.domain_name(domain);
}
endpoint = endpoint.tls_config(cfg)?;
}
let channel = endpoint.connect().await?;
Ok(StreamClient {
inner: NodeEventStreamClient::new(channel),
auth,
allow_old_node: self.allow_old_node,
compat: Arc::default(),
})
}
}
#[cfg(feature = "tls")]
fn ensure_crypto_provider() {
use std::sync::Once;
static ONCE: Once = Once::new();
ONCE.call_once(|| {
let _ = tokio_rustls::rustls::crypto::ring::default_provider().install_default();
});
}
#[derive(Clone)]
pub struct StreamClient {
inner: NodeEventStreamClient<Channel>,
auth: Option<MetadataValue<Ascii>>,
allow_old_node: bool,
compat: Arc<Mutex<CompatState>>,
}
#[derive(Debug, Default)]
struct CompatState {
node_version: Option<String>,
warned_for: Option<Option<String>>,
}
impl std::fmt::Debug for StreamClient {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("StreamClient")
.field("auth", &self.auth.as_ref().map(|_| "<redacted>"))
.finish_non_exhaustive()
}
}
impl StreamClient {
pub fn builder(endpoint: impl Into<String>) -> StreamClientBuilder {
StreamClientBuilder {
endpoint: endpoint.into(),
token: None,
allow_insecure_token: false,
keepalive: None,
allow_old_node: false,
#[cfg(feature = "tls")]
tls: None,
}
}
pub fn node_version(&self) -> Option<String> {
self.compat.lock().unwrap_or_else(|e| e.into_inner()).node_version.clone()
}
fn check_compat(&self, md: &MetadataMap) -> Result<(), StreamError> {
let (raw, node) = compat::node_version_from(md);
let schema = compat::schema_from(md);
let sdk = compat::parse(compat::SDK_VERSION).expect("the crate version parses");
let warn_for = {
let mut state = self.compat.lock().unwrap_or_else(|e| e.into_inner());
state.node_version = raw.clone();
if schema != satd_events_proto::SCHEMA_VERSION {
return Err(StreamError::SchemaMismatch {
node: schema,
sdk: satd_events_proto::SCHEMA_VERSION,
});
}
let warn = match compat::classify(sdk, node) {
Compat::Ok => false,
Compat::OneBehind => true,
Compat::TooOld if self.allow_old_node => true,
Compat::TooOld => {
return Err(StreamError::NodeTooOld {
node: compat::display_node_version(raw.as_deref()),
sdk: compat::SDK_VERSION,
});
}
};
if warn && state.warned_for.as_ref() != Some(&raw) {
state.warned_for = Some(raw.clone());
Some(compat::display_node_version(raw.as_deref()))
} else {
None
}
};
if let Some(shown) = warn_for {
tracing::warn!(
target: "satd_events_client::compat",
node_version = %shown,
sdk_version = compat::SDK_VERSION,
"satd node {} is older than this SDK ({}); features added after the node's \
release are unavailable and request fields it does not recognise are ignored. \
Upgrade the node — connections are refused once it is two minor versions behind.",
shown,
compat::SDK_VERSION,
);
}
Ok(())
}
fn authed<T>(&self, msg: T) -> tonic::Request<T> {
let mut req = tonic::Request::new(msg);
if let Some(value) = &self.auth {
req.metadata_mut().insert("authorization", value.clone());
}
req
}
pub async fn subscribe(
&mut self,
opts: SubscribeOptions,
) -> Result<EventStream, StreamError> {
let req = self.authed(opts.into_request());
let resp = self.inner.subscribe(req).await?;
self.check_compat(resp.metadata())?;
Ok(EventStream::new(resp.into_inner()))
}
pub fn resilient_subscribe(
&self,
opts: SubscribeOptions,
config: crate::ResilientConfig,
) -> crate::ResilientSubscription {
crate::ResilientSubscription::new(self.clone(), opts, config)
}
pub fn resilient_watch(&self, config: crate::ResilientWatchConfig) -> crate::ResilientWatch {
crate::ResilientWatch::new(self.clone(), config)
}
pub async fn watch(&mut self) -> Result<(WatchHandle, EventStream), StreamError> {
let (tx, rx) = mpsc::channel::<pb::SubscribeControl>(CONTROL_BUFFER);
let req = self.authed(ReceiverStream::new(rx));
let resp = self.inner.watch(req).await?;
self.check_compat(resp.metadata())?;
Ok((WatchHandle { tx }, EventStream::new(resp.into_inner())))
}
}
#[cfg(test)]
impl WatchHandle {
pub(crate) fn for_test() -> (Self, mpsc::Receiver<pb::SubscribeControl>) {
let (tx, rx) = mpsc::channel(64);
(WatchHandle { tx }, rx)
}
}
#[cfg(test)]
impl StreamClient {
pub(crate) fn for_test() -> Self {
let channel = Endpoint::from_static("http://127.0.0.1:1").connect_lazy();
StreamClient {
inner: NodeEventStreamClient::new(channel),
auth: None,
allow_old_node: false,
compat: Arc::default(),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use pb::subscribe_control::Msg;
fn handle() -> (WatchHandle, mpsc::Receiver<pb::SubscribeControl>) {
let (tx, rx) = mpsc::channel(16);
(WatchHandle { tx }, rx)
}
fn next(rx: &mut mpsc::Receiver<pb::SubscribeControl>) -> Msg {
rx.try_recv().expect("a control message was sent").msg.expect("msg set")
}
#[test]
fn only_an_https_scheme_counts_as_encrypted() {
assert!(endpoint_is_https("https://node.example:50051"));
assert!(endpoint_is_https("HTTPS://node.example:50051"));
assert!(!endpoint_is_https("http://node.example:50051"));
assert!(!endpoint_is_https("node.example:50051"));
assert!(!endpoint_is_https("127.0.0.1:50051"));
}
#[test]
fn encryption_requires_the_tls_feature_and_not_just_the_scheme() {
assert_eq!(endpoint_is_encrypted("https://node.example:50051"), cfg!(feature = "tls"));
assert!(!endpoint_is_encrypted("http://node.example:50051"));
assert!(!endpoint_is_encrypted("node.example:50051"));
}
#[test]
fn the_refusal_does_not_echo_url_userinfo() {
assert_eq!(
redact_endpoint_userinfo("http://user:hunter2@node.example:50051"),
"http://***@node.example:50051"
);
assert_eq!(
redact_endpoint_userinfo("http://user:hunter2@node.example:50051/a@b"),
"http://***@node.example:50051/a@b"
);
assert_eq!(redact_endpoint_userinfo("user:hunter2@node.example:50051"), "***@node.example:50051");
assert_eq!(
redact_endpoint_userinfo("https://node.example:50051/path"),
"https://node.example:50051/path"
);
assert_eq!(redact_endpoint_userinfo("node.example:50051"), "node.example:50051");
}
#[tokio::test]
async fn a_password_in_the_endpoint_is_not_quoted_back() {
let err = StreamClient::builder("http://user:hunter2@node.example:50051")
.bearer_token("SECRET")
.connect()
.await
.expect_err("a bearer token over plaintext must be refused");
assert!(!err.to_string().contains("hunter2"), "leaked the URL password: {err}");
assert!(!err.to_string().contains("SECRET"), "leaked the token: {err}");
}
#[tokio::test]
async fn a_bearer_token_over_plaintext_is_refused() {
for endpoint in ["http://127.0.0.1:1", "http://node.example:50051", "node.example:50051"] {
let err = StreamClient::builder(endpoint)
.bearer_token("SECRET")
.connect()
.await
.expect_err("a bearer token over plaintext must be refused");
assert!(
matches!(err, StreamError::InsecureCredential(_)),
"{endpoint}: got {err:?}"
);
assert!(
!err.to_string().contains("SECRET"),
"{endpoint}: the error must not quote the token: {err}"
);
}
}
fn assert_reached_the_transport(err: &StreamError, ctx: &str) {
assert!(
matches!(err, StreamError::Connect(_) | StreamError::Transport(_)),
"{ctx}: expected a transport/connect error, got {err:?}"
);
}
#[tokio::test]
async fn plaintext_without_a_token_is_unaffected() {
let err = StreamClient::builder("http://127.0.0.1:1")
.connect()
.await
.expect_err("nothing is listening on port 1");
assert_reached_the_transport(&err, "no token");
}
#[tokio::test]
async fn the_named_waiver_permits_a_token_over_plaintext() {
let err = StreamClient::builder("http://127.0.0.1:1")
.insecure_bearer_token("SECRET")
.connect()
.await
.expect_err("nothing is listening on port 1");
assert_reached_the_transport(&err, "waiver");
}
#[tokio::test]
async fn the_waiver_does_not_leak_into_a_subsequent_bearer_token() {
let err = StreamClient::builder("http://127.0.0.1:1")
.insecure_bearer_token("SECRET")
.bearer_token("SECRET")
.connect()
.await
.expect_err("the waiver must not survive a later bearer_token");
assert!(
matches!(err, StreamError::InsecureCredential(_)),
"got {err:?}"
);
}
#[cfg(feature = "tls")]
#[tokio::test]
async fn https_permits_a_token() {
let err = StreamClient::builder("https://127.0.0.1:1")
.tls()
.bearer_token("SECRET")
.connect()
.await
.expect_err("nothing is listening on port 1");
assert!(
!matches!(err, StreamError::InsecureCredential(_)),
"got {err:?}"
);
}
#[test]
fn subscribe_options_map_tweak_knobs_only_when_set() {
let default = SubscribeOptions { categories: Categories::TWEAKS, ..Default::default() }
.into_request();
assert_eq!(default.mempool_tweaks, None);
assert_eq!(default.tweak_outputs, None);
assert_eq!(default.tweak_unspent_only, None);
let full = SubscribeOptions {
categories: Categories::TWEAKS,
mempool_tweaks: true,
tweak_outputs: true,
tweaks_only: true,
tweak_unspent_only: true,
..Default::default()
}
.into_request();
assert_eq!(full.mempool_tweaks, Some(true));
assert_eq!(full.tweak_outputs, Some(true));
assert_eq!(full.tweaks_only, Some(true));
assert_eq!(full.tweak_unspent_only, Some(true));
}
#[tokio::test]
async fn add_scripts_mixed_floors_emits_parallel_min_values() {
let (h, mut rx) = handle();
h.add_scripts([([1u8; 32], Some(5_000)), ([2u8; 32], None)]).await.unwrap();
match next(&mut rx) {
Msg::AddScripts(a) => {
assert_eq!(a.scripthashes.len(), 2);
assert_eq!(a.min_values, vec![5_000, 0]);
}
other => panic!("expected AddScripts, got {other:?}"),
}
}
#[tokio::test]
async fn add_scripts_no_floors_emits_empty_min_values() {
let (h, mut rx) = handle();
h.add_scripts([([1u8; 32], None), ([2u8; 32], None)]).await.unwrap();
match next(&mut rx) {
Msg::AddScripts(a) => assert!(a.min_values.is_empty()),
other => panic!("expected AddScripts, got {other:?}"),
}
}
#[tokio::test]
async fn empty_inputs_send_nothing() {
let (h, mut rx) = handle();
h.add_scripts([]).await.unwrap();
h.add_outpoints([]).await.unwrap();
h.add_depth_alarms([[1u8; 32]], []).await.unwrap(); h.add_depth_alarms([], [3]).await.unwrap(); h.add_depth_alarms([[1u8; 32]], [0]).await.unwrap();
assert!(rx.try_recv().is_err(), "no message should have been sent");
}
#[tokio::test]
async fn lifecycle_vs_depth_alarms_dispatch_on_min_depths() {
let (h, mut rx) = handle();
h.add_tx_lifecycle([[7u8; 32]], AutoClose::AtDepth(6)).await.unwrap();
match next(&mut rx) {
Msg::AddTransactions(a) => {
assert!(a.min_depths.is_empty(), "lifecycle => empty min_depths");
assert_eq!(a.auto_close_depth, 6);
}
other => panic!("expected AddTransactions, got {other:?}"),
}
h.add_depth_alarms([[7u8; 32], [8u8; 32]], [3, 6]).await.unwrap();
match next(&mut rx) {
Msg::AddTransactions(a) => {
assert_eq!(a.txids.len(), 2);
assert_eq!(a.min_depths, vec![3, 6]); assert_eq!(a.auto_close_depth, 0);
}
other => panic!("expected AddTransactions, got {other:?}"),
}
}
#[tokio::test]
async fn prefix_validation_rejects_bad_length_and_bits() {
let (h, mut rx) = handle();
let err = h.add_script_prefixes([(vec![0xab], 16)]).await.unwrap_err();
assert!(matches!(err, StreamError::InvalidArgument(_)));
let err = h.add_script_prefixes([(vec![], 0)]).await.unwrap_err();
assert!(matches!(err, StreamError::InvalidArgument(_)));
let err = h.add_script_prefixes([(vec![0u8; 5], 33)]).await.unwrap_err();
assert!(matches!(err, StreamError::InvalidArgument(_)));
assert!(rx.try_recv().is_err());
h.add_script_prefixes([(vec![0xde, 0xad, 0xbe, 0xef], 32)]).await.unwrap();
match next(&mut rx) {
Msg::AddScriptPrefixes(a) => assert_eq!(a.prefixes[0].bits, 32),
other => panic!("expected AddScriptPrefixes, got {other:?}"),
}
h.add_script_prefixes([(vec![0xab, 0xcd], 16)]).await.unwrap();
match next(&mut rx) {
Msg::AddScriptPrefixes(a) => {
assert_eq!(a.prefixes.len(), 1);
assert_eq!(a.prefixes[0].bits, 16);
assert_eq!(a.prefixes[0].prefix, vec![0xab, 0xcd]);
}
other => panic!("expected AddScriptPrefixes, got {other:?}"),
}
}
#[tokio::test]
async fn set_cursor_wraps_cursor() {
let (h, mut rx) = handle();
let cursor = Cursor { height: 9, tx_index: 0, mempool_seq: 1, instance_id: 2 };
h.set_cursor(cursor).await.unwrap();
match next(&mut rx) {
Msg::SetCursor(s) => assert_eq!(s.cursor, Some(cursor)),
other => panic!("expected SetCursor, got {other:?}"),
}
}
#[cfg(feature = "tls")]
#[test]
fn tls_builder_debug_redacts_secrets() {
let b = StreamClient::builder("https://node:50051")
.bearer_token("SUPER_SECRET_TOKEN")
.tls_ca_pem(b"-----BEGIN CERTIFICATE-----SECRET_CA_BYTES".to_vec())
.tls_client_identity(b"CLIENT_CERT".to_vec(), b"CLIENT_PRIVATE_KEY".to_vec());
let dbg = format!("{b:?}");
assert!(dbg.contains("tls: true"), "expected tls: true, got {dbg}");
assert!(dbg.contains("<redacted>"), "token should be redacted: {dbg}");
assert!(!dbg.contains("SUPER_SECRET_TOKEN"));
assert!(!dbg.contains("SECRET_CA_BYTES"));
assert!(!dbg.contains("CLIENT_PRIVATE_KEY"));
}
#[cfg(feature = "tls")]
#[tokio::test]
async fn tls_connect_assembles_and_fails_cleanly() {
let err = StreamClient::builder("https://127.0.0.1:1")
.tls()
.tls_domain("node.example")
.connect()
.await
.expect_err("connect to a dead port must fail");
assert!(
!matches!(err, StreamError::InvalidEndpoint(_) | StreamError::InvalidToken),
"expected a transport/connect error, got {err:?}",
);
}
#[cfg(feature = "tls")]
#[tokio::test]
async fn tls_over_non_https_endpoint_is_refused() {
for ep in ["http://127.0.0.1:1", "127.0.0.1:1"] {
let err = StreamClient::builder(ep)
.tls_ca_pem(b"-----BEGIN CERTIFICATE-----".to_vec())
.bearer_token("SECRET")
.connect()
.await
.expect_err("TLS over non-https must be refused");
match err {
StreamError::InvalidEndpoint(msg) => assert!(
msg.contains("https"),
"expected an https-scheme error for {ep}, got {msg}"
),
other => panic!("expected InvalidEndpoint for {ep}, got {other:?}"),
}
}
}
}
#[cfg(test)]
mod version_check_tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
use satd_events_proto::v1::node_event_stream_server::{
NodeEventStream, NodeEventStreamServer,
};
use tokio_stream::Stream;
use crate::compat::{Version, SDK_VERSION};
use crate::{ResilientConfig, ResilientWatchConfig};
type EventResult = Result<pb::NodeEvent, tonic::Status>;
type BoxStream = std::pin::Pin<Box<dyn Stream<Item = EventResult> + Send>>;
#[derive(Clone)]
struct FakeNode {
version: Arc<Mutex<Option<String>>>,
schema: Option<&'static str>,
event_schema: Option<u32>,
opens: Arc<AtomicUsize>,
}
impl FakeNode {
fn new(version: Option<&str>) -> Self {
FakeNode {
version: Arc::new(Mutex::new(version.map(str::to_owned))),
schema: Some("1"),
event_schema: None,
opens: Arc::default(),
}
}
fn respond(&self) -> tonic::Response<BoxStream> {
self.opens.fetch_add(1, Ordering::SeqCst);
let events: Vec<EventResult> = self
.event_schema
.map(|schema_version| {
Ok(pb::NodeEvent {
schema_version,
body: Some(pb::node_event::Body::Heartbeat(pb::Heartbeat {
uptime_ns: 1,
})),
..Default::default()
})
})
.into_iter()
.collect();
let stream = tokio_stream::StreamExt::chain(
tokio_stream::iter(events),
tokio_stream::pending(),
);
let mut resp = tonic::Response::new(Box::pin(stream) as BoxStream);
let md = resp.metadata_mut();
if let Some(v) = self.version.lock().unwrap().as_deref() {
md.insert(compat::VERSION_HEADER, v.parse().unwrap());
}
if let Some(s) = self.schema {
md.insert(compat::SCHEMA_HEADER, s.parse().unwrap());
}
resp
}
}
#[tonic::async_trait]
impl NodeEventStream for FakeNode {
type SubscribeStream = BoxStream;
type WatchStream = BoxStream;
async fn subscribe(
&self,
_: tonic::Request<pb::SubscribeRequest>,
) -> Result<tonic::Response<BoxStream>, tonic::Status> {
Ok(self.respond())
}
async fn watch(
&self,
_: tonic::Request<tonic::Streaming<pb::SubscribeControl>>,
) -> Result<tonic::Response<BoxStream>, tonic::Status> {
Ok(self.respond())
}
}
async fn serve(fake: FakeNode) -> String {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let incoming =
tonic::transport::server::TcpIncoming::from_listener(listener, true, None).unwrap();
tokio::spawn(
tonic::transport::Server::builder()
.add_service(NodeEventStreamServer::new(fake))
.serve_with_incoming(incoming),
);
format!("http://{addr}")
}
async fn client_for(fake: &FakeNode) -> StreamClient {
StreamClient::builder(serve(fake.clone()).await).connect().await.expect("connect")
}
fn sdk() -> Version {
compat::parse(SDK_VERSION).unwrap()
}
fn minors_behind(n: u32) -> Option<String> {
let sdk = sdk();
(sdk.minor >= n).then(|| format!("{}.{}.0", sdk.major, sdk.minor - n))
}
fn too_old() -> String {
minors_behind(2).unwrap_or_else(|| format!("{}.9.0", sdk().major - 1))
}
thread_local! {
static THREAD_LOGS: std::cell::RefCell<Vec<u8>> =
const { std::cell::RefCell::new(Vec::new()) };
}
struct ThreadLogs;
thread_local! {
static PROBE_COMPAT: std::cell::RefCell<Option<Arc<Mutex<CompatState>>>> =
const { std::cell::RefCell::new(None) };
static PROBE_FREE: std::cell::RefCell<Vec<bool>> =
const { std::cell::RefCell::new(Vec::new()) };
}
struct ProbeLock;
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for ProbeLock {
fn on_event(
&self,
_: &tracing::Event<'_>,
_: tracing_subscriber::layer::Context<'_, S>,
) {
PROBE_COMPAT.with(|p| {
if let Some(compat) = p.borrow().as_ref() {
let free = compat.try_lock().is_ok();
PROBE_FREE.with(|f| f.borrow_mut().push(free));
}
});
}
}
struct Probe;
impl Probe {
fn arm(compat: Arc<Mutex<CompatState>>) -> Self {
PROBE_FREE.with(|f| f.borrow_mut().clear());
PROBE_COMPAT.with(|p| *p.borrow_mut() = Some(compat));
Probe
}
fn seen(&self) -> Vec<bool> {
PROBE_FREE.with(|f| f.borrow().clone())
}
}
impl Drop for Probe {
fn drop(&mut self) {
PROBE_COMPAT.with(|p| *p.borrow_mut() = None);
}
}
impl std::io::Write for ThreadLogs {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
THREAD_LOGS.with(|l| l.borrow_mut().extend_from_slice(buf));
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
struct Logs;
impl Logs {
fn capture() -> Self {
static INIT: std::sync::Once = std::sync::Once::new();
INIT.call_once(|| {
use tracing_subscriber::layer::SubscriberExt as _;
use tracing_subscriber::util::SubscriberInitExt as _;
let _ = tracing_subscriber::registry()
.with(
tracing_subscriber::fmt::layer()
.with_ansi(false)
.with_writer(|| ThreadLogs),
)
.with(ProbeLock)
.with(tracing_subscriber::filter::LevelFilter::TRACE)
.try_init();
});
THREAD_LOGS.with(|l| l.borrow_mut().clear());
Logs
}
fn lines(&self) -> Vec<String> {
THREAD_LOGS.with(|l| {
String::from_utf8(l.borrow().clone())
.unwrap()
.lines()
.map(str::to_owned)
.collect()
})
}
fn warnings(&self) -> Vec<String> {
self.lines().into_iter().filter(|l| l.contains("WARN")).collect()
}
}
#[tokio::test]
async fn same_version_node_opens_silently() {
let logs = Logs::capture();
let fake = FakeNode::new(Some(SDK_VERSION));
let mut client = client_for(&fake).await;
client.subscribe(SubscribeOptions::default()).await.expect("subscribe");
client.watch().await.expect("watch");
assert!(logs.warnings().is_empty(), "{:?}", logs.warnings());
}
#[tokio::test]
async fn newer_node_opens_silently() {
let logs = Logs::capture();
let sdk = sdk();
let fake = FakeNode::new(Some(&format!("{}.{}.0", sdk.major, sdk.minor + 3)));
let mut client = client_for(&fake).await;
client.subscribe(SubscribeOptions::default()).await.expect("subscribe");
assert!(logs.warnings().is_empty(), "{:?}", logs.warnings());
}
#[tokio::test]
async fn headerless_node_is_treated_as_0_5() {
let mut fake = FakeNode::new(None);
fake.schema = None;
let mut client = client_for(&fake).await;
let res = client.subscribe(SubscribeOptions::default()).await;
match compat::classify(sdk(), compat::HEADERLESS_NODE) {
Compat::TooOld => assert!(matches!(res, Err(StreamError::NodeTooOld { .. }))),
_ => assert!(res.is_ok(), "{:?}", res.err()),
}
assert_eq!(client.node_version(), None);
}
#[tokio::test]
async fn two_minors_behind_is_refused() {
let fake = FakeNode::new(Some(&too_old()));
let mut client = client_for(&fake).await;
let err = client.subscribe(SubscribeOptions::default()).await.err().expect("refused");
match &err {
StreamError::NodeTooOld { node, sdk } => {
assert_eq!(node, &too_old());
assert_eq!(*sdk, SDK_VERSION);
}
other => panic!("expected NodeTooOld, got {other:?}"),
}
assert!(!err.is_retryable());
let err = client.watch().await.err().expect("watch refused too");
assert!(matches!(err, StreamError::NodeTooOld { .. }), "{err:?}");
}
#[tokio::test]
async fn allow_old_node_downgrades_the_refusal_to_a_warning() {
let logs = Logs::capture();
let fake = FakeNode::new(Some(&too_old()));
let mut client = StreamClient::builder(serve(fake.clone()).await)
.allow_old_node()
.connect()
.await
.unwrap();
client.subscribe(SubscribeOptions::default()).await.expect("allowed");
assert_eq!(logs.warnings().len(), 1, "{:?}", logs.warnings());
}
#[tokio::test]
async fn schema_header_mismatch_is_refused_even_with_allow_old_node() {
let mut fake = FakeNode::new(Some(SDK_VERSION));
fake.schema = Some("2");
let mut client = StreamClient::builder(serve(fake.clone()).await)
.allow_old_node()
.connect()
.await
.unwrap();
let err = client.subscribe(SubscribeOptions::default()).await.err().expect("refused");
assert!(matches!(err, StreamError::SchemaMismatch { node: 2, sdk: 1 }), "{err:?}");
assert!(!err.is_retryable());
let err = client.watch().await.err().expect("watch refused too");
assert!(matches!(err, StreamError::SchemaMismatch { node: 2, sdk: 1 }), "{err:?}");
}
#[tokio::test]
async fn first_event_schema_mismatch_is_refused() {
let mut fake = FakeNode::new(Some(SDK_VERSION));
fake.event_schema = Some(2);
let mut client = client_for(&fake).await;
let mut stream = client.subscribe(SubscribeOptions::default()).await.expect("opens");
let err = stream.message().await.expect_err("refused");
assert!(matches!(err, StreamError::SchemaMismatch { node: 2, sdk: 1 }), "{err:?}");
let mut fake = FakeNode::new(Some(SDK_VERSION));
fake.event_schema = Some(1);
let mut client = client_for(&fake).await;
let (_handle, mut stream) = client.watch().await.expect("opens");
let ev = stream.message().await.expect("event").expect("not closed");
assert!(matches!(ev, Event::Heartbeat { .. }), "{ev:?}");
}
#[tokio::test]
async fn one_behind_warns_once_per_node_version() {
let logs = Logs::capture();
let (behind, allow) = match minors_behind(1) {
Some(v) => (v, false),
None => (too_old(), true),
};
let fake = FakeNode::new(Some(&behind));
let mut builder = StreamClient::builder(serve(fake.clone()).await);
if allow {
builder = builder.allow_old_node();
}
let mut client = builder.connect().await.unwrap();
client.subscribe(SubscribeOptions::default()).await.expect("subscribe");
client.subscribe(SubscribeOptions::default()).await.expect("subscribe again");
client.clone().watch().await.expect("watch");
let warnings = logs.warnings();
assert_eq!(warnings.len(), 1, "{warnings:?}");
assert!(warnings[0].contains(&format!("node_version={behind}")), "{warnings:?}");
assert!(warnings[0].contains(&format!("sdk_version=\"{SDK_VERSION}\"")), "{warnings:?}");
assert!(warnings[0].contains("is older than this SDK"), "{warnings:?}");
*fake.version.lock().unwrap() = Some(format!("{behind}-other"));
client.subscribe(SubscribeOptions::default()).await.expect("subscribe");
client.subscribe(SubscribeOptions::default()).await.expect("subscribe");
assert_eq!(logs.warnings().len(), 2, "{:?}", logs.warnings());
}
#[tokio::test]
async fn warning_is_logged_without_holding_the_compat_lock() {
let _logs = Logs::capture();
let (behind, allow) = match minors_behind(1) {
Some(v) => (v, false),
None => (too_old(), true),
};
let fake = FakeNode::new(Some(&behind));
let mut builder = StreamClient::builder(serve(fake.clone()).await);
if allow {
builder = builder.allow_old_node();
}
let mut client = builder.connect().await.unwrap();
let probe = Probe::arm(client.compat.clone());
client.subscribe(SubscribeOptions::default()).await.expect("subscribe");
let seen = probe.seen();
assert!(!seen.is_empty(), "the warning was not logged");
assert!(seen.iter().all(|f| *f), "an event was logged under the compat lock: {seen:?}");
}
#[tokio::test]
async fn node_version_is_recorded_once_a_stream_opens() {
let fake = FakeNode::new(Some(SDK_VERSION));
let client = client_for(&fake).await;
assert_eq!(client.node_version(), None);
client.clone().subscribe(SubscribeOptions::default()).await.expect("subscribe");
assert_eq!(client.node_version().as_deref(), Some(SDK_VERSION));
}
#[tokio::test]
async fn resilient_subscribe_surfaces_node_too_old_without_retrying() {
let fake = FakeNode::new(Some(&too_old()));
let client = client_for(&fake).await;
let mut sub =
client.resilient_subscribe(SubscribeOptions::default(), ResilientConfig::default());
let err = tokio::time::timeout(Duration::from_secs(5), sub.next())
.await
.expect("no backoff loop")
.expect_err("refused");
assert!(matches!(err, StreamError::NodeTooOld { .. }), "{err:?}");
assert_eq!(fake.opens.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn resilient_watch_surfaces_node_too_old_without_retrying() {
let fake = FakeNode::new(Some(&too_old()));
let client = client_for(&fake).await;
let mut watch = client.resilient_watch(ResilientWatchConfig::default());
let err = tokio::time::timeout(Duration::from_secs(5), watch.next())
.await
.expect("no backoff loop")
.expect_err("refused");
assert!(matches!(err, StreamError::NodeTooOld { .. }), "{err:?}");
assert_eq!(fake.opens.load(Ordering::SeqCst), 1);
}
}