satd_events_client/

client.rs

//! The streaming client: connect, `Subscribe` firehose, and the `Watch` handle
//! with its typed watch helpers. This module establishes the connection, auth,
//! and cursor-capturing event stream that the reconnect/replay resilience layer
//! (the `resilience` module) builds on.

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};

/// Bounded control-channel depth for a `Watch` stream.
const CONTROL_BUFFER: usize = 16;

/// Category bitfield helpers for [`SubscribeOptions::categories`].
///
/// `0` (the default, [`Categories::ALL`]) means "all categories". Combine the
/// others with `|`.
pub struct Categories;

impl Categories {
    /// All categories (the server default).
    pub const ALL: u32 = 0;
    /// Mempool events.
    pub const MEMPOOL: u32 = 1;
    /// Chain (block connect/disconnect/reorg) events.
    pub const CHAIN: u32 = 2;
    /// Heartbeats.
    pub const HEARTBEAT: u32 = 4;
    /// BIP 352 silent-payment per-block tweaks (Tier 1 client-side scan). Unlike
    /// the other bits this is **not** part of the [`ALL`](Self::ALL) (`0`)
    /// default: a subscription must request it explicitly, so an existing
    /// `0`-subscriber never starts receiving tweak volume after a node upgrade.
    /// Requires the node's tweak index (`silentpaymentindex=1`); a `tweaks`
    /// subscription against a node with it disabled is refused in-band.
    pub const TWEAKS: u32 = 8;
    /// Node-health conditions the daemon detects about itself
    /// ([`Event::Status`](crate::Event::Status)). Like
    /// [`TWEAKS`](Self::TWEAKS) this is **not** part of the [`ALL`](Self::ALL)
    /// (`0`) default — a client written against an older node never starts
    /// receiving a body it has no parser for. Unlike `TWEAKS` there is no index
    /// prerequisite: any node serves it.
    pub const STATUS: u32 = 16;
}

/// Options for a [`StreamClient::subscribe`] firehose.
#[derive(Debug, Clone, Default)]
pub struct SubscribeOptions {
    /// Category bitfield; `0` = all. See [`Categories`].
    pub categories: u32,
    /// Durable replay anchor. When set, the server replays confirmed history
    /// forward from this cursor, then joins live with no gap or duplicate.
    pub from_cursor: Option<Cursor>,
    /// Forward-only dedup filter: drop events with `seq <= since_seq`. Use after
    /// a brief reconnect within the broadcast window — not for durable replay.
    pub since_seq: Option<u64>,
    /// Silent-payment tweak dust floor (satoshis): drop [`TweakEntry`](crate::TweakEntry)
    /// whose `max_value` is below this, and flag the block `filtered`. Only
    /// meaningful with the [`TWEAKS`](Categories::TWEAKS) category. `None` = keep
    /// every tweak.
    pub tweak_dust_limit: Option<u64>,
    /// Compact tweak form: when `true`, tweak entries carry only the 33-byte
    /// tweak (no `txid`/`max_value`), minimizing bytes for a pure client-side
    /// scan. Only meaningful with the [`TWEAKS`](Categories::TWEAKS) category.
    pub tweaks_only: bool,
    /// Mempool-time tweaks ("Tier 1.5"): when `true`, the server additionally
    /// streams an [`Event::MempoolTweak`](crate::Event::MempoolTweak) at each
    /// SP-eligible transaction's admission, for mempool-latency detection without
    /// uploading a scan key. A modifier on the [`TWEAKS`](Categories::TWEAKS)
    /// category — the server rejects it if that bit is not set. `false` (default)
    /// = block tweaks only. Ephemeral and best-effort: mempool tweaks are not
    /// replayable, so a missed admission is caught at confirmation via
    /// [`Event::BlockTweaks`](crate::Event::BlockTweaks).
    pub mempool_tweaks: bool,
    /// Include the transaction's taproot outputs on each
    /// [`Event::BlockTweaks`](crate::Event::BlockTweaks) entry
    /// ([`TweakEntry::taproot_outputs`](crate::TweakEntry::taproot_outputs)), so
    /// a match is confirmed against the on-chain output without fetching the
    /// block. `false` (default) keeps the confirmed firehose lean. A modifier on
    /// the [`TWEAKS`](Categories::TWEAKS) category — the server rejects it if that
    /// bit is not set, or if the node has no block source. A
    /// [`MempoolTweak`](crate::Event::MempoolTweak) carries its outputs
    /// regardless of this flag.
    pub tweak_outputs: bool,
    /// Cut-through: drop [`Event::BlockTweaks`](crate::Event::BlockTweaks)
    /// entries whose taproot outputs are **all** already spent on the confirmed
    /// chain. A balance scan then never runs ECDH against a coin that is already
    /// gone. A surviving entry carries its **full** taproot output set —
    /// spentness decides whether an entry survives, never which outputs it
    /// carries, so BIP 352 `k` enumeration is never truncated.
    ///
    /// A **restore** that needs transaction history must leave this `false`: a
    /// payment received and later spent is omitted entirely, so the balance is
    /// right and the history is not. Spentness is evaluated at serve time, so the
    /// same historical block can yield fewer entries on a later scan. Dropped
    /// entries set the block's `filtered` flag. A modifier on the
    /// [`TWEAKS`](Categories::TWEAKS) category — the server rejects it without
    /// that bit, or on a node with no block source. Never applies to
    /// [`MempoolTweak`](crate::Event::MempoolTweak).
    pub tweak_unspent_only: bool,
}

