liminal-server 0.20.0

//! ACCESS-006 R1 and R3 at the wire routes and the final release seams: a
//! membership-protected channel's publish, subscribe, call, served reply,
//! participant conversation and queued delivery are each decided on the
//! connection's verified subject, and nothing is released after a deny.

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use liminal::channel::SchemaId as ChannelSchemaId;
use liminal::envelope::{Envelope, PublisherId};
use liminal::protocol::{
    CONVERSATION_REPLY_REQUESTED_FLAG, CausalContext, Frame, MessageEnvelope, ProtocolVersion,
    SchemaId, decode,
};
use liminal_protocol::credential::WireCredential;
use liminal_protocol::wire::{
    ClientRequest, Generation, RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};
use lys_pass::membership::GrantLog;
use lys_pass::rights::Mode;

use super::{apply_frame, participant_access_refusal};
use crate::ServerError;
use crate::auth_lys::session::ConsumerAdmitter;
use crate::auth_lys::{
    AccessAction, AccessDecisionError, AccessResource, DeliberateDecision, GrantReceipt, HotAction,
    LysDeliberateConsumer, LysPassConsumer, SubjectBinding,
};
use crate::auth_membership::fixture::{
    FixtureAuthority, SeededProvider, config, resource, subject,
};
use crate::auth_membership::{
    AuthorityState, MembershipAction, MembershipAuthority, MembershipChange, MembershipGate,
    MembershipListener, MembershipProvider,
};
use crate::link::LinkAction;
use crate::server::connection::conversation::{ConnectionConversation, ConversationResource};
use crate::server::connection::delivery::service_subscriptions;
use crate::server::connection::outbound::OutboundWriter;
use crate::server::connection::pending_reply::PendingReplyTable;
use crate::server::connection::services::{
    ConnectionServices, ConnectionSubscription, PublishOutcome, Reach, SubscriptionResource,
};
use crate::server::connection::state::{ConnectionProcessState, FrameAction};
use crate::server::connection::supervisor::ConnectionRuntime;
use crate::server::participant::{
    ParticipantConnectionContext, ParticipantConnectionConversations, ParticipantSemanticError,
    ParticipantSemanticHandler,
};

type Failure = Box<dyn std::error::Error>;

const PID: u64 = 21;
const STREAM: u32 = 4;
const WORKSPACE_KIND: &str = "liminal.workspace";
const CHANNEL_KIND: &str = "liminal.channel";
const MAPPED_CONVERSATION: u64 = 42;

fn log() -> GrantLog {
    GrantLog {
        identity: "grant-log-a".to_owned(),
        epoch: 0,
    }
}

/// A pass consumer whose pass allows every hot action, so every refusal
/// below is the membership's and no other's.
#[derive(Debug)]
struct Open {
    holder: &'static str,
    kind: &'static str,
}

impl LysPassConsumer for Open {
    type Pass = ();
    type Holder = str;
    type AdmissionError = std::io::Error;

    fn verify_offline(
        &self,
        bytes: &[u8],
        audience: &str,
        _now: u64,
    ) -> Result<Self::Pass, Self::AdmissionError> {
        if bytes != b"fixture-pass" || audience != "liminal" {
            return Err(std::io::Error::other("signature"));
        }
        Ok(())
    }

    fn verified_holder<'p>(&self, pass: &'p Self::Pass) -> &'p Self::Holder {
        let () = pass;
        self.holder
    }

    fn verified_holder_kind<'p>(&self, pass: &'p Self::Pass) -> Option<&'p str> {
        let () = pass;
        Some(self.kind)
    }

    fn decide_hot(
        &self,
        pass: &Self::Pass,
        subject: &SubjectBinding,
        action: HotAction,
        _now: u64,
    ) -> Result<GrantReceipt, AccessDecisionError> {
        let () = pass;
        Ok(GrantReceipt {
            grant: format!("pass-{}-{action:?}", subject.subject()),
        })
    }

    fn decide_link(
        &self,
        pass: &Self::Pass,
        subject: &SubjectBinding,
        action: LinkAction,
        _now: u64,
    ) -> Result<GrantReceipt, AccessDecisionError> {
        let () = pass;
        Ok(GrantReceipt {
            grant: format!("link-{}-{action:?}", subject.subject()),
        })
    }
}

