liminal-sdk 0.19.1

//! Setup and delivery contracts, including every production subscription module.

use super::*;

/// The named setup deadline SDK-010 installs: 5 s, the estate's already-ratified
/// constant generalized, never a new default.
const NAMED_SETUP_DEADLINE: Duration = Duration::from_secs(5);

/// A control-frame reply that is slow but well-behaved: past the retired 100 ms
/// `READER_POLL_TIMEOUT` cadence, far inside the named 5 s deadline. The delay is
/// the fixture's INTENDED slow reply, not a proof device.
// The fixed delay is the slow reply being tested, not fixture synchronisation.
// Readiness and teardown use their own signals independently.
const SLOW_BUT_ANSWERED: Duration = Duration::from_millis(250);

/// Runs a fake server that answers the handshake and the subscribe, optionally
/// delaying each control reply, then holds the socket until the client tears it
/// down. The owned server is released after the consumer finishes.
fn spawn_setup_server(reply_delay: Duration) -> Result<ScriptedServer, SdkError> {
    ScriptedServer::spawn(move |mut socket| {
        let mut buffer = Vec::new();
        read_and_discard_one(&mut socket, &mut buffer)?;
        std::thread::sleep(reply_delay);
        write_frame(
            &mut socket,
            &Frame::ConnectAck {
                flags: 0,
                selected_version: CLIENT_MAX_VERSION,
                capabilities: 0,
            },
        )?;
        read_and_discard_one(&mut socket, &mut buffer)?;
        std::thread::sleep(reply_delay);
        write_frame(
            &mut socket,
            &Frame::SubscribeAck {
                flags: 0,
                stream_id: SUBSCRIPTION_STREAM_ID,
                subscription_id: 42,
                selected_schema: SchemaId::new([7; SchemaId::WIRE_LEN]),
            },
        )?;
        // Read to the client's teardown so the socket outlives the assertions.
        wait_for_client_close(&mut socket)
    })
}

/// RED PIN (SDK-010 R2, direction (b)) — the disarm, pinned behaviorally.
///
/// `connect_socket` arms the retired 100 ms `READER_POLL_TIMEOUT` on this socket
/// and never takes it off, so the steady-state reader wakes ten times a second
/// forever on a quiet subscription — family F9 in the wiring ledger.
/// `read_timeout()` reads the live `SO_RCVTIMEO` off the very kernel socket the
/// reader thread blocks on (the writer handle and the reader's handle are
/// `try_clone` siblings), so this observes behaviour, not source.
///
/// The assertion is `None`, not "longer than before": a leaked setup deadline
/// would report `Some(5s)` and fail exactly as `Some(100ms)` does.
#[test]
fn the_subscription_reader_carries_no_read_window_in_steady_state() -> Result<(), SdkError> {
    let server = spawn_setup_server(Duration::ZERO)?;
    server.run(|address| {
        let subscription = SubscriptionStream::open(address, "orders", Vec::new())?;
        let observed =
            subscription
                .writer
                .read_timeout()
                .map_err(|source| SdkError::Connection {
                    description: format!(
                        "failed to read the subscription socket read timeout: {source}"
                    ),
                })?;
        assert_eq!(
            observed, None,
            "the subscription reader must block with no read window once the control \
             exchange is over; a Some(_) here is a cadence, whatever its period"
        );
        drop(subscription);
        Ok(())
    })
}

/// PARITY PIN (SDK-010 R3) — the same slow-reply fixture the push reader is red
/// against, aimed at the subscription reader.
///
/// This one is GREEN before the change as well as after: `subscription.rs`'s
/// `read_one_frame` already samples the 100 ms cadence against a 5 s
/// `SETUP_TIMEOUT` total deadline instead of dying on the first tick, so a
/// 250 ms reply already survives here. It is pinned so the two TCP readers are
/// held to ONE answer after the generalization, and so a regression toward the
/// push reader's fatal-on-first-timeout shape is caught.
#[test]
fn open_survives_control_replies_slower_than_the_retired_poll_cadence() -> Result<(), SdkError> {
    let server = spawn_setup_server(SLOW_BUT_ANSWERED)?;
    server.run(|address| {
        let subscription = SubscriptionStream::open(address, "orders", Vec::new())?;
        assert_eq!(subscription.subscription_id(), 42);
        assert!(
            NAMED_SETUP_DEADLINE > SLOW_BUT_ANSWERED,
            "the fixture's slow reply must sit inside the named deadline"
        );
        drop(subscription);
        Ok(())
    })
}