impl SubscribeOptions {
    // `..Default::default()` is redundant against the current workspace proto but
    // retained so this published crate still compiles if the schema grows fields
    // ahead of this literal (the same resilience the released-proto pin needs).
    #[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,
            // Send the flag only when set, so a default subscription is
            // byte-identical to one built before these knobs existed.
            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()
        }
    }
}

/// Server-enforced cap on the number of *distinct* labels one scan-key target
/// may carry (mirrors the node's `MAX_SP_LABELS_PER_TARGET`). The server rejects
/// an over-label target; enforcing the same bound client-side turns that into a
/// deterministic error at the call site instead of a silent skip.
pub const MAX_SP_LABELS_PER_TARGET: usize = 16;

/// Server-enforced cap on the number of scan-key targets per connection (mirrors
/// the node's `MAX_SP_TARGETS_PER_CONNECTION`). The server silently sheds an
/// over-cap add; a stateful [`ResilientWatch`](crate::ResilientWatch) enforces
/// this bound before recording/sending so it never mirrors and replays a target
/// the server dropped.
pub const MAX_SP_TARGETS_PER_CONNECTION: usize = 16;

/// A BIP 352 scan-key watch target (Tier 2): a `(scan_secret, spend_pubkey)`
/// pair plus optional labels, registered via
/// [`WatchHandle::add_silent_payments`]. The node runs the ECDH match and pushes
/// [`Event::SilentPaymentMatched`](crate::Event::SilentPaymentMatched).
///
/// `scan_secret` is a **watch credential**, not a spending key: disclosing it
/// lets the node (and anyone who compromises it) learn *which* outputs are
/// yours, but never spend them — spend authority stays with `B_spend`'s private
/// half, which is never sent. `Debug` redacts the scan secret.
#[derive(Clone)]
pub struct SilentPaymentTarget {
    /// 32-byte scan secret `b_scan`.
    pub scan_secret: [u8; 32],
    /// 33-byte compressed spend public key `B_spend` (public half only).
    pub spend_pubkey: [u8; 33],
    /// Receiver label integers to also match. Include `0` to catch change.
    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 {
    /// Scrub the retained scan secret on drop. `b_scan` is a watch credential —
    /// the long-lived copies held by a `ResilientWatch` mirror (and any clones)
    /// must not linger in freed memory once a target is removed or the mirror is
    /// torn down. (The transient `Vec<u8>` a `to_proto` copy places on the wire
    /// is owned by the outbound message and cleared with it; this covers every
    /// in-process copy the SDK retains.)
    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(),
        }
    }

    /// The target's identity `b_scan·G` (33-byte compressed) — the key used to
    /// remove it ([`WatchHandle::remove_silent_payments`]) and how a
    /// [`ResilientWatch`](crate::ResilientWatch) tracks it. Requires the
    /// `bitcoin` feature (enabled by default) for the elliptic-curve derivation.
    #[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())
    }

    /// Validate **both** curve values — the `scan_secret` scalar and the
    /// `spend_pubkey` point — and return the identity `b_scan·G`. Client-side
    /// validation turns a malformed target into a deterministic
    /// [`InvalidArgument`](StreamError::InvalidArgument) at the call site instead
    /// of a silent server-side skip that would return `Ok(())` while installing
    /// no watch (and repeat the invalid target on every resilient reconnect).
    /// Requires the `bitcoin` feature (enabled by default).
    #[cfg(feature = "bitcoin")]
    pub fn validate(&self) -> Result<[u8; 33], StreamError> {
        let id = self.scan_pubkey()?; // validates the b_scan scalar
        bitcoin::secp256k1::PublicKey::from_slice(&self.spend_pubkey)
            .map_err(|_| StreamError::InvalidArgument("spend_pubkey is not a valid point".into()))?;
        // Mirror the server's per-target label cap: an over-label target is
        // otherwise silently skipped by the node while the client believes it
        // was installed (and a resilient reconnect replays it forever).
        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)
    }
}

/// Lowercase hex, for redacting `Debug` impls (no external dep).
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
}

/// A live stream of typed [`Event`]s, shared by `Subscribe` and `Watch`.
///
/// As confirmed events flow, the stream captures their durable [`Cursor`];
/// [`cursor`](Self::cursor) returns the latest, which a consumer persists and
/// presents as [`SubscribeOptions::from_cursor`] to resume.
pub struct EventStream {
    inner: tonic::Streaming<pb::NodeEvent>,
    last_cursor: Option<Cursor>,
    /// Set once the first event's `schema_version` has been checked.
    schema_checked: bool,
}

impl EventStream {
    /// Await the next event. Returns `Ok(None)` when the server closes the
    /// stream.
    pub async fn message(&mut self) -> Result<Option<Event>, StreamError> {
        match self.inner.message().await {
            Ok(Some(ev)) => {
                // The response header already carried the schema on a node
                // that sends one. This covers a node that predates the header.
                // One event is enough: a node does not change schema mid-stream.
                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)),
        }
    }

    /// The most recent durable cursor seen on this stream, if any. Persist it
    /// to resume after a disconnect.
    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 }
    }
}