impl LysDeliberateConsumer for Open {
    fn decide_live(
        &self,
        pass: &Self::Pass,
        resource: &AccessResource,
        action: AccessAction,
        _now: u64,
    ) -> Result<DeliberateDecision, AccessDecisionError> {
        let () = pass;
        Ok(DeliberateDecision::Outright(GrantReceipt {
            grant: format!("open-{}-{action:?}", resource.id),
        }))
    }
}

/// A subscription inbox preloaded with envelopes.
#[derive(Debug)]
struct Queued(VecDeque<Envelope>);

impl SubscriptionResource for Queued {
    fn unsubscribe(self: Box<Self>) -> Result<(), ServerError> {
        Ok(())
    }

    fn try_next(&mut self) -> Option<Envelope> {
        self.0.pop_front()
    }

    fn has_pending(&self) -> bool {
        !self.0.is_empty()
    }
}

/// A conversation whose actor answers nothing.
#[derive(Debug)]
struct Quiet;

impl ConversationResource for Quiet {
    fn message(&self, _envelope: &MessageEnvelope, _op_id: Option<u64>) -> Result<(), ServerError> {
        Ok(())
    }

    fn participant_pids(&self) -> Vec<u64> {
        Vec::new()
    }

    fn has_detected_crash(&self) -> bool {
        false
    }

    fn await_crash(&self, _timeout: Duration) -> Option<Instant> {
        None
    }

    fn receive_reply(&self, _timeout: Duration) -> Result<MessageEnvelope, ServerError> {
        Err(ServerError::ConfigValidation {
            message: "the quiet conversation never replies".to_owned(),
        })
    }

    fn close(self: Box<Self>) -> Result<(), ServerError> {
        Ok(())
    }

    fn finalize(self: Box<Self>) {}
}

/// Services counting every effect that reached them.
#[derive(Debug, Default)]
struct Door {
    publishes: Mutex<usize>,
    subscribes: Mutex<usize>,
    messages: Mutex<usize>,
}

fn count(counter: &Mutex<usize>) -> Result<usize, Failure> {
    Ok(*counter.lock().map_err(|_| "a counter was poisoned")?)
}

fn bump(counter: &Mutex<usize>) -> Result<(), ServerError> {
    let mut held = counter.lock().map_err(|_| ServerError::ConfigValidation {
        message: "a counter was poisoned".to_owned(),
    })?;
    *held = held.saturating_add(1);
    drop(held);
    Ok(())
}

impl ConnectionServices for Door {
    fn publish(
        &self,
        _connection: u64,
        _channel: &str,
        _envelope: &MessageEnvelope,
        _idempotency_key: Option<&str>,
    ) -> Result<PublishOutcome, ServerError> {
        bump(&self.publishes)?;
        Ok(PublishOutcome {
            message_id: 1,
            delivered: false,
            durable_position: None,
            reach: Reach::NoSubscriber,
        })
    }

    fn subscribe(
        &self,
        _channel: &str,
        _accepted_schemas: &[SchemaId],
        _install: Option<liminal::channel::InboxInstall>,
    ) -> Result<ConnectionSubscription, ServerError> {
        bump(&self.subscribes)?;
        let queue = [b"first".to_vec(), b"second".to_vec()]
            .into_iter()
            .map(|payload| {
                Envelope::new(
                    payload,
                    None,
                    ChannelSchemaId::new(),
                    PublisherId::default(),
                )
            })
            .collect();
        Ok(ConnectionSubscription::new(
            7,
            SchemaId::new([3; SchemaId::WIRE_LEN]),
            Box::new(Queued(queue)),
        ))
    }

    fn unsubscribe(&self, _subscription: ConnectionSubscription) -> Result<(), ServerError> {
        Ok(())
    }

