liminal-server 0.20.0

//! ACCESS-005: Lys grant revocation applied to admission and live rights,
//! through the public `RevocationAuthority`. The Lys producer and decision
//! seams are fixtures here, standing in for DIRECTORY-089 R1/R2, which Lys
//! has not published; every race is held by the authority's own owned phase
//! (stage, then commit) and by the order events are applied, never by a
//! sleep or a timer.

use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock, Weak};

use liminal_server::auth_lys::HotAction;
use liminal_server::auth_revocation::{
    Applied, AuthenticatedBinding, BoundRightClaim, CloseWrite, DecisionReply, Digest32, GapLength,
    GrantBindingSource, GrantDecisionSource, GrantFeedChange, GrantFeedEvent, GrantId, GrantLogId,
    MAX_DIAGNOSTIC_EVENTS, Readiness, Revision, RevocationAuthority, RevocationDiagnostics,
    RevocationEffect, RevocationPolicy, RevocationRefusal, RightAction, UnpublishedGrantDecisions,
    verify_binding,
};

type TestResult = Result<(), Box<dyn std::error::Error>>;
type Failure = Box<dyn std::error::Error>;

/// The authority's diagnostics snapshot; a poisoned authority is a failed test.
fn snapshot(authority: &RevocationAuthority) -> Result<RevocationDiagnostics, Failure> {
    Ok(authority.diagnostics()?)
}

const ISSUER: Digest32 = Digest32([7; 32]);
const AUDIENCE: &str = "liminal";
const MARKER_HOLDER: &[u8] = b"holder-marker-7f3a";

fn log_l() -> Result<GrantLogId, Failure> {
    Ok(GrantLogId::new("grants-L", 0)?)
}

fn log_m() -> Result<GrantLogId, Failure> {
    Ok(GrantLogId::new("grants-M", 1)?)
}

fn grant(id: &str) -> Result<GrantId, Failure> {
    Ok(GrantId::new(id)?)
}

fn digest(credential: &[u8]) -> Digest32 {
    Digest32(*blake3::hash(credential).as_bytes())
}

fn evidence(revision: u64) -> Digest32 {
    let mut bytes = [0; 32];
    for (slot, byte) in bytes.iter_mut().zip(revision.to_be_bytes()) {
        *slot = byte;
    }
    Digest32(bytes)
}

fn policy() -> Result<RevocationPolicy, Failure> {
    Ok(RevocationPolicy {
        log: log_l()?,
        audience: AUDIENCE.to_owned(),
        issuer: ISSUER,
    })
}

/// The producer fixture: what Lys would have bound to each credential.
#[derive(Debug, Default)]
struct Producer {
    bindings: Mutex<BTreeMap<Vec<u8>, AuthenticatedBinding>>,
}

impl Producer {
    fn produce(&self, credential: &[u8], binding: AuthenticatedBinding) -> TestResult {
        self.bindings
            .lock()
            .map_err(|_| "producer poisoned")?
            .insert(credential.to_vec(), binding);
        Ok(())
    }
}

impl GrantBindingSource for Producer {
    fn pass_digest(&self, credential: &[u8]) -> Result<Digest32, RevocationRefusal> {
        Ok(digest(credential))
    }

    fn binding_for(&self, credential: &[u8]) -> Result<AuthenticatedBinding, RevocationRefusal> {
        self.bindings
            .lock()
            .map_err(|_| RevocationRefusal::BindingMissing)?
            .get(credential)
            .cloned()
            .ok_or(RevocationRefusal::BindingMissing)
    }
}

/// The decision fixture: answers each cold ask at a set revision, counts the
/// asks, and can run one event against the authority while the ask is out
/// (the reply then arrives after that event, as a delayed reply would).
#[derive(Default)]
struct Decisions {
    asks: AtomicU64,
    answer_at: Mutex<Option<u64>>,
    during_ask: Mutex<Option<GrantFeedEvent>>,
    authority: OnceLock<Weak<RevocationAuthority>>,
}

impl std::fmt::Debug for Decisions {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str("Decisions")
    }
}

impl GrantDecisionSource for Decisions {
    fn decide(
        &self,
        log: &GrantLogId,
        grants: &[GrantId],
        at_least: Revision,
    ) -> Result<DecisionReply, RevocationRefusal> {
        self.asks.fetch_add(1, Ordering::SeqCst);
        let during = self
            .during_ask
            .lock()
            .map_err(|_| RevocationRefusal::RevisionUnavailable)?
            .take();
        if let Some(event) = during
            && let Some(authority) = self.authority.get().and_then(Weak::upgrade)
        {
            let _applied = authority.apply(&event);
        }
        let revision = self
            .answer_at
            .lock()
            .map_err(|_| RevocationRefusal::RevisionUnavailable)?
            .map_or(at_least, Revision::new);
        Ok(DecisionReply {
            log: log.clone(),
            revision,
            decisions: grants.iter().map(|grant| (grant.clone(), false)).collect(),
        })
    }
}