/// Auto-close policy for a transaction lifecycle watch: self-evict and emit
/// `TxidFinalized` once the tx is buried this many confirmations deep.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum AutoClose {
    /// Never auto-close; the lifecycle watch persists until removed.
    #[default]
    Never,
    /// Auto-close once the tx reaches this confirmation depth (>= 1).
    AtDepth(u32),
}

impl AutoClose {
    fn depth(self) -> u32 {
        match self {
            AutoClose::Never => 0,
            AutoClose::AtDepth(d) => d,
        }
    }
}

/// A handle to send control messages on a bidirectional `Watch` stream.
///
/// Typed helpers build and send the right [`SubscribeControl`](pb::SubscribeControl)
/// for each watch kind; [`send_control`](Self::send_control) remains for raw
/// access. Empty inputs are no-ops (nothing is sent). Dropping the handle closes
/// the control channel, which the server uses to tear the stream down.
pub struct WatchHandle {
    tx: mpsc::Sender<pb::SubscribeControl>,
}

impl WatchHandle {
    /// Send a raw [`SubscribeControl`](pb::SubscribeControl) message.
    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
    }

    /// Add scripthashes (each `sha256(scriptPubKey)`), with an optional per-script
    /// `min_value` floor in satoshis. Matches below a script's floor are
    /// suppressed (a `None` floor delivers everything). Re-asserting a held
    /// scripthash updates its floor. Charges one watch-quota unit per scripthash.
    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();
        // `min_values` is empty when no floor is set on any script; otherwise it
        // must be parallel to `scripthashes`, with 0 (deliver-all) for the
        // unfloored entries.
        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
    }

    /// Remove scripthashes from the watch-set, releasing their quota.
    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
    }

    /// Add outpoints (`txid:vout`) to the watch-set. Charges one unit each.
    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
    }

    /// Remove outpoints from the watch-set, releasing their quota.
    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
    }

    /// Add persistent lifecycle watches on txids (seen → confirmed → replaced /
    /// evicted / unconfirmed). `auto_close` optionally self-evicts the watch once
    /// the tx is buried. Charges one unit per txid.
    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
    }

    /// Remove lifecycle watches on txids.
    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
    }

    /// Arm single-shot depth alarms over the **cross product** of `txids` and
    /// `depths`: each txid fires `TxidDepthReached` once it is `depth`
    /// confirmations deep, then self-evicts. Charges one unit per (txid, depth).
    ///
    /// Depths must be `>= 1`; depths `< 1` are dropped client-side. If that
    /// leaves no valid depths (or no txids), this is a no-op — importantly, it
    /// does **not** send an empty `min_depths`, which the server would otherwise
    /// reinterpret as a *lifecycle* add rather than a depth alarm.
    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
    }

    /// Remove depth alarms over the cross product of `txids` and `depths`.
    /// Depths `< 1` are dropped client-side; an all-invalid (or empty) call is a
    /// no-op and never sends an empty `min_depths` (which would target lifecycle
    /// watches instead).
    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
    }

    /// Expand a public output descriptor into a watch-set over the window
    /// `[start, start + gap_limit)`. The server retains the descriptor →
    /// scripthashes membership, so the client owns gap-limit advancement by
    /// re-sending with an advanced `start` (the server reconciles the slid
    /// window) and can drop the whole window with
    /// [`remove_descriptor`](Self::remove_descriptor).
    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
    }

    /// Remove a descriptor previously added with [`add_descriptor`](Self::add_descriptor),
    /// releasing every scripthash its window contributed whose last owner this
    /// drops. A scripthash the descriptor shares with a direct `add_scripts` or
    /// another descriptor stays watched. `descriptor` must byte-match the string
    /// it was added with; removing an unknown descriptor is a no-op.
    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
    }

    /// Add privacy-preserving script-prefix buckets: each is a `bits`-bit prefix
    /// of `sha256(scriptPubKey)`, carried as the top `ceil(bits/8)` bytes. The
    /// server delivers every tx in the `2^-bits` bucket and the client filters
    /// locally, so the server learns only the bucket. Charged by coarseness
    /// (smaller `bits` costs more). `bits` must be in `1..=32` (the server
    /// buckets on the top 32 bits of the scripthash; wider is meaningless and
    /// silently dropped) and `prefix` exactly `ceil(bits/8)` bytes; the server
    /// additionally enforces its configured
    /// `[streamprefixminbits, streamprefixmaxbits]` range.
    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
    }

    /// Remove script-prefix buckets, releasing their quota.
    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
    }

    /// Register BIP 352 scan-key watch targets (Tier 2). The node runs the ECDH
    /// match server-side and pushes an
    /// [`Event::SilentPaymentMatched`](crate::Event::SilentPaymentMatched) for
    /// every output paying one of them. Up to 16 targets per connection; each
    /// charges one watch-quota unit. Re-registering an existing target (same
    /// `b_scan·G`) refreshes its labels.
    ///
    /// The `scan_secret` is a watch credential disclosed to the node — see
    /// [`SilentPaymentTarget`]. With the `bitcoin` feature (the default) each
    /// target's `scan_secret` and `spend_pubkey` are validated as curve values
    /// up front (see [`SilentPaymentTarget::validate`]), so a malformed target is
    /// a deterministic [`InvalidArgument`](StreamError::InvalidArgument) here
    /// rather than a silent server-side skip that returns `Ok(())` while
    /// installing no watch. Without the feature this helper moves bytes only —
    /// validate the field widths yourself or let the server reject a malformed
    /// target.
    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
    }

    /// Remove scan-key watch targets by their identity `b_scan·G` (33-byte
    /// compressed, from [`SilentPaymentTarget::scan_pubkey`]), releasing quota.
    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
    }

    /// Adjust the live firehose category bitfield (see [`Categories`]). Applies
    /// immediately; does not affect the watch-set.
    pub async fn set_categories(&self, categories: u32) -> Result<(), StreamError> {
        self.send_msg(pb::subscribe_control::Msg::SetCategories(pb::SetCategories {
            categories,
        }))
        .await
    }

    /// Set per-stream delivery options. With `include_raw_tx = true`, subsequent
    /// [`Event::ScriptMatched`](crate::Event::ScriptMatched) on this stream carry
    /// the full serialized matching transaction in
    /// [`raw_tx`](crate::Event::ScriptMatched); `false` restores the default
    /// (empty). Applies immediately; does not affect the watch-set. Bandwidth-
    /// heavy — the value fields (`amount`) already cover the common case, so
    /// enable this only when you need the whole transaction.
    ///
    /// When driving a [`ResilientWatch`](crate::ResilientWatch), prefer its
    /// [`set_watch_options`](crate::ResilientWatch::set_watch_options) so the
    /// opt-in is re-applied across reconnects.
    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
    }

    /// Mid-stream re-anchor: replay confirmed history forward from `cursor`, then
    /// resume live, without tearing down the watch-set. Rate-limited per
    /// principal; only one re-anchor drains at a time.
    ///
    /// `Ok(())` means the request was written to the control stream — **not**
    /// that the re-anchor ran. The outcome arrives **in-band on the event
    /// stream** as exactly one of (#439):
    ///
    /// - [`Event::CursorAccepted`](crate::Event::CursorAccepted) — admitted;
    ///   confirmed-history replay follows it (watch for `clamped` to learn the
    ///   server truncated the lower end and you must full-resync below
    ///   `earliest_replayed`).
    /// - [`Event::CursorRejected`](crate::Event::CursorRejected) — declined
    ///   (rate-limited, concurrent re-anchor, empty cursor, or no block source);
    ///   the live stream is unchanged. Retry, back off, or escalate to a full
    ///   resnapshot per the [`reason`](crate::CursorRejectReason).
    ///
    /// A consumer that needs at-least-once delivery should drive its catch-up
    /// state machine off these events rather than treating `Ok(())` as success.
    pub async fn set_cursor(&self, cursor: Cursor) -> Result<(), StreamError> {
        self.send_msg(pb::subscribe_control::Msg::SetCursor(pb::SetCursor {
            cursor: Some(cursor),
        }))
        .await
    }

    /// Request a bounded historical rescan of the current watch-set over the
    /// inclusive height range `[from_height, to_height]`. The outcome arrives
    /// in-band on the event stream as [`Event`](crate::Event)`::RescanAccepted` /
    /// `RescanRejected`; on accept, confirmed watch-matches for the range follow
    /// in height order, terminated by `RescanComplete`.
    ///
    /// A side query: it does not move the durable cursor and runs independently
    /// of the live tail or any in-flight re-anchor. The server span-caps the
    /// range and admits at most one rescan at a time (a second is rejected
    /// `ConcurrentRescan`). `Ok(())` means the request was sent, not that it was
    /// admitted — drive off the in-band result.
    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
    }
}