    fn open_conversation(
        &self,
        _conversation_id: u64,
        subject: &str,
    ) -> Result<ConnectionConversation, ServerError> {
        Ok(ConnectionConversation::new(
            Box::new(Quiet),
            subject.to_owned(),
        ))
    }

    fn conversation_message(
        &self,
        _conversation: &ConnectionConversation,
        _envelope: &MessageEnvelope,
        _op_id: Option<u64>,
    ) -> Result<(), ServerError> {
        bump(&self.messages)
    }

    fn close_conversation(&self, _conversation: ConnectionConversation) -> Result<(), ServerError> {
        Ok(())
    }

    fn flush_durable_state(&self) -> Result<(), ServerError> {
        Ok(())
    }
}

/// A ward of two channels: `reader` reads `ward.a`; `poster` posts on
/// `ward.a`; `both` reads and posts on `ward.b`.
fn world() -> Result<Arc<SeededProvider>, Failure> {
    let workspace = resource(WORKSPACE_KIND, "ward");
    let world = SeededProvider::new(&log())
        .place(resource(CHANNEL_KIND, "ward-a"), workspace.clone(), false)
        .place(resource(CHANNEL_KIND, "ward-b"), workspace, false);
    world.grant(
        "reader-grant",
        subject("reader", "person"),
        resource(CHANNEL_KIND, "ward-a"),
        &["read"],
        Mode::Outright,
    )?;
    world.grant(
        "poster-grant",
        subject("poster", "person"),
        resource(CHANNEL_KIND, "ward-a"),
        &["post"],
        Mode::Outright,
    )?;
    world.grant(
        "both-grant",
        subject("both", "person"),
        resource(CHANNEL_KIND, "ward-b"),
        &["read", "post"],
        Mode::Outright,
    )?;
    Ok(Arc::new(world))
}

struct Harness {
    runtime: ConnectionRuntime,
    state: ConnectionProcessState,
    door: Arc<Door>,
    gate: Arc<MembershipGate>,
    authority: Arc<FixtureAuthority>,
}

/// A connected process over `door`, admitted as `holder` (or on no pass
/// when `None`), with the membership gate over `world` installed.
fn connected(
    world: &Arc<SeededProvider>,
    holder: Option<&'static str>,
) -> Result<Harness, Failure> {
    let door = Arc::new(Door::default());
    let runtime = ConnectionRuntime::for_tests(Arc::clone(&door) as Arc<dyn ConnectionServices>);
    runtime
        .install_lys_admitter(
            Arc::new(ConsumerAdmitter::new(Arc::new(Open {
                holder: holder.unwrap_or("nobody"),
                kind: "person",
            }))),
            "network-a".to_owned(),
        )
        .map_err(|_| "a lys admitter was already installed")?;
    let gate = Arc::new(MembershipGate::new(
        &config(
            &log(),
            (WORKSPACE_KIND, CHANNEL_KIND),
            &[("ward.a", "ward", "ward-a"), ("ward.b", "ward", "ward-b")],
            &[(MAPPED_CONVERSATION, "ward.a")],
            (16, 65_536, 64),
        ),
        Arc::clone(world) as Arc<dyn MembershipProvider>,
    )?);
    let authority = Arc::new(FixtureAuthority::ready(&log(), world.revision(), 1));
    gate.install_authority(Arc::clone(&authority) as Arc<dyn MembershipAuthority>)
        .map_err(|_| "an authority was already installed")?;
    runtime
        .install_membership_gate(Arc::clone(&gate))
        .map_err(|_| "a membership gate was already installed")?;
    runtime.register_for_tests(PID)?;
    let mut state = ConnectionProcessState::default();
    let version = ProtocolVersion::new(1, 0);
    let auth_token = holder.map_or_else(Vec::new, |_| WireCredential::lys_record(b"fixture-pass"));
    let action = apply_frame(
        PID,
        &runtime,
        &mut state,
        Frame::Connect {
            flags: 0,
            min_version: version,
            max_version: version,
            auth_token,
        },
    );
    if !matches!(action, FrameAction::Respond(Frame::ConnectAck { .. })) {
        return Err(format!("the connection was not admitted: {action:?}").into());
    }
    Ok(Harness {
        runtime,
        state,
        door,
        gate,
        authority,
    })
}

