liminal-server 0.24.5

//! A real broker's quiet drains and deliveries end on the transport fence, never
//! on a read window closing: pinned by the handle's pump counts. The printed
//! latencies are information, never a predicate.

use std::time::Instant;

use super::{
    ClientRequest, Error, G1_CONVERSATION, G1Participants, Generation, ParticipantDelivery,
    RecordAdmission, RecordAdmissionAttemptToken, RemoteParticipantInbound, SdkSocketFixture,
    ServerPush, ServerValue, enroll_g1_participants, exchange, expect_applied, expect_push,
};

#[test]
fn sdk_try_receive_local_broker_latency() -> Result<(), Box<dyn Error>> {
    let home = tempfile::tempdir()?;
    let server = SdkSocketFixture::start(&home.path().join("pump-latency"))?;
    let G1Participants {
        sender,
        sender_bound,
        peer_one,
        peer_two,
        ..
    } = enroll_g1_participants(server.address()?)?;
    let mut quiet = Vec::new();
    let mut delivery = Vec::new();
    // Five samples expose the per-turn cadence rather than just cold startup.
    for sample in 0_u8..5 {
        let before = peer_one.pump_diagnostics();
        let started = Instant::now();
        assert!(peer_one.try_receive()?.is_none(), "quiet seat had a frame");
        let elapsed = started.elapsed();
        let after = peer_one.pump_diagnostics();
        assert_eq!(
            after.pumps_run,
            before.pumps_run + 1,
            "a quiet try_receive runs exactly one pump"
        );
        assert_eq!(
            after.reads_timed_out, before.reads_timed_out,
            "a quiet drain ends on its fence, never on a read window closing"
        );
        println!(
            "PUMP_LATENCY quiet sample={sample} elapsed_us={}",
            elapsed.as_micros()
        );
        quiet.push(elapsed);

        let committed = expect_applied(exchange(
            &sender,
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id: G1_CONVERSATION,
                participant_id: sender_bound.participant_id(),
                capability_generation: Generation::ONE,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new(
                    [0xB0 + sample; 16],
                ),
                payload: vec![sample],
            }),
        )?)?;
        let ServerValue::RecordCommitted(committed) = committed else {
            return Err(format!("sample publish did not commit: {committed:?}").into());
        };
        let before = peer_one.pump_diagnostics();
        let started = Instant::now();
        let inbound = peer_one.try_receive()?;
        let elapsed = started.elapsed();
        assert_eq!(
            peer_one.pump_diagnostics().reads_timed_out,
            before.reads_timed_out,
            "an available delivery is returned without a read window closing"
        );
        assert!(
            matches!(
                inbound,
                Some(RemoteParticipantInbound::Push {
                    value: ServerPush::ParticipantDelivery(ParticipantDelivery { delivery_seq, .. }),
                    ..
                }) if delivery_seq == committed.delivery_seq()
            ),
            "expected this seat's exact committed delivery, got {inbound:?}"
        );
        println!(
            "PUMP_LATENCY delivery sample={sample} elapsed_us={}",
            elapsed.as_micros()
        );
        delivery.push(elapsed);
        // Keep the other peer drained; reads are not acknowledgements.
        let _ = expect_push(&peer_two)?;
    }
    quiet.sort();
    delivery.sort();
    println!(
        "PUMP_LATENCY summary quiet_median_us={} quiet_max_us={} delivery_median_us={} delivery_max_us={}",
        quiet[2].as_micros(),
        quiet[4].as_micros(),
        delivery[2].as_micros(),
        delivery[4].as_micros()
    );
    // The latencies above are information. What this pins is a count: no
    // quiet drain and no delivery ended on a read window closing (asserted per
    // sample), which is what the old five-second cadence did on every quiet
    // turn.
    drop((sender, peer_one, peer_two));
    server.stop()?;
    crate::test_fixture::pin_store_opens(module_path!(), "sdk_try_receive_local_broker_latency")?;
    Ok(())
}