liminal-server 0.14.2

//! The wire leg of A1 §2's capacity declaration: what a client's
//! `Subscribe.max_in_flight` becomes by the time it reaches the inbox.
//!
//! Verifier finding 4. The wiring record's "A1 bites first" rests on the A1
//! bound (`max_in_flight + max_buffer_depth`) fitting inside the §5 depth cap,
//! because `SubscriptionInbox::admit` checks the A1 band first and the fairness
//! trip second — and the two refusals are not interchangeable. An A1 Reject
//! paces one message; a §5 trip sets the sticky overflow marker and the whole
//! subscription is shed. A window large enough to lift the A1 Reject band above
//! the cap turns "your consumer is behind" into "your subscription is gone".
//!
//! On the wire path that bound is CLIENT-declared. `Frame::validate` refuses
//! only zero. So the property has to be enforced by code that compares the two
//! numbers, and this file is the instrument that says it is — through
//! `apply_frame`, on the real frame, with the install observed where it is
//! actually handed over.
//!
//! The observation is a spy that delegates every method to a real
//! `LiminalConnectionServices` and records only the `InboxInstall` it was given.
//! A test that computed the expected capacity itself would be asserting on its
//! own arithmetic; this one reads what the server produced.

use std::error::Error;
use std::sync::{Arc, Mutex};

use haematite::{Database, DatabaseConfig, EventStore};
use liminal::durability::{DurableStore, HaematiteStore};
use liminal::protocol::MessageEnvelope;
use tempfile::TempDir;

use super::*;
use crate::config::types::{LimitsConfig, ServerConfig};
use crate::server::connection::channel_registry::{ChannelAccessError, ChannelRegistration};
use crate::server::connection::conversation::ConnectionConversation;
use crate::server::connection::services::{
    ConnectionSubscription, LiminalConnectionServices, PublishOutcome,
};

/// Fixed connection pid, as in the sibling `apply_frame` unit tests.
const TEST_PID: u64 = 1;

/// A non-default depth cap, so a test that happened to pass against the
/// shipped 4096 cannot pass by coincidence: every assertion below is derived
/// from THIS number and from `DEFAULT_MAX_BUFFER_DEPTH`, never from a literal
/// bound.
const DEPTH_CAP: usize = 1_500;

/// A real adapter that records the install every `subscribe` was given.
#[derive(Debug)]
struct InstallSpy {
    inner: Arc<LiminalConnectionServices>,
    seen: Mutex<Vec<Option<liminal::pressure::ConsumerCapacity>>>,
}

impl InstallSpy {
    fn capacities(&self) -> Vec<Option<liminal::pressure::ConsumerCapacity>> {
        self.seen
            .lock()
            .map(|seen| seen.clone())
            .unwrap_or_default()
    }
}

impl ConnectionServices for InstallSpy {
    fn admit_channel(
        &self,
        operation: ChannelOperation,
        channel: &str,
    ) -> Result<(), ChannelAccessError> {
        ConnectionServices::admit_channel(self.inner.as_ref(), operation, channel)
    }

    fn publish(
        &self,
        channel: &str,
        envelope: &MessageEnvelope,
        idempotency_key: Option<&str>,
    ) -> Result<PublishOutcome, ServerError> {
        self.inner.publish(channel, envelope, idempotency_key)
    }

    fn subscribe(
        &self,
        channel: &str,
        accepted_schemas: &[ProtocolSchemaId],
        install: Option<liminal::channel::InboxInstall>,
    ) -> Result<ConnectionSubscription, ServerError> {
        if let Ok(mut seen) = self.seen.lock() {
            seen.push(install.as_ref().and_then(|install| install.capacity));
        }
        self.inner.subscribe(channel, accepted_schemas, install)
    }

    fn unsubscribe(&self, subscription: ConnectionSubscription) -> Result<(), ServerError> {
        self.inner.unsubscribe(subscription)
    }

    fn open_conversation(
        &self,
        conversation_id: u64,
        subject: &str,
    ) -> Result<ConnectionConversation, ServerError> {
        self.inner.open_conversation(conversation_id, subject)
    }

    fn conversation_message(
        &self,
        conversation: &ConnectionConversation,
        envelope: &MessageEnvelope,
        op_id: Option<u64>,
    ) -> Result<(), ServerError> {
        self.inner
            .conversation_message(conversation, envelope, op_id)
    }

    fn close_conversation(&self, conversation: ConnectionConversation) -> Result<(), ServerError> {
        self.inner.close_conversation(conversation)
    }

    fn flush_durable_state(&self) -> Result<(), ServerError> {
        self.inner.flush_durable_state()
    }
}