fn envelope() -> MessageEnvelope {
    MessageEnvelope::new(
        SchemaId::new([3; SchemaId::WIRE_LEN]),
        CausalContext::independent(),
        b"patient-note".to_vec(),
    )
}

fn publish(harness: &mut Harness, channel: &str) -> FrameAction {
    apply_frame(
        PID,
        &harness.runtime,
        &mut harness.state,
        Frame::Publish {
            flags: 0,
            stream_id: STREAM,
            channel: channel.to_owned(),
            envelope: envelope(),
            idempotency_key: None,
        },
    )
}

fn subscribe(harness: &mut Harness, channel: &str) -> FrameAction {
    apply_frame(
        PID,
        &harness.runtime,
        &mut harness.state,
        Frame::Subscribe {
            flags: 0,
            stream_id: STREAM,
            channel: channel.to_owned(),
            accepted_schemas: vec![SchemaId::new([3; SchemaId::WIRE_LEN])],
            max_in_flight: 8,
        },
    )
}

fn refused_with(action: &FrameAction, name: &str) -> bool {
    let message = match action {
        FrameAction::Respond(
            Frame::PublishError { message, .. }
            | Frame::SubscribeError { message, .. }
            | Frame::ConversationError { message, .. },
        ) => message.as_deref(),
        _ => None,
    };
    message.is_some_and(|message| message.starts_with(name))
}

#[test]
fn a_read_only_holder_subscribes_and_cannot_publish() -> Result<(), Failure> {
    let world = world()?;
    let mut harness = connected(&world, Some("reader"))?;
    assert!(matches!(
        subscribe(&mut harness, "ward.a"),
        FrameAction::Respond(Frame::SubscribeAck { .. })
    ));
    assert!(refused_with(&publish(&mut harness, "ward.a"), "NotHeld"));
    assert!(refused_with(&publish(&mut harness, "ward.b"), "NotHeld"));
    assert!(refused_with(&subscribe(&mut harness, "ward.b"), "NotHeld"));
    assert_eq!(count(&harness.door.publishes)?, 0);
    assert_eq!(count(&harness.door.subscribes)?, 1);
    Ok(())
}

#[test]
fn a_posting_holder_publishes_only_on_its_channel() -> Result<(), Failure> {
    let world = world()?;
    let mut harness = connected(&world, Some("poster"))?;
    assert!(matches!(
        publish(&mut harness, "ward.a"),
        FrameAction::Respond(Frame::PublishAck { .. })
    ));
    assert!(refused_with(&publish(&mut harness, "ward.b"), "NotHeld"));
    assert!(refused_with(&subscribe(&mut harness, "ward.a"), "NotHeld"));
    assert_eq!(count(&harness.door.publishes)?, 1);
    assert_eq!(count(&harness.door.subscribes)?, 0);
    Ok(())
}

#[test]
fn a_connection_without_a_lys_holder_gets_nothing_protected_and_keeps_the_rest()
-> Result<(), Failure> {
    let world = world()?;
    let mut harness = connected(&world, None)?;
    assert!(refused_with(
        &publish(&mut harness, "ward.a"),
        "membership_holder_unverified"
    ));
    assert!(refused_with(
        &subscribe(&mut harness, "ward.a"),
        "membership_holder_unverified"
    ));
    // An unprotected channel keeps its prior contract.
    assert!(matches!(
        publish(&mut harness, "general"),
        FrameAction::Respond(Frame::PublishAck { .. })
    ));
    assert_eq!(count(&harness.door.publishes)?, 1);
    assert_eq!(world.asked(), 0);
    Ok(())
}