/// A right claim on `chains` of grant ids.
fn right(
    action: RightAction,
    prefix: &str,
    chains: &[&[&str]],
) -> Result<BoundRightClaim, Failure> {
    let chains = chains
        .iter()
        .map(|chain| {
            chain
                .iter()
                .map(|id| grant(id))
                .collect::<Result<Vec<_>, _>>()
        })
        .collect::<Result<Vec<_>, _>>()?;
    Ok(BoundRightClaim {
        action,
        prefix: prefix.to_owned(),
        chains,
    })
}

fn binding_for(
    credential: &[u8],
    required: u64,
    rights: Vec<BoundRightClaim>,
) -> Result<AuthenticatedBinding, Failure> {
    Ok(AuthenticatedBinding {
        issuer: ISSUER,
        audience: AUDIENCE.to_owned(),
        holder: MARKER_HOLDER.to_vec(),
        pass_digest: digest(credential),
        log: log_l()?,
        required: Revision::new(required),
        rights,
    })
}

/// The standard rights: Connect on chain `conn`, publish under `orders.` on
/// `p`, publish under `billing.` on `q`.
fn standard_rights() -> Result<Vec<BoundRightClaim>, Failure> {
    Ok(vec![
        right(RightAction::Connect, "", &[&["conn"]])?,
        right(RightAction::Hot(HotAction::Publish), "orders.", &[&["p"]])?,
        right(RightAction::Hot(HotAction::Publish), "billing.", &[&["q"]])?,
    ])
}

struct Harness {
    authority: Arc<RevocationAuthority>,
    producer: Arc<Producer>,
    decisions: Arc<Decisions>,
}

fn harness() -> Result<Harness, Failure> {
    let producer = Arc::new(Producer::default());
    let decisions = Arc::new(Decisions::default());
    let authority = Arc::new(RevocationAuthority::new(
        policy()?,
        Arc::clone(&producer) as Arc<dyn GrantBindingSource>,
        Arc::clone(&decisions) as Arc<dyn GrantDecisionSource>,
    ));
    decisions
        .authority
        .set(Arc::downgrade(&authority))
        .map_err(|_| "authority already set")?;
    Ok(Harness {
        authority,
        producer,
        decisions,
    })
}

fn baseline(through: u64, grants: &[&str]) -> Result<GrantFeedEvent, Failure> {
    Ok(GrantFeedEvent::Baseline {
        log: log_l()?,
        through: Revision::new(through),
        decisions: grants
            .iter()
            .map(|id| grant(id).map(|grant| (grant, false)))
            .collect::<Result<_, _>>()?,
    })
}

fn ready(through: u64) -> Result<GrantFeedEvent, Failure> {
    Ok(GrantFeedEvent::Ready {
        log: log_l()?,
        through: Revision::new(through),
    })
}

fn change(revision: u64, change: GrantFeedChange) -> Result<GrantFeedEvent, Failure> {
    Ok(GrantFeedEvent::Change {
        log: log_l()?,
        revision: Revision::new(revision),
        change,
        evidence: evidence(revision),
    })
}

fn revoke(revision: u64, id: &str) -> Result<GrantFeedEvent, Failure> {
    change(revision, GrantFeedChange::Revoke(grant(id)?))
}

/// Baseline at `through` over `grants`, then `Ready`.
fn settle(harness: &Harness, through: u64, grants: &[&str]) -> TestResult {
    assert_eq!(
        harness.authority.apply(&baseline(through, grants)?).applied,
        Applied::Baseline
    );
    assert_eq!(
        harness.authority.apply(&ready(through)?).applied,
        Applied::Ready
    );
    Ok(())
}

/// A credential for connection `n`, bound with the standard rights at
/// `required`.
fn credential(harness: &Harness, n: u64, required: u64) -> Result<Vec<u8>, Failure> {
    let credential = format!("credential-{n}").into_bytes();
    harness.producer.produce(
        &credential,
        binding_for(&credential, required, standard_rights()?)?,
    )?;
    Ok(credential)
}

// R1 ----------------------------------------------------------------------

#[test]
fn r1_the_cache_version_is_the_receipt_index_plus_one() {
    assert_eq!(Revision::from_receipt_index(40), Some(Revision::new(41)));
    assert_eq!(Revision::from_receipt_index(u64::MAX), None);
}

