liminal-server 0.23.1

//! LIMINAL-SEAT-1 baseline/after harness: a real durable local broker and fresh
//! SDK handles. Client LPCR storage is the existing in-memory fixture (reported
//! explicitly); application filesystem/identity work is not hidden in connect.
//! History means seeded ordinary records. Lifecycle rows added by the ten fresh
//! seats are reported through the observed replay cursor, not silently counted
//! as additional seeded history. No production clock or protocol is changed.

use std::time::Instant;

use liminal_protocol::wire::{LeaveAttemptToken, LeaveRequest};

use super::super::test_participant_config;
use super::{
    AttachAttemptToken, ClientRequest, CredentialAttachRequest, EnrollBound, EnrollmentRequest,
    EnrollmentToken, Error, Generation, ParticipantAck, ParticipantRecord, RecordAdmission,
    RecordAdmissionAttemptToken, RemoteParticipantInbound, SdkParticipant, SdkSocketFixture,
    ServerPush, ServerValue, connect_participant, send_operation,
};

#[path = "e2e_sdk_seat_replay_tests.rs"]
mod replay_tests;

const SAMPLES: usize = 10;
/// The longest seeded history the matrix runs by default. The recorded
/// LIMINAL-SEAT-1 table is the 10,000-record run, asked for explicitly with
/// `LIMINAL_SEAT_MATRIX_HISTORY=10000`; the gate runs the same arms at 100.
const DEFAULT_LONGEST_HISTORY: u64 = 100;
// Observation only, not a production deadline change: the initial release
// baseline proved that 10,000-record Leave exceeds the SDK's 60 s reply bound.
// Ten minutes lets the benchmark record the actual server completion rather
// than lose the remaining samples during fixture teardown. The same budget is
// used before and after. Open timings remain reported for the separately owned
// haematite row; the final assertion follows Waffles's explicit Leave-only scope.
const LEAVE_OBSERVATION_BUDGET: std::time::Duration = std::time::Duration::from_secs(600);

#[derive(Debug, Default)]
struct Pushes {
    ordinary: u64,
    total: u64,
    through: u64,
}

impl Pushes {
    fn absorb(&mut self, value: ServerPush) {
        self.total += 1;
        if let ServerPush::ParticipantDelivery(delivery) = value {
            self.through = self.through.max(delivery.delivery_seq);
            if matches!(delivery.record, ParticipantRecord::OrdinaryRecord { .. }) {
                self.ordinary += 1;
            }
        }
    }
}

fn exchange(
    seat: &SdkParticipant,
    request: ClientRequest,
    pushes: &mut Pushes,
) -> Result<ServerValue, Box<dyn Error>> {
    let observe_long_leave = matches!(&request, ClientRequest::Leave(_));
    let started = Instant::now();
    send_operation(seat, request)?;
    loop {
        let inbound = if observe_long_leave {
            let remaining = LEAVE_OBSERVATION_BUDGET.saturating_sub(started.elapsed());
            seat.receive_within(remaining)?
                .ok_or("benchmark Leave observation budget expired")?
        } else {
            seat.receive()?
        };
        match inbound {
            RemoteParticipantInbound::Applied { value, .. }
            | RemoteParticipantInbound::OutcomeUnknown { value, .. } => return Ok(value),
            RemoteParticipantInbound::Push { value, .. } => pushes.absorb(value),
            other @ RemoteParticipantInbound::Refused { .. } => {
                return Err(format!("seat operation was not applied: {other:?}").into());
            }
        }
    }
}

fn enroll(
    seat: &SdkParticipant,
    conversation: u64,
    token: u8,
) -> Result<EnrollBound, Box<dyn Error>> {
    let value = exchange(
        seat,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: conversation,
            enrollment_token: EnrollmentToken::new([token; 16]),
        }),
        &mut Pushes::default(),
    )?;
    let ServerValue::EnrollBound(bound) = value else {
        return Err(format!("fresh seat enrollment refused: {value:?}").into());
    };
    Ok(bound)
}

#[derive(Debug)]
struct Sample {
    connect: u128,
    enroll: u128,
    attach_reply: u128,
    replay_settle: u128,
    attach: u128,
    leave: u128,
    open: u128,
    ordinary_replayed: u64,
}