/// The maximum meaningful prefix width. The server buckets on the top 32 bits of
/// `sha256(scriptPubKey)` (its mask saturates at 32) and its default
/// `streamprefixmaxbits` is 32, so a `bits > 32` registration can never be more
/// selective and is silently dropped server-side (the control path has no ack).
/// We reject it client-side rather than let a caller believe a watch was
/// installed. A server may further *lower* the max via `streamprefixmaxbits`;
/// that bound is not advertised over the wire, so an over-precise (but ≤ 32)
/// prefix can still be silently dropped — there is no client-side signal for it.
pub(crate) const MAX_PREFIX_BITS: u32 = 32;

/// Validate a prefix/bits pair before it reaches the wire.
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 })
}

/// Whether `endpoint` carries the `https://` scheme.
///
/// tonic selects TLS from the URI scheme alone, so the scheme — not the
/// builder's `tls` field — is what decides whether bytes are encrypted. A
/// scheme-less endpoint (`host:port`, the common form) is plaintext.
///
/// Only the TLS-capable build consults this: without the `tls` feature the
/// scheme carries no security meaning at all, which is the whole point of
/// [`endpoint_is_encrypted`]. Kept compiled in both so the unit test that
/// pins its behaviour runs in both.
#[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"))
}

/// Whether connecting to `endpoint` will actually encrypt the bytes.
///
/// This is *not* the same question as [`endpoint_is_https`], and conflating the
/// two is how a credential leaks. tonic only honours the `https://` scheme when
/// it is itself compiled with its `tls` feature: the scheme test and the whole
/// TLS branch of its connector are `#[cfg(feature = "tls")]`, and it disables
/// `enforce_http`. In a build without that feature an `https://` URI therefore
/// opens an ordinary TCP connection and speaks cleartext h2c — no error, no
/// downgrade warning. So the scheme is transport-truthful only in a TLS-capable
/// build; in a build without one, nothing is, and this returns `false` for every
/// endpoint.
#[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
}