#[test]
fn r1_a_decision_from_another_log_cannot_satisfy_the_binding() -> TestResult {
    let harness = harness()?;
    let wrong = GrantFeedEvent::Baseline {
        log: log_m()?,
        through: Revision::new(999),
        decisions: vec![(grant("conn")?, false)],
    };
    assert!(matches!(
        harness.authority.apply(&wrong).applied,
        Applied::Refused(_)
    ));
    let wrong_ready = GrantFeedEvent::Ready {
        log: log_m()?,
        through: Revision::new(999),
    };
    assert!(matches!(
        harness.authority.apply(&wrong_ready).applied,
        Applied::Refused(_)
    ));
    let pass = credential(&harness, 1, 41)?;
    assert_eq!(
        harness.authority.admit(1, &pass, MARKER_HOLDER).err(),
        Some(RevocationRefusal::StreamUnavailable)
    );
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.retained_decisions, 0);
    assert_eq!(diagnostics.counts.admissions_committed, 0);
    assert_eq!(diagnostics.counts.refused_events, 2);
    Ok(())
}

#[test]
fn r1_a_substituted_binding_is_refused_field_by_field() -> TestResult {
    let harness = harness()?;
    settle(&harness, 41, &["conn", "p", "q"])?;
    let pass = b"presented-pass".to_vec();
    let honest = binding_for(&pass, 41, standard_rights()?)?;
    let mut substitutes = Vec::new();
    let mut issuer = honest.clone();
    issuer.issuer = Digest32([8; 32]);
    substitutes.push(issuer);
    let mut audience = honest.clone();
    audience.audience = "another-product".to_owned();
    substitutes.push(audience);
    let mut holder = honest.clone();
    holder.holder = b"someone-else".to_vec();
    substitutes.push(holder);
    let mut moved = honest.clone();
    moved.pass_digest = digest(b"another-pass");
    substitutes.push(moved);
    for substitute in substitutes {
        harness.producer.produce(&pass, substitute)?;
        assert_eq!(
            harness.authority.admit(1, &pass, MARKER_HOLDER).err(),
            Some(RevocationRefusal::BindingMissing)
        );
    }
    harness.producer.produce(&pass, honest)?;
    assert!(harness.authority.admit(1, &pass, MARKER_HOLDER).is_ok());
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.counts.admissions_committed, 1);
    assert_eq!(diagnostics.counts.admissions_refused, 4);
    Ok(())
}

#[test]
fn r1_without_a_published_producer_every_admission_names_the_missing_binding() -> TestResult {
    let authority = RevocationAuthority::unpublished(policy()?);
    assert_eq!(
        authority.admit(1, b"any-pass", MARKER_HOLDER).err(),
        Some(RevocationRefusal::BindingMissing)
    );
    assert_eq!(snapshot(&authority)?.counts.admissions_committed, 0);
    Ok(())
}

#[test]
fn r1_a_binding_without_an_admitting_right_or_over_its_bounds_is_refused() -> TestResult {
    let pass = b"p".to_vec();
    let producer = Producer::default();
    let no_connect = binding_for(
        &pass,
        1,
        vec![right(
            RightAction::Hot(HotAction::Publish),
            "orders.",
            &[&["p"]],
        )?],
    )?;
    producer.produce(&pass, no_connect)?;
    assert_eq!(
        verify_binding(&policy()?, &producer, &pass, MARKER_HOLDER).err(),
        Some(RevocationRefusal::BindingMissing)
    );
    let long_chain: Vec<&str> = std::iter::repeat_n("g", 17).collect();
    let over = binding_for(
        &pass,
        1,
        vec![right(RightAction::Connect, "", &[long_chain.as_slice()])?],
    )?;
    producer.produce(&pass, over)?;
    assert_eq!(
        verify_binding(&policy()?, &producer, &pass, MARKER_HOLDER).err(),
        Some(RevocationRefusal::BindingMissing)
    );
    Ok(())
}

#[test]
fn r1_an_independent_chain_keeps_the_admitting_right() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["a", "b"])?;
    let pass = b"two-chains".to_vec();
    harness.producer.produce(
        &pass,
        binding_for(
            &pass,
            40,
            vec![right(RightAction::Connect, "", &[&["a"], &["b"]])?],
        )?,
    )?;
    let grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    let first = harness.authority.apply(&revoke(41, "a")?);
    assert!(first.effects.is_empty());
    assert_eq!(grants.close_pending(), None);
    let second = harness.authority.apply(&revoke(42, "b")?);
    assert_eq!(
        second.effects,
        vec![RevocationEffect::Close {
            pid: 1,
            reason: RevocationRefusal::GrantRevoked
        }]
    );
    assert_eq!(
        grants.close_pending(),
        Some(RevocationRefusal::GrantRevoked)
    );
    Ok(())
}