fn sample(
    server: &SdkSocketFixture,
    conversation: u64,
    sample_index: u8,
    history: u64,
) -> Result<Sample, Box<dyn Error>> {
    let opening = Instant::now();
    let seat = connect_participant(server.address()?, conversation)?;
    let connect_us = opening.elapsed().as_micros();
    let stage = Instant::now();
    let bound = enroll(&seat, conversation, 0x30 + sample_index)?;
    let enroll_us = stage.elapsed().as_micros();
    let stage = Instant::now();
    let mut pushes = Pushes::default();
    let value = exchange(
        &seat,
        ClientRequest::CredentialAttach(CredentialAttachRequest {
            conversation_id: conversation,
            participant_id: bound.participant_id(),
            capability_generation: bound.capability_generation(),
            attach_secret: bound.attach_secret(),
            attach_attempt_token: AttachAttemptToken::new([0x60 + sample_index; 16]),
            accept_marker_delivery_seq: None,
        }),
        &mut pushes,
    )?;
    let ServerValue::AttachBound(attached) = value else {
        return Err(format!("fresh credential attach refused: {value:?}").into());
    };
    let attach_reply_us = stage.elapsed().as_micros();
    // Measure replay/settling separately from the correlated attach response.
    // The SDK drain's transport fence is event-driven (LIMINAL-PUMP-1), not a
    // sleep. A later-arriving push is still consumed while awaiting Leave.
    while let Some(inbound) = seat.try_receive()? {
        match inbound {
            RemoteParticipantInbound::Push { value, .. } => pushes.absorb(value),
            other => return Err(format!("unexpected replay inbound: {other:?}").into()),
        }
    }
    if pushes.through > attached.persisted_cursor() {
        let value = exchange(
            &seat,
            ClientRequest::ParticipantAck(ParticipantAck {
                conversation_id: conversation,
                participant_id: bound.participant_id(),
                capability_generation: attached.capability_generation(),
                through_seq: pushes.through,
            }),
            &mut pushes,
        )?;
        assert!(
            matches!(
                value,
                ServerValue::AckCommitted(_) | ServerValue::AckNoOp(_)
            ),
            "replay acknowledgement refused: {value:?}"
        );
    }
    let attach_us = stage.elapsed().as_micros();
    let open_us = opening.elapsed().as_micros();
    let ordinary_at_open = pushes.ordinary;
    println!(
        "SEAT_OPEN history={history} sample={sample_index} connect_us={connect_us} enrol_us={enroll_us} attach_reply_us={attach_reply_us} attach_us={attach_us} open_us={open_us} ordinary_at_open={ordinary_at_open}"
    );
    let stage = Instant::now();
    let value = match exchange(
        &seat,
        ClientRequest::Leave(LeaveRequest {
            conversation_id: conversation,
            participant_id: bound.participant_id(),
            capability_generation: attached.capability_generation(),
            attach_secret: attached.attach_secret(),
            leave_attempt_token: LeaveAttemptToken::new([0x90 + sample_index; 16]),
        }),
        &mut pushes,
    ) {
        Ok(value) => value,
        Err(error) => {
            eprintln!(
                "SEAT_LEAVE_FAILED history={history} sample={sample_index} elapsed_us={} error={error}",
                stage.elapsed().as_micros()
            );
            return Err(error);
        }
    };
    let leave_us = stage.elapsed().as_micros();
    assert!(
        matches!(value, ServerValue::LeaveCommitted(_)),
        "leave refused: {value:?}"
    );
    println!(
        "SEAT_SAMPLE history={history} sample={sample_index} connect_us={connect_us} enrol_us={enroll_us} attach_reply_us={attach_reply_us} replay_settle_us={} attach_us={attach_us} leave_us={leave_us} open_us={open_us} ordinary_at_open={ordinary_at_open} ordinary_after_leave={} pushes={} through={} persisted_cursor={} client_store=memory broker_store=disk",
        attach_us.saturating_sub(attach_reply_us),
        pushes.ordinary,
        pushes.total,
        pushes.through,
        attached.persisted_cursor()
    );
    Ok(Sample {
        connect: connect_us,
        enroll: enroll_us,
        attach_reply: attach_reply_us,
        replay_settle: attach_us.saturating_sub(attach_reply_us),
        attach: attach_us,
        leave: leave_us,
        open: open_us,
        ordinary_replayed: pushes.ordinary,
    })
}

