liminal-server 0.19.1

//! RED PIN: a killed-then-detached member's ordinary Leave commits.
//!
//! With retention headroom the victim's connection loss commits its Died
//! terminal and the Ordinary binding fate completes at once, so the slot rests
//! Detached. The fate carries the FRONTIER's copy of the victim's cursor to the
//! high watermark (`claim_frontier/binding_fate_transition.rs`, the
//! non-finalizer arm) and leaves the member's acknowledged boundary where it
//! was. That split is deliberate (`docs/design/DETACHED-LEAVE-GUARD-2026-10-07.md`).
//! The settled Leave's identity check (`claim_frontier.rs`
//! `validate_leave_identity`) demanded the two be equal, refused with
//! `Identity`, and `ops_leave.rs`'s key-planning `map_err` turned that into an
//! invariant that discarded the conversation.
//!
//! The guard now accepts a Detached member's frontier cursor at or above its
//! member cursor. The Leave commits:
//!
//! - live: on the handler that ran the connection-fate transaction;
//! - replay: the aggregate reference replay of the durable log restores the
//!   live owner's state exactly, split included, and the Leave then commits;
//! - boot: a fresh handler on the same store rebuilds the same state through
//!   its boot replay, and the Leave commits there.
//!
//! Everything goes through the production seam on the reachability fixture.

use std::error::Error;
use std::sync::Arc;

use liminal_protocol::lifecycle::BindingState;
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, LeaveAttemptToken, LeaveRequest, ServerValue,
};

use crate::server::participant::ParticipantConnectionConversations;

use super::ProductionParticipantHandler;
use super::log::OperationLog;
use super::state::ConversationAuthority;
use super::tests::{dispatch_tracked, test_participant_config};
use super::tests_pending_died_live_reachability::{
    CONVERSATION, LiveResidence, build_residence_with_retention, live_binding_state,
};

/// Retention with room for every row the fixture and its Leave produce, so
/// the victim's terminal commits and its fate completes in the same
/// transaction.
const HEADROOM_RETAINED_ROWS: u64 = 16;

/// The victim's exact-secret Leave under its current generation.
const fn victim_leave(residence: &LiveResidence, token_byte: u8) -> LeaveRequest {
    LeaveRequest {
        conversation_id: CONVERSATION,
        participant_id: residence.victim_id,
        capability_generation: residence.victim_epoch.capability_generation,
        attach_secret: residence.victim_attach_secret,
        leave_attempt_token: LeaveAttemptToken::new([token_byte; 16]),
    }
}

fn send_leave(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    request: LeaveRequest,
) -> Result<ServerValue, Box<dyn Error>> {
    dispatch_tracked(
        handler,
        connection,
        &mut ParticipantConnectionConversations::default(),
        ClientRequest::Leave(request),
    )
}

/// Asserts the Leave committed, rendering whatever the seam produced instead.
fn assert_leave_committed(outcome: &Result<ServerValue, Box<dyn Error>>, context: &str) {
    let rendered = match outcome {
        Ok(answer) => format!("answered {answer:?}"),
        Err(error) => error.to_string(),
    };
    assert!(
        matches!(outcome, Ok(ServerValue::LeaveCommitted(_))),
        "{context}: the detached victim's ordinary Leave did not commit: {rendered}"
    );
}

/// The killed victim with retention headroom, checked to rest Detached.
fn killed_detached_residence() -> Result<LiveResidence, Box<dyn Error>> {
    let residence = build_residence_with_retention(true, HEADROOM_RETAINED_ROWS)?;
    let binding = live_binding_state(&residence)?;
    if !matches!(binding, Some(BindingState::Detached)) {
        return Err(
            format!("the killed headroom fixture did not rest Detached: {binding:?}").into(),
        );
    }
    Ok(residence)
}