/// TOMBSTONE (SDK-010 R5) — the retired reader poll family must not reappear in
/// this reader's production source.
#[test]
fn subscription_source_has_no_retired_reader_poll_family() {
    const SOURCES: [&str; 5] = [
        include_str!("subscription.rs"),
        include_str!("subscription/subscription_reader.rs"),
        include_str!("subscription/subscription_setup.rs"),
        include_str!("subscription/subscription_signals.rs"),
        include_str!("subscription/subscription_wire.rs"),
    ];
    for source in SOURCES {
        let production = source.split("#[cfg(test)]").next().unwrap_or(source);
        for forbidden in [
            "READER_POLL_TIMEOUT",
            "AtomicBool",
            "stop.load",
            "stop.store",
            "re-check the stop flag",
            "poll the stop flag",
        ] {
            assert!(
                !production.contains(forbidden),
                "retired subscription-reader poll-family source `{forbidden}` reappeared"
            );
        }
    }
}

/// The boot and id a 1.2 server gives a delivery (Fable 4c).
const DELIVERED: DeliveredId = DeliveredId {
    boot: 1_791_328_512_345,
    id: 42,
};

/// 4c: a delivery that carries the server's id names its reply's parent as
/// `<boot>:<id>`; one that carries none names no parent, never an invented one.
#[test]
fn a_delivered_message_names_its_reply_parent_only_when_it_carries_an_id() {
    let with_id = DeliveredMessage {
        delivery_seq: 1,
        schema_id: SchemaId::new([1; SchemaId::WIRE_LEN]),
        payload: Vec::new(),
        message: Some(DELIVERED),
    };
    assert_eq!(with_id.message(), Some(DELIVERED));
    assert_eq!(
        with_id.reply_parent(),
        Some(MessageId::from("1791328512345:42"))
    );
    let without = DeliveredMessage {
        message: None,
        ..with_id
    };
    assert_eq!(without.message(), None);
    assert_eq!(without.reply_parent(), None);
}

/// 4c: the subscription offers 1.2, so a 1.2 server may send the id.
#[test]
fn the_subscription_offers_the_delivered_id_version() {
    assert_eq!(CLIENT_MAX_VERSION, DELIVER_MESSAGE_ID_VERSION);
}

/// 4c: the reader keeps the id a Deliver carried.
#[test]
fn the_reader_keeps_the_delivered_id() -> Result<(), SdkError> {
    let server = ScriptedServer::spawn(move |mut socket| {
        let mut buffer = Vec::new();
        read_and_discard_one(&mut socket, &mut buffer)?;
        write_frame(
            &mut socket,
            &Frame::ConnectAck {
                flags: 0,
                selected_version: DELIVER_MESSAGE_ID_VERSION,
                capabilities: 0,
            },
        )?;
        read_and_discard_one(&mut socket, &mut buffer)?;
        let schema = SchemaId::new([7; SchemaId::WIRE_LEN]);
        write_frame(
            &mut socket,
            &Frame::SubscribeAck {
                flags: 0,
                stream_id: SUBSCRIPTION_STREAM_ID,
                subscription_id: 42,
                selected_schema: schema,
            },
        )?;
        write_frame(
            &mut socket,
            &Frame::Deliver {
                flags: 0,
                stream_id: SUBSCRIPTION_STREAM_ID,
                delivery_seq: 1,
                envelope: MessageEnvelope::new(schema, CausalContext::independent(), vec![1]),
                message: Some(DELIVERED),
            },
        )?;
        wait_for_client_close(&mut socket)
    })?;

    server.run(|address| {
        let subscription = SubscriptionStream::open(address, "orders", Vec::new())?;
        let delivered = subscription.recv_timeout(Duration::from_secs(2))?;
        assert_eq!(delivered.message(), Some(DELIVERED));
        drop(subscription);
        Ok(())
    })
}