fn median(samples: &[Sample], value: impl Fn(&Sample) -> u128) -> u128 {
    let mut values: Vec<_> = samples.iter().map(value).collect();
    values.sort_unstable();
    u128::midpoint(values[SAMPLES / 2 - 1], values[SAMPLES / 2])
}

#[test]
fn sdk_fresh_seat_stage_latency_matrix() -> Result<(), Box<dyn Error>> {
    let longest_history: u64 = match std::env::var("LIMINAL_SEAT_MATRIX_HISTORY") {
        Ok(value) => value.parse()?,
        Err(std::env::VarError::NotPresent) => DEFAULT_LONGEST_HISTORY,
        Err(error) => return Err(error.into()),
    };
    println!(
        "SEAT_OBSERVATION leave_budget_s={} normal_sdk_reply_deadline_s=60 client_store=memory broker_store=disk",
        LEAVE_OBSERVATION_BUDGET.as_secs()
    );
    let home = tempfile::tempdir()?;
    let mut config = test_participant_config();
    // Preserve every seeded ordinary record plus the measured lifecycle rows;
    // fixture capacity must not turn this into a compacted short-stream test.
    let retained = longest_history + 2_000;
    config.max_retained_record_rows = retained;
    config.retained_capacity_entries = retained * 2;
    config.identity_slots = 12;
    let server = SdkSocketFixture::start_with_config(&home.path().join("seat-matrix"), config)?;
    let mut longest = Vec::new();
    for history in [0, longest_history / 10, longest_history] {
        let conversation = 0x53_00 + history;
        let producer = connect_participant(server.address()?, conversation)?;
        let bound = enroll(&producer, conversation, 0x11)?;
        let seeded = Instant::now();
        for record in 0..history {
            let mut token = [0xA1; 16];
            token[..8].copy_from_slice(&record.to_be_bytes());
            let value = exchange(
                &producer,
                ClientRequest::RecordAdmission(RecordAdmission {
                    conversation_id: conversation,
                    participant_id: bound.participant_id(),
                    capability_generation: Generation::ONE,
                    record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
                    payload: record.to_be_bytes().to_vec(),
                }),
                &mut Pushes::default(),
            )?;
            assert!(
                matches!(value, ServerValue::RecordCommitted(_)),
                "seed refused: {value:?}"
            );
            if (record + 1) % 1_000 == 0 {
                println!(
                    "SEAT_SEED history={history} committed={} elapsed_ms={}",
                    record + 1,
                    seeded.elapsed().as_millis()
                );
            }
        }
        println!(
            "SEAT_SEEDED history={history} elapsed_ms={}",
            seeded.elapsed().as_millis()
        );
        let mut samples = Vec::new();
        for index in 0..SAMPLES {
            let index = u8::try_from(index)?;
            samples.push(sample(&server, conversation, index, history)?);
        }
        println!(
            "SEAT_MEDIAN history={history} n={SAMPLES} connect_us={} enrol_us={} attach_reply_us={} replay_settle_us={} attach_us={} leave_us={} open_us={} ordinary_replayed_max={}",
            median(&samples, |s| s.connect),
            median(&samples, |s| s.enroll),
            median(&samples, |s| s.attach_reply),
            median(&samples, |s| s.replay_settle),
            median(&samples, |s| s.attach),
            median(&samples, |s| s.leave),
            median(&samples, |s| s.open),
            samples
                .iter()
                .map(|s| s.ordinary_replayed)
                .max()
                .unwrap_or(0)
        );
        if history == longest_history {
            longest = samples;
        }
        drop(producer);
    }
    server.stop()?;
    // Waffles's 2026-09-06 scope ruling moves the open/append target to the
    // haematite row. Keep reporting it above; SEAT-1 owns the Leave replay cut.
    // Every measured Leave must fit the SDK's unchanged 60-second reply bound,
    // which the original release baseline exceeded. The read-slope oracle
    // separately enforces history-independent durable reads without clock noise.
    assert!(
        longest.iter().all(|s| s.leave < 60_000_000),
        "Leave at {longest_history} records exceeded the unchanged SDK reply deadline: \
         {longest:?}"
    );
    assert!(
        longest.iter().all(|s| s.ordinary_replayed == 0),
        "a fresh default seat replayed history without requesting it: {longest:?}"
    );
    Ok(())
}