fn drained(state: &mut ConnectionProcessState) -> Result<(Vec<Frame>, Vec<u64>), Failure> {
    let mut outbound = OutboundWriter::with_capacity(1 << 20);
    let shed = service_subscriptions(state, &mut outbound, 32)
        .map_err(|error| format!("the pump failed: {error}"))?;
    let bytes = outbound.take_bytes();
    let mut frames = Vec::new();
    let mut rest = bytes.as_slice();
    while !rest.is_empty() {
        let (frame, consumed) = decode(rest).map_err(|error| format!("decode: {error}"))?;
        frames.push(frame);
        rest = rest.get(consumed..).unwrap_or(&[]);
    }
    Ok((frames, shed))
}

#[test]
fn a_revoked_read_releases_no_queued_delivery_and_retires_the_subscription() -> Result<(), Failure>
{
    let world = world()?;
    let mut harness = connected(&world, Some("reader"))?;
    assert!(matches!(
        subscribe(&mut harness, "ward.a"),
        FrameAction::Respond(Frame::SubscribeAck { .. })
    ));
    let revision = world.revoke("reader-grant")?;
    let applied = harness.gate.changed(&MembershipChange {
        log: log(),
        revision,
        subject: Some(subject("reader", "person")),
        resource: resource(CHANNEL_KIND, "ward-a"),
    })?;
    assert_eq!(applied.affected, 1);
    let (frames, shed) = drained(&mut harness.state)?;
    assert_eq!(shed, vec![7]);
    assert!(
        frames
            .iter()
            .all(|frame| !matches!(frame, Frame::Deliver { .. })),
        "no queued byte is released after the deny: {frames:?}"
    );
    assert!(frames.iter().any(|frame| matches!(
        frame,
        Frame::SubscribeError { message: Some(message), .. } if message.starts_with("Revoked")
    )));
    Ok(())
}

#[test]
fn a_lost_authority_holds_queued_deliveries_and_ready_releases_them() -> Result<(), Failure> {
    let world = world()?;
    let mut harness = connected(&world, Some("reader"))?;
    assert!(matches!(
        subscribe(&mut harness, "ward.a"),
        FrameAction::Respond(Frame::SubscribeAck { .. })
    ));
    harness.authority.set(AuthorityState::Unavailable {
        cause: "stream_lost".to_owned(),
    })?;
    let (held, shed) = drained(&mut harness.state)?;
    assert!(
        held.is_empty(),
        "nothing is released while the authority is lost"
    );
    assert!(shed.is_empty(), "a hold is not a retirement");
    // A new protected act refuses with the authority's cause, not the last
    // applied allow.
    assert!(refused_with(
        &subscribe(&mut harness, "ward.a"),
        "membership_unavailable"
    ));
    harness.authority.set(AuthorityState::Ready {
        log: log(),
        revision: world.revision(),
        generation: 1,
    })?;
    let (released, _) = drained(&mut harness.state)?;
    assert_eq!(
        released
            .iter()
            .filter(|frame| matches!(frame, Frame::Deliver { .. }))
            .count(),
        2
    );
    Ok(())
}

#[test]
fn a_reply_needs_read_before_the_request_is_forwarded() -> Result<(), Failure> {
    let world = world()?;
    let mut harness = connected(&world, Some("poster"))?;
    // The poster may call on ward.a, but may not read a reply there.
    assert_eq!(
        apply_frame(
            PID,
            &harness.runtime,
            &mut harness.state,
            Frame::ConversationOpen {
                flags: 0,
                stream_id: STREAM,
                conversation_id: 9,
                subject: "ward.a".to_owned(),
            },
        ),
        FrameAction::NoResponse
    );
    let asked = apply_frame(
        PID,
        &harness.runtime,
        &mut harness.state,
        Frame::ConversationMessage {
            flags: CONVERSATION_REPLY_REQUESTED_FLAG,
            stream_id: STREAM,
            conversation_id: 9,
            envelope: envelope(),
        },
    );
    assert!(refused_with(&asked, "NotHeld"));
    assert_eq!(
        count(&harness.door.messages)?,
        0,
        "the request was never forwarded"
    );
    Ok(())
}