/// Strip `user:password@` userinfo out of an endpoint before it is put into an
/// error string. Refusals are normally logged, and a password in a URL is a
/// credential too — an error whose whole purpose is to stop a credential
/// reaching the wire must not spill a different one into the log.
fn redact_endpoint_userinfo(endpoint: &str) -> String {
    let Some((scheme, rest)) = endpoint.split_once("://") else {
        // No scheme, so the whole string is an authority (`host:port`).
        return match endpoint.rsplit_once('@') {
            Some((_, host)) => format!("***@{host}"),
            None => endpoint.to_string(),
        };
    };
    // Userinfo lives in the authority only: everything before the first `/`.
    // Splitting there keeps a later `@` in a path from being mistaken for it.
    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(),
    }
}

/// The endpoint as it should appear in an [`StreamError::InsecureCredential`],
/// naming the reason when the reason is not visible in the endpoint itself.
#[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)
    )
}

/// TLS settings assembled by the `StreamClientBuilder::tls*` methods and applied
/// in [`connect`](StreamClientBuilder::connect). `Some(..)` on the builder means
/// TLS is enabled; the fields refine trust and identity.
#[cfg(feature = "tls")]
#[derive(Clone, Default)]
struct TlsSettings {
    /// PEM CA (or self-signed leaf) to verify the server cert. `None` → trust the
    /// bundled Mozilla webpki roots (public-CA servers).
    ca_pem: Option<Vec<u8>>,
    /// mTLS client identity: `(cert_pem, key_pem)`.
    identity: Option<(Vec<u8>, Vec<u8>)>,
    /// SNI / certificate domain-name override (e.g. when connecting by IP).
    domain: Option<String>,
}

/// Builder for a [`StreamClient`].
///
/// `Debug` is hand-written to redact the bearer token (and never leak TLS key
/// material) — never derive it here.
#[derive(Clone)]
pub struct StreamClientBuilder {
    endpoint: String,
    token: Option<String>,
    /// Set only by [`StreamClientBuilder::insecure_bearer_token`]: the caller
    /// has explicitly accepted sending the token over an unencrypted endpoint.
    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>"))
            // Not the token, but worth seeing in a bug report: it says whether
            // the plaintext waiver is in force.
            .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 {
    /// Attach a bearer token, sent as `authorization: Bearer <token>` metadata
    /// on every RPC.
    ///
    /// The token is only honored when the server enforces auth
    /// (`-eventsgrpcauth`); a no-auth (loopback-trust) server ignores it.
    ///
    /// **The endpoint must be encrypted.** [`connect`](Self::connect) returns
    /// [`StreamError::InsecureCredential`] for a token combined with a
    /// non-`https://` endpoint rather than putting it on the wire in the clear.
    /// Pair this with [`tls`](Self::tls) or [`tls_ca_pem`](Self::tls_ca_pem),
    /// or use [`insecure_bearer_token`](Self::insecure_bearer_token) to accept
    /// the risk explicitly.
    pub fn bearer_token(mut self, token: impl Into<String>) -> Self {
        self.token = Some(token.into());
        self.allow_insecure_token = false;
        self
    }

    /// Like [`bearer_token`](Self::bearer_token), but permits sending the token
    /// over an unencrypted endpoint.
    ///
    /// Only for loopback and test harnesses. Over anything else the token is
    /// readable by every host on the path, and so is everything the stream
    /// carries — including the BIP 352 scan keys a Tier 2 watch registers,
    /// which disclose which outputs belong to the receiver.
    ///
    /// This is a separate method rather than a flag so that the unsafe choice
    /// has to be named at the call site, and so that a later switch back to
    /// [`bearer_token`](Self::bearer_token) cannot silently leave the waiver
    /// behind.
    pub fn insecure_bearer_token(mut self, token: impl Into<String>) -> Self {
        self.token = Some(token.into());
        self.allow_insecure_token = true;
        self
    }

    /// Configure client-side HTTP/2 keepalive. Defaults match the server
    /// (30s interval / 20s timeout) when [`keepalive_default`](Self::keepalive_default)
    /// is used; this overrides with explicit seconds.
    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
    }

    /// Use the server-matching keepalive defaults (30s interval / 20s timeout).
    pub fn keepalive_default(self) -> Self {
        self.keepalive(30, 20)
    }

    /// Accept a node two or more minor versions older than this SDK.
    ///
    /// By default [`subscribe`](StreamClient::subscribe) and
    /// [`watch`](StreamClient::watch) refuse such a node with
    /// [`StreamError::NodeTooOld`]. With this set they log the same warning as
    /// for a node one minor version behind and carry on. Meant for a rolling
    /// upgrade where the SDK side went first. Features added after the node's
    /// release are still unavailable, and request fields it does not recognise
    /// are still ignored.
    ///
    /// This does not bypass [`StreamError::SchemaMismatch`].
    pub fn allow_old_node(mut self) -> Self {
        self.allow_old_node = true;
        self
    }