#[test]
fn r1_an_unrelated_revocation_changes_no_cache_entry() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let _grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    let before = snapshot(&harness.authority)?;
    let outcome = harness.authority.apply(&revoke(41, "unrelated")?);
    assert!(outcome.effects.is_empty());
    let after = snapshot(&harness.authority)?;
    assert_eq!(after.retained_decisions, before.retained_decisions);
    assert_eq!(after.counts.affected_visits, 0);
    Ok(())
}

#[test]
fn r1_debug_and_errors_carry_no_holder_prefix_or_grant_bytes() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    let binding = binding_for(&pass, 40, standard_rights()?)?;
    let shown = [
        format!("{binding:?}"),
        format!("{grants:?}"),
        format!("{:?}", snapshot(&harness.authority)?),
        format!("{:?}", grant("conn")?),
        format!("{}", RevocationRefusal::GrantRevoked),
    ];
    for text in shown {
        assert!(!text.contains("holder-marker"), "{text}");
        assert!(!text.contains("orders."), "{text}");
        assert!(!text.contains("billing."), "{text}");
        assert!(!text.contains("\"conn\""), "{text}");
        assert!(!text.contains("credential-1"), "{text}");
    }
    Ok(())
}

// R2 ----------------------------------------------------------------------

#[test]
fn r2_nothing_admits_before_ready_and_three_admit_after() -> TestResult {
    let harness = harness()?;
    let passes: Vec<Vec<u8>> = (1..=3)
        .map(|n| credential(&harness, n, 40))
        .collect::<Result<_, _>>()?;
    for (pid, pass) in (1_u64..).zip(&passes) {
        assert_eq!(
            harness.authority.admit(pid, pass, MARKER_HOLDER).err(),
            Some(RevocationRefusal::StreamUnavailable)
        );
    }
    assert_eq!(
        harness
            .authority
            .apply(&baseline(40, &["conn", "p", "q"])?)
            .applied,
        Applied::Baseline
    );
    assert_eq!(
        harness
            .authority
            .admit(1, passes.first().ok_or("pass")?, MARKER_HOLDER)
            .err(),
        Some(RevocationRefusal::StreamUnavailable)
    );
    assert_eq!(harness.authority.apply(&ready(40)?).applied, Applied::Ready);
    for (pid, pass) in (1_u64..).zip(&passes) {
        assert!(harness.authority.admit(pid, pass, MARKER_HOLDER).is_ok());
    }
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.counts.admissions_committed, 3);
    assert_eq!(diagnostics.bound_connections, 3);
    Ok(())
}

#[test]
fn r2_a_gap_refuses_new_admissions_keeps_existing_and_resume_rechecks_all() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let mut held = Vec::new();
    for pid in 1..=3 {
        let pass = credential(&harness, pid, 40)?;
        held.push(harness.authority.admit(pid, &pass, MARKER_HOLDER)?);
    }
    let gap = harness
        .authority
        .apply(&change(42, GrantFeedChange::Issue(grant("x")?))?);
    assert_eq!(gap.applied, Applied::Gap);
    assert!(gap.effects.is_empty());
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.readiness, Readiness::Gap);
    assert_eq!(diagnostics.missing, GapLength::Known(1));
    assert_eq!(diagnostics.counts.gap_transitions, 1);
    let late = credential(&harness, 4, 40)?;
    assert_eq!(
        harness.authority.admit(4, &late, MARKER_HOLDER).err(),
        Some(RevocationRefusal::StreamGap)
    );
    assert!(held.iter().all(|grants| grants.close_pending().is_none()));
    // The resumed contiguous tail, then Ready: every open connection is
    // rechecked once before Ready is published.
    assert_eq!(
        harness
            .authority
            .apply(&change(41, GrantFeedChange::Issue(grant("w")?))?)
            .applied,
        Applied::Change
    );
    assert_eq!(
        harness
            .authority
            .apply(&change(42, GrantFeedChange::Issue(grant("x")?))?)
            .applied,
        Applied::Change
    );
    assert_eq!(snapshot(&harness.authority)?.counts.resume_rechecks, 0);
    assert_eq!(harness.authority.apply(&ready(42)?).applied, Applied::Ready);
    let settled = snapshot(&harness.authority)?;
    assert_eq!(settled.counts.resume_rechecks, 3);
    assert_eq!(settled.readiness, Readiness::Ready);
    assert_eq!(settled.settled_generation, settled.resume_generation);
    assert_eq!(settled.counts.gap_transitions, 1);
    Ok(())
}