/// The victim's binding, member cursor and frontier cursor, and the owner's
/// log head, rendered for comparison between a live and a rebuilt owner.
fn victim_state(
    authority: &ConversationAuthority,
    participant_id: u64,
) -> Result<String, Box<dyn Error>> {
    let slot = authority
        .slots
        .get(&participant_id)
        .ok_or("the victim's slot is absent")?;
    let frontier = authority
        .frontier()
        .ok_or("the frontier owner is taken")?
        .frontiers()
        .active_identities()
        .participants()
        .iter()
        .find(|participant| participant.participant_index() == participant_id)
        .ok_or("the victim is absent from the frontier")?
        .cursor();
    Ok(format!(
        "binding {:?}, member cursor {}, frontier cursor {frontier}, log head {}",
        slot.binding,
        slot.member.cursor(),
        authority.next_log_sequence
    ))
}

/// The victim's state read out of the LIVE in-memory authority.
fn live_victim_state(residence: &LiveResidence) -> Result<String, Box<dyn Error>> {
    let cell = residence.handler.cell(CONVERSATION)?;
    let owner = cell
        .lock()
        .map_err(|_| "detached-leave conversation lock was poisoned")?;
    let authority = owner
        .as_ref()
        .ok_or("detached-leave conversation owner was discarded")?;
    let state = victim_state(authority, residence.victim_id);
    drop(owner);
    state
}

/// THE RED PIN. The killed-then-detached victim's ordinary Leave, sent from a
/// new connection, commits.
#[test]
fn a_killed_then_detached_victims_ordinary_leave_commits() -> Result<(), Box<dyn Error>> {
    let residence = killed_detached_residence()?;
    let outcome = send_leave(
        &residence.handler,
        ConnectionIncarnation::new(511, 22),
        victim_leave(&residence, 0xF5),
    );
    assert_leave_committed(&outcome, "live");
    Ok(())
}

/// REPLAY OF A STORED FATE. Replaying the durable log, stored Ordinary fate
/// row included, rebuilds exactly the live owner's state, and the Leave then
/// commits.
#[test]
fn replaying_the_stored_ordinary_fate_restores_the_live_state_and_the_leave_commits()
-> Result<(), Box<dyn Error>> {
    let residence = killed_detached_residence()?;
    let live = live_victim_state(&residence)?;
    let log = OperationLog::new(Arc::clone(&residence.store), CONVERSATION);
    let replayed = residence
        .handler
        .replay_aggregate_reference(CONVERSATION, &log)?;
    assert_eq!(
        victim_state(&replayed, residence.victim_id)?,
        live,
        "replay of the stored fate disagrees with the live owner"
    );
    let outcome = send_leave(
        &residence.handler,
        ConnectionIncarnation::new(511, 23),
        victim_leave(&residence, 0xF6),
    );
    assert_leave_committed(&outcome, "after replay");
    Ok(())
}

/// BOOT. A fresh handler on the same store rebuilds the live state through its
/// boot replay, and the victim's ordinary Leave commits there.
#[test]
fn after_a_reboot_the_detached_victims_ordinary_leave_commits() -> Result<(), Box<dyn Error>> {
    let residence = killed_detached_residence()?;
    let live = live_victim_state(&residence)?;
    let victim_id = residence.victim_id;
    let request = victim_leave(&residence, 0xF7);
    let LiveResidence { store, handler, .. } = residence;
    drop(handler);

    let mut config = test_participant_config();
    config.max_retained_record_rows = HEADROOM_RETAINED_ROWS;
    let rebooted = ProductionParticipantHandler::new(Arc::clone(&store), config)?;
    let log = OperationLog::new(Arc::clone(&store), CONVERSATION);
    let booted = rebooted.replay_and_repair(CONVERSATION, &log)?;
    assert_eq!(
        victim_state(&booted, victim_id)?,
        live,
        "the boot replay disagrees with the live owner"
    );
    drop(booted);
    let outcome = send_leave(&rebooted, ConnectionIncarnation::new(512, 1), request);
    assert_leave_committed(&outcome, "after a reboot");
    Ok(())
}