liminal-server 0.25.4

//! Test fixtures: connection services that carry a participant service and
//! nothing else, and a participant whose start can be held, so a test can
//! build a supervisor with an incarnation authority and hold its start.

use std::sync::mpsc::{Receiver, Sender, channel};
use std::sync::{Arc, Mutex};

use liminal::durability::{DurableStore, open_ephemeral};
use liminal::protocol::{MessageEnvelope, SchemaId};
use liminal_protocol::wire::{ClientRequest, ServerValue};

use super::conversation::ConnectionConversation;
use super::services::{ConnectionServices, ConnectionSubscription, PublishOutcome};
use super::supervisor::ConnectionSupervisor;
use crate::ServerError;
use crate::server::participant::{
    InstalledParticipantService, ParticipantConnectionContext, ParticipantConnectionConversations,
    ParticipantSemanticError, ParticipantSemanticHandler,
};

/// Connection services with a participant service and no channels.
#[derive(Debug)]
pub(in crate::server) struct ParticipantOnlyServices {
    participant_service: InstalledParticipantService,
}

impl ParticipantOnlyServices {
    pub(in crate::server) const fn new(participant_service: InstalledParticipantService) -> Self {
        Self {
            participant_service,
        }
    }

    fn unsupported(operation: &str) -> ServerError {
        ServerError::ListenerAccept {
            message: format!("participant runtime fixture does not support {operation}"),
        }
    }
}

impl ConnectionServices for ParticipantOnlyServices {
    fn participant_service(&self) -> Option<InstalledParticipantService> {
        Some(self.participant_service.clone())
    }

    fn publish(
        &self,
        _connection: u64,
        _channel: &str,
        _envelope: &MessageEnvelope,
        _idempotency_key: Option<&str>,
    ) -> Result<PublishOutcome, ServerError> {
        Err(Self::unsupported("publish"))
    }

    fn subscribe(
        &self,
        _channel: &str,
        _accepted_schemas: &[SchemaId],
        _install: Option<liminal::channel::InboxInstall>,
    ) -> Result<ConnectionSubscription, ServerError> {
        Err(Self::unsupported("subscribe"))
    }

    fn unsubscribe(&self, _subscription: ConnectionSubscription) -> Result<(), ServerError> {
        Err(Self::unsupported("unsubscribe"))
    }

    fn open_conversation(
        &self,
        _conversation_id: u64,
        _subject: &str,
    ) -> Result<ConnectionConversation, ServerError> {
        Err(Self::unsupported("conversation open"))
    }

    fn conversation_message(
        &self,
        _conversation: &ConnectionConversation,
        _envelope: &MessageEnvelope,
        _op_id: Option<u64>,
    ) -> Result<(), ServerError> {
        Err(Self::unsupported("conversation message"))
    }

    fn close_conversation(&self, _conversation: ConnectionConversation) -> Result<(), ServerError> {
        Err(Self::unsupported("conversation close"))
    }

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

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

fn internal(error: impl std::fmt::Display) -> ParticipantSemanticError {
    ParticipantSemanticError::Internal {
        message: error.to_string(),
    }
}

/// A participant whose restart repair, which every incarnation start runs,
/// says it has begun and then parks until the test answers whether it
/// succeeds.
#[derive(Debug)]
struct HeldRepair {
    entered: Mutex<Sender<()>>,
    release: Mutex<Receiver<bool>>,
}

impl ParticipantSemanticHandler for HeldRepair {
    fn handle(
        &self,
        _context: ParticipantConnectionContext,
        _conversations: &mut ParticipantConnectionConversations,
        _request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        Err(ParticipantSemanticError::Unavailable)
    }

    fn repair_unclean_server_restart(
        &self,
        _current_server_incarnation: u64,
    ) -> Result<(), ParticipantSemanticError> {
        self.entered
            .lock()
            .map_err(internal)?
            .send(())
            .map_err(internal)?;
        let succeeds = self
            .release
            .lock()
            .map_err(internal)?
            .recv()
            .map_err(internal)?;
        if succeeds {
            Ok(())
        } else {
            Err(internal("injected restart repair refusal"))
        }
    }
}

/// A supervisor whose incarnation start, once begun, parks in the restart
/// repair: `entered` says it has begun, and sending on `release` lets it
/// finish (`true`) or fail (`false`).
pub(in crate::server) struct HeldStart {
    pub(in crate::server) supervisor: ConnectionSupervisor,
    pub(in crate::server) entered: Receiver<()>,
    pub(in crate::server) release: Sender<bool>,
}

pub(in crate::server) fn held_start_supervisor() -> Result<HeldStart, Box<dyn std::error::Error>> {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let (entered_send, entered) = channel();
    let (release, released) = channel();
    let participant_service = InstalledParticipantService::new(
        Arc::new(HeldRepair {
            entered: Mutex::new(entered_send),
            release: Mutex::new(released),
        }),
        store,
        u64::MAX,
    )
    .map_err(|error| format!("invalid participant test wire-frame limit: {error:?}"))?;
    let services: Arc<dyn ConnectionServices> =
        Arc::new(ParticipantOnlyServices::new(participant_service));
    Ok(HeldStart {
        supervisor: ConnectionSupervisor::with_services(services)?,
        entered,
        release,
    })
}

/// A participant that answers every request Unavailable and repairs nothing.
#[derive(Debug)]
struct Absorbing;

impl ParticipantSemanticHandler for Absorbing {
    fn handle(
        &self,
        _context: ParticipantConnectionContext,
        _conversations: &mut ParticipantConnectionConversations,
        _request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        Err(ParticipantSemanticError::Unavailable)
    }
}

/// A supervisor whose incarnation authority reads and writes `store`.
pub(in crate::server) fn supervisor_over(
    store: Arc<dyn DurableStore>,
) -> Result<ConnectionSupervisor, Box<dyn std::error::Error>> {
    let participant_service =
        InstalledParticipantService::new(Arc::new(Absorbing), store, u64::MAX)
            .map_err(|error| format!("invalid participant test wire-frame limit: {error:?}"))?;
    let services: Arc<dyn ConnectionServices> =
        Arc::new(ParticipantOnlyServices::new(participant_service));
    Ok(ConnectionSupervisor::with_services(services)?)
}