#[test]
fn r2_a_duplicate_changes_nothing_and_a_conflict_is_terminal() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let _grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    assert_eq!(
        harness.authority.apply(&revoke(41, "p")?).applied,
        Applied::Change
    );
    let before = snapshot(&harness.authority)?;
    let duplicate = harness.authority.apply(&revoke(41, "p")?);
    assert_eq!(duplicate.applied, Applied::Duplicate);
    assert!(duplicate.effects.is_empty());
    let after = snapshot(&harness.authority)?;
    assert_eq!(after.retained_decisions, before.retained_decisions);
    assert_eq!(after.counts.withdrawals, before.counts.withdrawals);
    let conflicting = GrantFeedEvent::Change {
        log: log_l()?,
        revision: Revision::new(41),
        change: GrantFeedChange::Revoke(grant("q")?),
        evidence: evidence(999),
    };
    assert_eq!(
        harness.authority.apply(&conflicting).applied,
        Applied::Terminal("conflicting_evidence")
    );
    assert_eq!(snapshot(&harness.authority)?.readiness, Readiness::Terminal);
    Ok(())
}

#[test]
fn r2_a_lost_transport_refuses_new_admissions_and_keeps_denials() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    let _withdrawn = harness.authority.apply(&revoke(41, "p")?);
    assert_eq!(
        harness.authority.apply(&GrantFeedEvent::Lost).applied,
        Applied::Lost
    );
    let other = credential(&harness, 2, 40)?;
    assert_eq!(
        harness.authority.admit(2, &other, MARKER_HOLDER).err(),
        Some(RevocationRefusal::StreamUnavailable)
    );
    assert_eq!(
        grants.allows(HotAction::Publish, "orders.eu").err(),
        Some(RevocationRefusal::GrantRevoked)
    );
    assert!(grants.allows(HotAction::Publish, "billing.eu").is_ok());
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.missing, GapLength::Unknown);
    assert_eq!(diagnostics.counts.loss_transitions, 1);
    Ok(())
}

#[test]
fn r2_a_log_reset_refuses_until_the_new_baseline_and_closes_old_bindings() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    let reset = GrantFeedEvent::Reset {
        from: log_l()?,
        to: log_m()?,
    };
    assert_eq!(harness.authority.apply(&reset).applied, Applied::Reset);
    let other = credential(&harness, 2, 40)?;
    assert_eq!(
        harness.authority.admit(2, &other, MARKER_HOLDER).err(),
        Some(RevocationRefusal::LogReset)
    );
    // Revisions from L never satisfy M: a change on L is refused.
    assert!(matches!(
        harness.authority.apply(&revoke(41, "p")?).applied,
        Applied::Refused(_)
    ));
    let m_baseline = GrantFeedEvent::Baseline {
        log: log_m()?,
        through: Revision::new(5),
        decisions: Vec::new(),
    };
    assert_eq!(
        harness.authority.apply(&m_baseline).applied,
        Applied::Baseline
    );
    let m_ready = GrantFeedEvent::Ready {
        log: log_m()?,
        through: Revision::new(5),
    };
    let settled = harness.authority.apply(&m_ready);
    assert_eq!(
        settled.effects,
        vec![RevocationEffect::Close {
            pid: 1,
            reason: RevocationRefusal::LogReset
        }]
    );
    assert_eq!(grants.close_pending(), Some(RevocationRefusal::LogReset));
    Ok(())
}

#[test]
fn r2_an_unknown_event_is_a_named_terminal_error_granting_nothing() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let outcome = harness.authority.apply(&GrantFeedEvent::Invalid {
        reason: "unknown_event_version",
    });
    assert_eq!(outcome.applied, Applied::Terminal("unknown_event_version"));
    let pass = credential(&harness, 1, 40)?;
    assert_eq!(
        harness.authority.admit(1, &pass, MARKER_HOLDER).err(),
        Some(RevocationRefusal::StreamUnavailable)
    );
    assert!(matches!(
        harness.authority.apply(&ready(40)?).applied,
        Applied::Refused(_)
    ));
    assert_eq!(snapshot(&harness.authority)?.counts.terminal_errors, 1);
    Ok(())
}

#[test]
fn r2_a_trusted_watermark_closes_the_interval_without_a_false_gap() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let watermark = GrantFeedEvent::Watermark {
        log: log_l()?,
        through: Revision::new(41),
    };
    assert_eq!(
        harness.authority.apply(&watermark).applied,
        Applied::Watermark
    );
    assert_eq!(
        harness.authority.apply(&revoke(42, "q")?).applied,
        Applied::Change
    );
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.applied, Some(Revision::new(42)));
    assert_eq!(diagnostics.counts.gap_transitions, 0);
    assert_eq!(diagnostics.readiness, Readiness::Ready);
    Ok(())
}