fn config() -> ServerConfig {
    ServerConfig {
        listen_address: std::net::SocketAddr::from(([127, 0, 0, 1], 0)),
        health_listen_address: std::net::SocketAddr::from(([127, 0, 0, 1], 0)),
        drain_timeout_ms: 30_000,
        channels: Vec::new(),
        routing_rules: Vec::new(),
        persistence_path: None,
        cluster: None,
        auth: None,
        services: crate::config::types::ServicesConfig::default(),
        limits: LimitsConfig {
            max_channels: Some(4),
            max_subscription_inbox_depth: DEPTH_CAP,
            ..LimitsConfig::default()
        },
        participant: None,
        websocket: None,
    }
}

fn subscribe_frame(channel: &str, max_in_flight: u32) -> Frame {
    Frame::Subscribe {
        flags: 0,
        stream_id: 5,
        channel: channel.to_owned(),
        accepted_schemas: Vec::new(),
        max_in_flight,
    }
}

/// Builds the spy over a real adapter with one registered channel, and returns
/// it with the runtime `apply_frame` needs.
fn fixture() -> Result<(Arc<InstallSpy>, ConnectionRuntime, TempDir), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let database = Database::create(DatabaseConfig {
        data_dir: dir.path().join("db"),
        shard_count: 4,
        distributed: None,
        executor_threads: None,
        node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
    })?;
    let store: Arc<dyn DurableStore> =
        Arc::new(HaematiteStore::new(Arc::new(EventStore::new(database))));
    let inner = Arc::new(LiminalConnectionServices::from_config_with_store(
        &config(),
        store,
    )?);
    inner.register_channel(&ChannelRegistration {
        name: "orders".to_owned(),
        schema_bytes: None,
        durable: false,
    })?;
    let spy = Arc::new(InstallSpy {
        inner,
        seen: Mutex::new(Vec::new()),
    });
    let runtime = ConnectionRuntime::for_tests_with_limits(
        Arc::clone(&spy) as Arc<dyn ConnectionServices>,
        config().limits,
    );
    Ok((spy, runtime, dir))
}

/// **PIN — the A1 bound the wire installs always fits inside the §5 depth cap.**
///
/// Three declarations through the real `Subscribe` frame: one that already
/// fits, one exactly at the boundary, and one far above it. The invariant is
/// checked as a predicate over the capacity the server actually installed,
/// against the configured cap — never against a literal, so moving
/// `max_subscription_inbox_depth` or the default buffer band moves the
/// assertion with it. This is the difference between an instruction and a
/// control.
#[test]
fn a_declared_window_is_clamped_so_the_a1_band_stays_under_the_depth_cap()
-> Result<(), Box<dyn Error>> {
    let (spy, runtime, _dir) = fixture()?;
    let policy_band = liminal::channel::DEFAULT_MAX_BUFFER_DEPTH;
    let headroom = DEPTH_CAP - policy_band;

    // A declaration that fits, one at the boundary, and one far past it. The
    // last is the shape the wire path took VERBATIM before this clamp existed.
    let declared = [16_u32, u32::try_from(headroom)?, 1_000_000];
    for window in declared {
        let mut state = ConnectionProcessState::default();
        let action = apply_frame(
            TEST_PID,
            &runtime,
            &mut state,
            subscribe_frame("orders", window),
        );
        assert!(
            matches!(action, FrameAction::Respond(Frame::SubscribeAck { .. })),
            "declaring {window} must be accepted, not refused — got {action:?}"
        );
    }

    let installed = spy.capacities();
    assert_eq!(
        installed.len(),
        declared.len(),
        "every subscribe reached the service with an install"
    );
    for (window, capacity) in declared.iter().zip(installed) {
        let capacity = capacity.ok_or("the wire path must install a capacity")?;
        assert!(
            capacity.max_in_flight > 0,
            "a zero window would Reject every publish (declared {window})"
        );
        assert!(
            capacity.max_in_flight + capacity.max_buffer_depth <= DEPTH_CAP,
            "declaring {window} installed {capacity:?}, whose A1 bound is above \
             the §5 depth cap {DEPTH_CAP} — the terminal fairness shed would \
             fire before the non-terminal A1 Reject"
        );
        // The clamp is not a blunt instrument: the bus's policy band survives,
        // and a declaration that already fits is passed through.
        assert_eq!(
            capacity.max_buffer_depth, policy_band,
            "the buffer band is bus policy and must not be given away"
        );
        let expected = std::cmp::min(usize::try_from(*window)?, headroom);
        assert_eq!(
            capacity.max_in_flight, expected,
            "declaring {window} must install exactly min(declared, headroom)"
        );
    }
    Ok(())
}