    /// Enable TLS for the connection, trusting the bundled Mozilla root CAs
    /// (for a server with a publicly-trusted certificate). For a private or
    /// self-signed CA — the usual case for a satd node — use
    /// [`tls_ca_pem`](Self::tls_ca_pem) instead. Requires the `tls` feature
    /// (on by default).
    #[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
    }

    /// Enable TLS and verify the server certificate against the PEM CA (or
    /// self-signed leaf) in `pem` — the usual choice for a satd node serving its
    /// own certificate. Replaces the bundled roots for this connection.
    #[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
    }

    /// Enable mutual TLS: present this PEM client certificate + private key to a
    /// server configured for mTLS (`-eventsgrpcmtls`). Combine with
    /// [`tls_ca_pem`](Self::tls_ca_pem) to pin the server's CA — without it the
    /// *server* certificate is verified against the bundled public Mozilla roots,
    /// so a self-signed satd node (the usual mTLS case) will fail the handshake
    /// unless you also call `tls_ca_pem`.
    #[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
    }

    /// Override the certificate domain name verified during the TLS handshake
    /// (SNI). Use when the endpoint host differs from the certificate subject —
    /// e.g. connecting by IP, or through a proxy. Enables TLS if not already.
    #[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
    }

    /// Connect the transport and return a ready [`StreamClient`].
    pub async fn connect(self) -> Result<StreamClient, StreamError> {
        // Captured before the match below moves `self.token`; the credential
        // check that reads it is ordered later on purpose (see there).
        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,
        };

        // TLS is selected by tonic purely from the URI scheme, *not* from
        // whether a `ClientTlsConfig` is attached: an `http://` (or scheme-less)
        // endpoint silently connects in cleartext even with TLS configured,
        // leaking the bearer token and the whole event stream while the caller
        // believes the link is encrypted. Fail closed rather than downgrade.
        #[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(),
            ));
        }

        // A bearer token on an unencrypted endpoint crosses the network in the
        // clear, and so does everything the stream carries — including the BIP
        // 352 scan keys a Tier 2 watch registers. Nothing downstream catches
        // this: the token is attached as ordinary metadata, so tonic has no
        // credential to apply a transport-security requirement to, and a
        // remote-bound node with `eventsgrpcauth=1` and no TLS is a supported
        // server configuration. Both ends behave as configured while the
        // credential is on the wire (#521).
        //
        // The predicate is `endpoint_is_encrypted`, not the builder's `tls`
        // field and not the scheme on its own: tonic selects TLS from the URI
        // scheme, but only in a build that has its `tls` feature. Without that
        // feature an `https://` endpoint connects in cleartext rather than
        // failing, so this crate's own `tls` feature is part of the question.
        // See `endpoint_is_encrypted`.
        //
        // Ordered after the TLS-scheme check above so the more specific
        // diagnosis wins when a caller asked for TLS *and* gave an http:// URL,
        // matching the Go SDK, where the scheme conflict is likewise reported
        // first.
        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();
            // A pinned CA verifies exactly that authority (the satd self-signed
            // case); otherwise fall back to the bundled public roots.
            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(),
        })
    }
}

/// Install `ring` as the process-default rustls `CryptoProvider`, once. In this
/// workspace `tonic/tls` resolves to ring only, so rustls already auto-selects
/// it; this is belt-and-suspenders for a build that *also* compiled `aws-lc-rs`
/// (a downstream consumer that doesn't pin the provider), where the default
/// would otherwise be ambiguous and panic. Best-effort — if the application
/// already installed a provider, that one stays.
#[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();
    });
}

/// An async client for the satd `satd.events.v1` streaming API.
///
/// `Debug` is hand-written to redact the auth credential — never derive it here.
#[derive(Clone)]
pub struct StreamClient {
    inner: NodeEventStreamClient<Channel>,
    auth: Option<MetadataValue<Ascii>>,
    allow_old_node: bool,
    /// Shared by clones, so the resilient layers (which hold a clone) update
    /// [`node_version`](Self::node_version) and warn at most once per node
    /// version for the whole client.
    compat: Arc<Mutex<CompatState>>,
}