// R3 ----------------------------------------------------------------------

#[test]
fn r3_a_revocation_between_staging_and_commit_wins() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let staged = harness.authority.stage(&pass, MARKER_HOLDER)?;
    assert_eq!(
        harness.authority.apply(&revoke(41, "conn")?).applied,
        Applied::Change
    );
    assert_eq!(
        harness.authority.commit(1, staged).err(),
        Some(RevocationRefusal::GrantRevoked)
    );
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.bound_connections, 0);
    assert_eq!(diagnostics.index_entries, 0);
    assert_eq!(diagnostics.inflight_asks, 0);
    assert_eq!(diagnostics.counts.admissions_committed, 0);
    // Reconnecting with the same revoked chain is refused too.
    assert_eq!(
        harness.authority.admit(1, &pass, MARKER_HOLDER).err(),
        Some(RevocationRefusal::GrantRevoked)
    );
    Ok(())
}

#[test]
fn r3_a_commit_before_the_revocation_is_closed_at_the_later_revision() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let staged = harness.authority.stage(&pass, MARKER_HOLDER)?;
    let grants = harness.authority.commit(1, staged)?;
    let outcome = harness.authority.apply(&revoke(41, "conn")?);
    assert_eq!(
        outcome.effects,
        vec![RevocationEffect::Close {
            pid: 1,
            reason: RevocationRefusal::GrantRevoked
        }]
    );
    assert_eq!(
        grants.close_pending(),
        Some(RevocationRefusal::GrantRevoked)
    );
    assert!(harness.authority.release(1));
    assert_eq!(snapshot(&harness.authority)?.index_entries, 0);
    Ok(())
}

#[test]
fn r3_a_delayed_allowing_reply_never_overwrites_a_later_deny() -> TestResult {
    let harness = harness()?;
    // The baseline does not carry `p`, so admission asks for it cold.
    settle(&harness, 40, &["conn", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    *harness
        .decisions
        .during_ask
        .lock()
        .map_err(|_| "poisoned")? = Some(revoke(41, "p")?);
    *harness.decisions.answer_at.lock().map_err(|_| "poisoned")? = Some(40);
    let grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    assert_eq!(harness.decisions.asks.load(Ordering::SeqCst), 1);
    assert_eq!(
        grants.allows(HotAction::Publish, "orders.eu").err(),
        Some(RevocationRefusal::GrantRevoked)
    );
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.counts.stale_overwrites_prevented, 1);
    assert_eq!(diagnostics.applied, Some(Revision::new(41)));
    Ok(())
}

#[test]
fn r3_a_reply_older_than_its_ask_is_refused() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    *harness.decisions.answer_at.lock().map_err(|_| "poisoned")? = Some(39);
    assert_eq!(
        harness.authority.admit(1, &pass, MARKER_HOLDER).err(),
        Some(RevocationRefusal::RevisionUnavailable)
    );
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.counts.stale_replies_refused, 1);
    assert_eq!(diagnostics.inflight_asks, 0);
    Ok(())
}

#[test]
fn r3_a_required_revision_ahead_of_the_cache_is_refused() -> TestResult {
    let harness = harness()?;
    settle(&harness, 41, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 42)?;
    assert_eq!(
        harness.authority.admit(1, &pass, MARKER_HOLDER).err(),
        Some(RevocationRefusal::RevisionUnavailable)
    );
    Ok(())
}

#[test]
fn r3_a_second_binding_never_replaces_the_first() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let first = credential(&harness, 1, 40)?;
    let second = credential(&harness, 2, 40)?;
    let grants = harness.authority.admit(1, &first, MARKER_HOLDER)?;
    assert_eq!(
        harness.authority.admit(1, &second, MARKER_HOLDER).err(),
        Some(RevocationRefusal::BindingMissing)
    );
    assert_eq!(grants.binding().required(), Revision::new(40));
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.counts.binding_replacements_refused, 1);
    assert_eq!(diagnostics.bound_connections, 1);
    Ok(())
}

// R4 ----------------------------------------------------------------------

#[test]
fn r4_losing_one_right_withdraws_only_that_right() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    let outcome = harness.authority.apply(&revoke(41, "p")?);
    assert_eq!(
        outcome.effects,
        vec![RevocationEffect::Withdraw { pid: 1, right: 1 }]
    );
    assert_eq!(
        grants.allows(HotAction::Publish, "orders.eu").err(),
        Some(RevocationRefusal::GrantRevoked)
    );
    assert!(grants.allows(HotAction::Publish, "billing.eu").is_ok());
    assert_eq!(grants.close_pending(), None);
    assert_eq!(
        grants.allows(HotAction::Subscribe, "orders.eu").err(),
        Some(RevocationRefusal::BindingMissing)
    );
    Ok(())
}