#[test]
fn a_queued_reply_after_a_revoked_read_is_refused_at_release() -> Result<(), Failure> {
    let world = world()?;
    let harness = connected(&world, Some("both"))?;
    let release = harness
        .state
        .membership
        .release_check("ward.b", MembershipAction::Read)
        .ok_or("ward.b is protected and the holder verified")?;
    let mut table = PendingReplyTable::default();
    let op_id = table.admit(9, STREAM, Instant::now())?;
    table.bind_release(op_id, release.clone());
    let revision = world.revoke("both-grant")?;
    harness.gate.changed(&MembershipChange {
        log: log(),
        revision,
        subject: None,
        resource: resource(CHANNEL_KIND, "ward-b"),
    })?;
    let answered = table.match_reply(9, op_id, envelope());
    assert!(
        matches!(
            &answered,
            Some(Frame::ConversationError { message: Some(message), .. }) if message.starts_with("Revoked")
        ),
        "the reply's bytes are not released: {answered:?}"
    );
    // A granted reply on an unprotected binding is released unchanged.
    let unbound = table.admit(9, STREAM, Instant::now())?;
    assert!(matches!(
        table.match_reply(9, unbound, envelope()),
        Some(Frame::ConversationMessage { .. })
    ));
    assert!(release.check().is_err());
    Ok(())
}

/// A participant handler that places no conversation itself: the subject
/// comes from the membership mapping or nowhere.
#[derive(Debug)]
struct Unplaced;

impl ParticipantSemanticHandler for Unplaced {
    fn handle(
        &self,
        _context: ParticipantConnectionContext,
        _conversations: &mut ParticipantConnectionConversations,
        _request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        Err(ParticipantSemanticError::Internal {
            message: "the participant handler is not a route under test".to_owned(),
        })
    }
}

fn admission(conversation_id: u64) -> ClientRequest {
    ClientRequest::RecordAdmission(RecordAdmission {
        conversation_id,
        participant_id: 2,
        capability_generation: Generation::ONE,
        record_admission_attempt_token: RecordAdmissionAttemptToken::new([0xA7; 16]),
        payload: vec![1, 2, 3],
    })
}

#[test]
fn a_mapped_participant_conversation_is_decided_on_its_mapped_channel() -> Result<(), Failure> {
    let world = world()?;
    let poster = connected(&world, Some("poster"))?;
    // RecordAdmission is a publish: the poster's post on ward.a allows it.
    assert_eq!(
        participant_access_refusal(
            &poster.runtime,
            &poster.state,
            &Unplaced,
            &admission(MAPPED_CONVERSATION)
        ),
        None
    );
    let reader = connected(&world, Some("reader"))?;
    let refused = participant_access_refusal(
        &reader.runtime,
        &reader.state,
        &Unplaced,
        &admission(MAPPED_CONVERSATION),
    )
    .ok_or("the reader may not post on ward.a")?;
    assert!(refused.starts_with("NotHeld"), "{refused}");
    // An unmapped conversation is still refused by name, never guessed.
    let unmapped =
        participant_access_refusal(&poster.runtime, &poster.state, &Unplaced, &admission(43))
            .ok_or("an unmapped conversation was not refused")?;
    assert!(
        unmapped.contains("no subject mapping for conversation 43"),
        "{unmapped}"
    );
    // A participant delivery on the mapped conversation is released to the
    // reader and held from the poster.
    assert!(
        reader
            .state
            .membership
            .release_conversation(MAPPED_CONVERSATION)
            .is_ok()
    );
    assert!(
        poster
            .state
            .membership
            .release_conversation(MAPPED_CONVERSATION)
            .is_err()
    );
    Ok(())
}

#[test]
fn closing_the_connection_releases_its_bindings() -> Result<(), Failure> {
    let world = world()?;
    let mut harness = connected(&world, Some("reader"))?;
    assert!(matches!(
        subscribe(&mut harness, "ward.a"),
        FrameAction::Respond(Frame::SubscribeAck { .. })
    ));
    assert_eq!(harness.gate.retained()?, 1);
    let Harness { state, gate, .. } = harness;
    drop(state);
    assert_eq!(gate.retained()?, 0);
    assert_eq!(gate.counts().released, 1);
    Ok(())
}