/// What the client last learned from a node's response headers.
#[derive(Debug, Default)]
struct CompatState {
    /// The raw `satd-version` header from the most recent stream.
    node_version: Option<String>,
    /// The node version string the old-node warning was last logged for
    /// (`Some(None)` for a node that sent no header).
    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 {
    /// Start building a client for `endpoint` (e.g. `http://node:50051`, or
    /// `https://node:50051` with [`tls`](StreamClientBuilder::tls)).
    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,
        }
    }

    /// The version the node advertised on the most recently opened stream, e.g.
    /// `"0.6.0"`.
    ///
    /// `None` until a stream has been opened, and when the node predates the
    /// `satd-version` header (0.5.x and older).
    pub fn node_version(&self) -> Option<String> {
        self.compat.lock().unwrap_or_else(|e| e.into_inner()).node_version.clone()
    }

    /// Apply the SDK ↔ node compatibility rule to a stream-opening response's
    /// headers (`STABILITY_POLICY.md` → "Streaming API & SDK compatibility").
    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");

        // Decide under the lock, log after releasing it: a `tracing` subscriber
        // is user code, and one that calls `node_version()` (or blocks) must not
        // deadlock or stall every other stream open on this client.
        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,
                    });
                }
            };
            // Once per observed version: a reconnect loop against the same node
            // stays quiet, a node upgraded (or downgraded) in between warns again.
            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(())
    }

    /// Wrap a message in a request carrying the configured auth metadata.
    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
    }

    /// Open a server-streaming firehose. Requires the `stream:subscribe`
    /// capability when the server enforces auth.
    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()))
    }

    /// Open a reconnect-and-replay-aware firehose. Unlike [`subscribe`](Self::subscribe),
    /// the returned [`ResilientSubscription`](crate::ResilientSubscription)
    /// reconnects with backoff, persists and replays the resume cursor, recovers
    /// from `Lagged` per the [`LagPolicy`](crate::LagPolicy), and surfaces
    /// replay-truncation gaps. Connects lazily on the first
    /// [`next`](crate::ResilientSubscription::next).
    pub fn resilient_subscribe(
        &self,
        opts: SubscribeOptions,
        config: crate::ResilientConfig,
    ) -> crate::ResilientSubscription {
        crate::ResilientSubscription::new(self.clone(), opts, config)
    }

    /// Open a reconnect-and-replay-aware bidirectional watch. Unlike
    /// [`watch`](Self::watch), the returned [`ResilientWatch`](crate::ResilientWatch)
    /// mirrors the watch-set and re-registers it on reconnect, re-anchors off the
    /// deterministic `set_cursor` results, and persists the resume cursor.
    /// Connects lazily on the first [`next`](crate::ResilientWatch::next).
    pub fn resilient_watch(&self, config: crate::ResilientWatchConfig) -> crate::ResilientWatch {
        crate::ResilientWatch::new(self.clone(), config)
    }

    /// Open a bidirectional watch stream, returning a [`WatchHandle`] to send
    /// control messages and an [`EventStream`] of matches + firehose.
    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 {
    /// A handle wired to an inspectable receiver — for unit tests in sibling
    /// modules (e.g. `ResilientWatch::reload`) that assert the control messages a
    /// layer sends. No server needed.
    pub(crate) fn for_test() -> (Self, mpsc::Receiver<pb::SubscribeControl>) {
        let (tx, rx) = mpsc::channel(64);
        (WatchHandle { tx }, rx)
    }
}

#[cfg(test)]
impl StreamClient {
    /// Build a client over a lazily-connected channel to a dummy endpoint, for
    /// unit tests of layers (e.g. [`ResilientSubscription`](crate::ResilientSubscription))
    /// that need a `StreamClient` value but drive their logic via injected events
    /// and never actually poll the transport. `connect_lazy` performs no I/O, so
    /// no server is required; it must be called from within a Tokio runtime.
    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;

    /// A handle wired to a receiver we can inspect — no server needed.
    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"));
        // The common scheme-less form. tonic treats it as plaintext, so this
        // must not be mistaken for encrypted.
        assert!(!endpoint_is_https("node.example:50051"));
        assert!(!endpoint_is_https("127.0.0.1:50051"));
    }

    /// The scheme decides only in a build that can actually do TLS. tonic gates
    /// its whole https branch on its own `tls` feature and turns `enforce_http`
    /// off, so without that feature an `https://` URI connects in cleartext h2c
    /// instead of failing — which would have let a token past the gate below in
    /// exactly the configuration the gate advertises as safe.
    #[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"));
    }

    /// The refusal is normally logged, so it must not spill a password that was
    /// embedded in the endpoint. A later `@` in the path is not userinfo.
    #[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");
        // Untouched when there is nothing to redact.
        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}");
    }

    /// The #521 guard. The token is attached as ordinary metadata, so there is
    /// no `PerRPCCredentials` for tonic to apply a transport-security
    /// requirement to, and a remote-bound node with `eventsgrpcauth=1` and no
    /// TLS is a supported server configuration — nothing else stops the
    /// credential reaching the wire in the clear.
    #[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}"
            );
        }
    }

    /// Reaching the transport is what proves the gate did not fire. Asserting
    /// merely "not `InsecureCredential`" would also hold if `connect()` started
    /// failing for some unrelated wrong reason, so name the errors a connect to
    /// a dead port is allowed to produce.
    fn assert_reached_the_transport(err: &StreamError, ctx: &str) {
        assert!(
            matches!(err, StreamError::Connect(_) | StreamError::Transport(_)),
            "{ctx}: expected a transport/connect error, got {err:?}"
        );
    }

    /// A plaintext connection with no token is the default local setup and is
    /// unaffected: it gets as far as the transport (and fails there, since
    /// nothing is listening on port 1).
    #[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");
    }

    /// The named waiver gets past the gate — it too fails at the transport,
    /// which is the proof that the credential check no longer short-circuits.
    #[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");
    }

    /// A later `bearer_token` must clear an earlier waiver rather than inherit
    /// it — the reason the waiver is a token-carrying method and not a flag.
    #[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:?}"
        );
    }

    /// An `https://` endpoint is accepted with a token whether or not a `tls*`
    /// method was called — tonic selects TLS from the scheme, so the scheme is
    /// what decides.
    #[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() {
        // A default subscription sends no tweak knobs (byte-identical to one
        // built before they existed); each flag maps to Some(true) 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);
                // Unfloored entry becomes a 0 floor to keep the vec parallel.
                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(); // empty depths
        h.add_depth_alarms([], [3]).await.unwrap(); // empty txids
        // depths < 1 are filtered out; an all-invalid call must NOT send an
        // empty min_depths (the server would treat that as a lifecycle add).
        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]); // cross product, server-side
                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();
        // 16 bits needs 2 bytes; give 1.
        let err = h.add_script_prefixes([(vec![0xab], 16)]).await.unwrap_err();
        assert!(matches!(err, StreamError::InvalidArgument(_)));
        // bits out of range (too small).
        let err = h.add_script_prefixes([(vec![], 0)]).await.unwrap_err();
        assert!(matches!(err, StreamError::InvalidArgument(_)));
        // bits above the server's 32-bit ceiling: rejected client-side rather
        // than sent and silently dropped by the server (5 bytes = ceil(33/8)).
        let err = h.add_script_prefixes([(vec![0u8; 5], 33)]).await.unwrap_err();
        assert!(matches!(err, StreamError::InvalidArgument(_)));
        // Nothing reached the wire.
        assert!(rx.try_recv().is_err());

        // Valid boundary: exactly 32 bits, 4 bytes.
        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:?}"),
        }

        // Valid: 16 bits, 2 bytes.
        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:?}"),
        }
    }

    /// The builder's `Debug` must show only that TLS is enabled — never the CA
    /// bytes, the client key, or the bearer token.
    #[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"));
    }

    /// The TLS connect path assembles a rustls config and installs the `ring`
    /// provider without panicking; against a dead port it fails cleanly (a
    /// transport error, not `InvalidEndpoint`/`InvalidToken`). Exercises the
    /// whole `tls()` → `ClientTlsConfig::with_webpki_roots` → `tls_config`
    /// assembly end to end, no server required.
    #[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:?}",
        );
    }

    /// TLS requested over a plaintext `http://` endpoint must fail closed at
    /// `connect()` rather than silently connecting in cleartext — tonic selects
    /// TLS from the URI scheme alone, so without this guard the bearer token and
    /// event stream would leak on the wire. Covers `http://` and a scheme-less
    /// endpoint; the `https://` happy path is exercised above.
    #[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:?}"),
            }
        }
    }
}