#[test]
fn r4_a_change_visits_only_the_bound_connections() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q", "shared"])?;
    let mut held = Vec::new();
    for pid in 1..=100_u64 {
        let pass = format!("credential-{pid}").into_bytes();
        let connect = if pid <= 5 { "shared" } else { "conn" };
        harness.producer.produce(
            &pass,
            binding_for(
                &pass,
                40,
                vec![
                    right(RightAction::Connect, "", &[&[connect]])?,
                    right(RightAction::Hot(HotAction::Publish), "orders.", &[&["p"]])?,
                ],
            )?,
        )?;
        held.push(harness.authority.admit(pid, &pass, MARKER_HOLDER)?);
    }
    let outcome = harness.authority.apply(&revoke(41, "shared")?);
    assert_eq!(outcome.effects.len(), 5);
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.counts.affected_visits, 5);
    assert_eq!(diagnostics.counts.closes_decided, 5);
    let duplicate = harness.authority.apply(&revoke(41, "shared")?);
    assert!(duplicate.effects.is_empty());
    assert_eq!(snapshot(&harness.authority)?.counts.affected_visits, 5);
    // A resume rechecks every remaining open connection exactly once.
    assert_eq!(
        harness.authority.apply(&GrantFeedEvent::Lost).applied,
        Applied::Lost
    );
    assert_eq!(harness.authority.apply(&ready(41)?).applied, Applied::Ready);
    assert_eq!(snapshot(&harness.authority)?.counts.resume_rechecks, 100);
    let closed = held
        .iter()
        .filter(|grants| grants.close_pending().is_some())
        .count();
    assert_eq!(closed, 5);
    Ok(())
}

#[test]
fn r4_hot_decisions_ask_nothing_and_read_no_rows() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    let before = snapshot(&harness.authority)?;
    for _ in 0..10_000 {
        grants.allows(HotAction::Publish, "orders.eu")?;
    }
    let after = snapshot(&harness.authority)?;
    assert_eq!(after.counts.decision_asks, before.counts.decision_asks);
    assert_eq!(after.counts.rows_read, before.counts.rows_read);
    assert_eq!(after.retained_decisions, before.retained_decisions);
    assert_eq!(after.hot_allowed, before.hot_allowed.saturating_add(10_000));
    assert_eq!(harness.decisions.asks.load(Ordering::SeqCst), 0);
    Ok(())
}

// R6 ----------------------------------------------------------------------

#[test]
fn r6_release_is_idempotent_and_retires_every_index_entry() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let _grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    assert_eq!(snapshot(&harness.authority)?.index_entries, 3);
    assert!(harness.authority.release(1));
    assert!(!harness.authority.release(1));
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.index_entries, 0);
    assert_eq!(diagnostics.bound_connections, 0);
    assert_eq!(diagnostics.counts.cleanups, 1);
    Ok(())
}

#[test]
fn r6_an_evicted_decision_is_a_miss_never_an_allow() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "q"])?;
    // `p` is revoked at 41 while nothing references it, then evicted.
    assert_eq!(
        harness.authority.apply(&revoke(41, "p")?).applied,
        Applied::Change
    );
    harness.authority.prune_unreferenced();
    // An old allowing reply at 40 for `p` arrives for a later admission.
    *harness.decisions.answer_at.lock().map_err(|_| "poisoned")? = Some(40);
    let pass = credential(&harness, 1, 40)?;
    assert_eq!(
        harness.authority.admit(1, &pass, MARKER_HOLDER).err(),
        Some(RevocationRefusal::RevisionUnavailable)
    );
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.counts.stale_replies_refused, 1);
    assert_eq!(diagnostics.counts.admissions_committed, 0);
    Ok(())
}

#[test]
fn r6_a_referenced_deny_is_never_pruned() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    let _withdrawn = harness.authority.apply(&revoke(41, "p")?);
    harness.authority.prune_unreferenced();
    let other = credential(&harness, 2, 40)?;
    let second = harness.authority.admit(2, &other, MARKER_HOLDER)?;
    assert_eq!(
        second.allows(HotAction::Publish, "orders.eu").err(),
        Some(RevocationRefusal::GrantRevoked)
    );
    assert_eq!(
        grants.allows(HotAction::Publish, "orders.eu").err(),
        Some(RevocationRefusal::GrantRevoked)
    );
    Ok(())
}

#[test]
fn r6_repeated_cycles_return_retained_state_to_its_baseline() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let baseline = snapshot(&harness.authority)?;
    for cycle in 1..=20_u64 {
        let pass = credential(&harness, cycle, 40)?;
        let _grants = harness.authority.admit(cycle, &pass, MARKER_HOLDER)?;
        assert!(harness.authority.release(cycle));
    }
    let after = snapshot(&harness.authority)?;
    assert_eq!(after.retained_decisions, baseline.retained_decisions);
    assert_eq!(after.index_entries, 0);
    assert_eq!(after.bound_connections, 0);
    assert_eq!(after.inflight_asks, 0);
    Ok(())
}

#[test]
fn r6_a_refused_close_publication_is_counted_by_name() -> TestResult {
    let harness = harness()?;
    harness.authority.record_close_publication_refused(9);
    assert_eq!(
        snapshot(&harness.authority)?
            .counts
            .close_publication_refused,
        1
    );
    Ok(())
}

// R7 ----------------------------------------------------------------------

#[test]
fn r7_close_writes_are_counted_as_attempted_never_as_delivered() -> TestResult {
    let harness = harness()?;
    for write in [
        CloseWrite::Flushed,
        CloseWrite::Residue,
        CloseWrite::NotQueued,
        CloseWrite::TransportFailed,
        CloseWrite::NoStream,
    ] {
        harness.authority.record_close_write(1, write);
    }
    let counts = snapshot(&harness.authority)?.counts;
    assert_eq!(counts.closes_attempted, 5);
    assert_eq!(counts.closes_flushed, 1);
    assert_eq!(counts.close_transport_failures, 2);
    Ok(())
}

#[test]
fn r7_a_gap_reports_its_length_and_a_loss_reports_unknown() -> TestResult {
    let harness = harness()?;
    assert_eq!(snapshot(&harness.authority)?.missing, GapLength::Unknown);
    settle(&harness, 40, &["conn"])?;
    assert_eq!(snapshot(&harness.authority)?.missing, GapLength::None);
    let _gap = harness.authority.apply(&revoke(42, "conn")?);
    let gap = snapshot(&harness.authority)?;
    assert_eq!(gap.missing, GapLength::Known(1));
    assert_eq!(gap.high_water, Some(Revision::new(42)));
    assert_eq!(gap.reason, Some(RevocationRefusal::StreamGap));
    let _lost = harness.authority.apply(&GrantFeedEvent::Lost);
    let lost = snapshot(&harness.authority)?;
    assert_eq!(lost.missing, GapLength::Unknown);
    assert_eq!(lost.high_water, None);
    assert_eq!(lost.transitions.len(), 3);
    Ok(())
}

#[test]
fn r7_a_repeated_ready_rechecks_nothing() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn", "p", "q"])?;
    let pass = credential(&harness, 1, 40)?;
    let _grants = harness.authority.admit(1, &pass, MARKER_HOLDER)?;
    let repeated = harness.authority.apply(&ready(40)?);
    assert_eq!(repeated.applied, Applied::Duplicate);
    assert_eq!(snapshot(&harness.authority)?.counts.resume_rechecks, 0);
    Ok(())
}

#[test]
fn r7_the_transition_ring_is_bounded_and_its_overflow_is_counted() -> TestResult {
    let harness = harness()?;
    settle(&harness, 40, &["conn"])?;
    let transitions = MAX_DIAGNOSTIC_EVENTS.saturating_add(10);
    for _ in 0..transitions {
        let _lost = harness.authority.apply(&GrantFeedEvent::Lost);
        let _ready = harness.authority.apply(&ready(40)?);
    }
    let diagnostics = snapshot(&harness.authority)?;
    assert_eq!(diagnostics.transitions.len(), MAX_DIAGNOSTIC_EVENTS);
    assert!(diagnostics.transitions_dropped > 0);
    Ok(())
}

#[test]
fn r7_without_any_decision_source_cold_asks_are_refused_by_name() -> TestResult {
    let producer = Arc::new(Producer::default());
    let authority = RevocationAuthority::new(
        policy()?,
        Arc::clone(&producer) as Arc<dyn GrantBindingSource>,
        Arc::new(UnpublishedGrantDecisions),
    );
    assert_eq!(
        authority.apply(&baseline(40, &["conn"])?).applied,
        Applied::Baseline
    );
    assert_eq!(authority.apply(&ready(40)?).applied, Applied::Ready);
    let pass = b"cold".to_vec();
    producer.produce(&pass, binding_for(&pass, 40, standard_rights()?)?)?;
    assert_eq!(
        authority.admit(1, &pass, MARKER_HOLDER).err(),
        Some(RevocationRefusal::RevisionUnavailable)
    );
    let diagnostics = snapshot(&authority)?;
    assert_eq!(diagnostics.counts.decision_asks, 1);
    assert_eq!(diagnostics.inflight_asks, 0);
    Ok(())
}