/// The node-version check against an in-process tonic server that sends
/// whatever version headers a test asks for.
#[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>>;

    /// A node stand-in. `version` can be changed between calls to model a node
    /// upgraded underneath a client.
    #[derive(Clone)]
    struct FakeNode {
        version: Arc<Mutex<Option<String>>>,
        schema: Option<&'static str>,
        /// When set, every stream delivers one heartbeat with this schema.
        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()
    }

    /// A node version `n` minor versions behind this build's SDK, derived so the
    /// tests keep meaning the same thing after every version bump. `None` when
    /// the SDK's minor is too small for that.
    fn minors_behind(n: u32) -> Option<String> {
        let sdk = sdk();
        (sdk.minor >= n).then(|| format!("{}.{}.0", sdk.major, sdk.minor - n))
    }

    /// A node version the SDK refuses: two minors behind, or failing that a
    /// major behind.
    fn too_old() -> String {
        minors_behind(2).unwrap_or_else(|| format!("{}.9.0", sdk().major - 1))
    }

    thread_local! {
        /// Where the test subscriber writes this thread's events.
        static THREAD_LOGS: std::cell::RefCell<Vec<u8>> =
            const { std::cell::RefCell::new(Vec::new()) };
    }

    struct ThreadLogs;

    thread_local! {
        /// Set by a test that wants every event checked against its client's
        /// compat lock. Per-thread, so it describes only the test that armed
        /// it even while other tests are logging on their own threads.
        static PROBE_COMPAT: std::cell::RefCell<Option<Arc<Mutex<CompatState>>>> =
            const { std::cell::RefCell::new(None) };
        /// One entry per event seen while the probe is armed: was the compat
        /// lock free when it was logged?
        static PROBE_FREE: std::cell::RefCell<Vec<bool>> =
            const { std::cell::RefCell::new(Vec::new()) };
    }

    /// Records, for each event on an arming thread, whether the client's
    /// compat lock was free. Part of the one global subscriber rather than a
    /// scoped one for the reason given on [`Logs`].
    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));
                }
            });
        }
    }

    /// Arms [`ProbeLock`] for this thread and disarms it on drop.
    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(())
        }
    }

    /// Captures `tracing` output for the duration of a test.
    ///
    /// One process-wide subscriber, installed once, writing into a per-thread
    /// buffer -- deliberately not a scoped `set_default` per test. Whether a
    /// `warn!` is evaluated at all is decided by tracing's process-wide
    /// callsite interest and max-level hint, both of which a scoped
    /// subscriber on *any* thread moves. Two of these tests running at once
    /// could therefore leave one with an empty capture, which is how
    /// `allow_old_node_downgrades_the_refusal_to_a_warning` failed in CI
    /// asserting one warning against none. libtest gives each test its own
    /// thread, so a per-thread buffer keeps them apart without a lock.
    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());
    }

    /// A node that sends no headers is taken to be 0.5; what that means depends
    /// on this build's version, so the expectation is derived from the rule.
    #[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:?}");
    }

    /// A node that predates the headers still stamps `schema_version` on every
    /// event; the first one is checked.
    #[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:?}");

        // The matching schema passes through as an ordinary event.
        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();
        // At a `.0` minor there is no one-behind in the same major; the
        // downgraded refusal logs the same warning.
        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");
        // A clone (as the resilient layers hold) shares the state.
        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:?}");

        // The node changes version (a rolling downgrade, say): warn again.
        *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());
    }

    /// The warning is logged with the compat lock released. A subscriber is user
    /// code: one that reads `node_version()` from inside the event would
    /// otherwise deadlock, and a slow one would stall every stream open.
    #[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);
    }
}