liminal-server 0.25.4

//! The publish reply's two additions, observed at the response frame.
//!
//! An unresolved publish answers with its own reason code on the existing
//! `PublishError` frame, chosen by the error's type, while every other publish
//! error keeps the undifferentiated code. A successful publish carries the
//! outcome's durable position on its ack only on a connection whose handshake
//! selected `PUBLISH_ACK_POSITION_VERSION`; a connection at 1.0 keeps the
//! older ack.

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

use liminal::durability::{DurableStore, open_ephemeral};
use liminal::protocol::{CausalContext, MessageEnvelope, SchemaId};

use super::*;
use crate::config::types::{LimitsConfig, ServerConfig};
use crate::server::connection::channel_registry::ChannelRegistration;
use crate::server::connection::conversation::ConnectionConversation;
use crate::server::connection::services::{
    ConnectionSubscription, LiminalConnectionServices, PublishOutcome,
};

/// Fixed connection pid, as in the sibling `apply_frame` unit tests.
const TEST_PID: u64 = 1;

/// The durable position the scripted outcomes carry, as the ack must echo it.
const POSITION: u64 = 7717;

// ---------------------------------------------------------------------------
// The typed unresolved error and the outcome's durable position
// ---------------------------------------------------------------------------

#[test]
fn publish_unresolved_display_names_channel_key_sequence_and_message() {
    let text = ServerError::PublishUnresolved {
        channel: "orders".into(),
        idempotency_key: Some("k1".into()),
        sequence: Some(7717),
        message: "injected".into(),
    }
    .to_string();
    for part in ["orders", "k1", "7717", "injected"] {
        assert!(text.contains(part), "{text:?} must name {part:?}");
    }
}

#[test]
fn publish_unresolved_display_names_channel_and_message_without_key_or_sequence() {
    let text = ServerError::PublishUnresolved {
        channel: "orders".into(),
        idempotency_key: None,
        sequence: None,
        message: "m".into(),
    }
    .to_string();
    assert!(text.contains("orders"), "{text:?} must name the channel");
    assert!(text.contains('m'), "{text:?} must carry the message");
}

/// This card adds the field; CHANNELPUBLISHRECOVERY-001 sets it. A publish
/// through the real adapter answers `None`.
#[test]
fn publish_outcome_carries_no_durable_position_in_this_card() -> Result<(), Box<dyn Error>> {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let services = LiminalConnectionServices::from_config_with_store(&config(), store)?;
    services.register_channel(&ChannelRegistration {
        name: "orders".to_owned(),
        schema_bytes: None,
        durable: false,
    })?;

    let outcome = services.publish(7, "orders", &envelope(), None)?;

    assert_eq!(outcome.durable_position, None);
    Ok(())
}

// ---------------------------------------------------------------------------
// PublishError: the unresolved code and the undifferentiated code
// ---------------------------------------------------------------------------

#[test]
fn unresolved_publish_answers_its_own_reason_code() -> Result<(), Box<dyn Error>> {
    let action = Connection::new(Script::Unresolved).apply(publish_frame());

    let FrameAction::Respond(Frame::PublishError {
        stream_id,
        reason_code,
        message,
        ..
    }) = action
    else {
        return Err(format!("expected a PublishError, got {action:?}").into());
    };
    assert_eq!(stream_id, 3);
    assert_eq!(reason_code, 0x0103);
    let text = message.ok_or("the unresolved reply must carry its message")?;
    assert!(text.contains("orders"), "{text:?} must name the channel");
    assert!(
        text.contains("k1"),
        "{text:?} must name the idempotency key"
    );
    Ok(())
}

#[test]
fn definite_publish_error_keeps_the_undifferentiated_code() -> Result<(), Box<dyn Error>> {
    let action = Connection::new(Script::ListenerAccept).apply(publish_frame());

    let FrameAction::Respond(Frame::PublishError {
        reason_code,
        message,
        ..
    }) = action
    else {
        return Err(format!("expected a PublishError, got {action:?}").into());
    };
    assert_eq!(reason_code, 0xFFFF);
    assert_eq!(
        message,
        Some(
            ServerError::ListenerAccept {
                message: "m".into()
            }
            .to_string()
        )
    );
    Ok(())
}

// ---------------------------------------------------------------------------
// PublishAck: the durable position only on a connection that negotiated it
// ---------------------------------------------------------------------------

#[test]
fn ack_carries_the_position_on_a_connection_that_negotiated_it() -> Result<(), Box<dyn Error>> {
    let mut connection = Connection::new(Script::Outcome(Some(POSITION)));
    connection.handshake(ProtocolVersion::new(1, 1))?;

    let action = connection.apply(publish_frame());

    assert!(
        matches!(
            action,
            FrameAction::Respond(Frame::PublishAck {
                message_id: 42,
                durable_position: Some(POSITION),
                ..
            })
        ),
        "got {action:?}"
    );
    Ok(())
}

#[test]
fn ack_keeps_the_older_layout_on_a_connection_at_1_0() -> Result<(), Box<dyn Error>> {
    let mut connection = Connection::new(Script::Outcome(Some(POSITION)));
    connection.handshake(ProtocolVersion::new(1, 0))?;

    let action = connection.apply(publish_frame());

    let FrameAction::Respond(Frame::PublishAck {
        message_id,
        durable_position,
        flags,
        ..
    }) = action
    else {
        return Err(format!("expected a PublishAck, got {action:?}").into());
    };
    assert_eq!(message_id, 42);
    assert_eq!(durable_position, None);
    assert_eq!(flags & 0x02, 0);
    Ok(())
}

#[test]
fn ack_without_a_position_keeps_the_older_layout() -> Result<(), Box<dyn Error>> {
    let mut connection = Connection::new(Script::Outcome(None));
    connection.handshake(ProtocolVersion::new(1, 1))?;

    let action = connection.apply(publish_frame());

    assert!(
        matches!(
            action,
            FrameAction::Respond(Frame::PublishAck {
                durable_position: None,
                ..
            })
        ),
        "got {action:?}"
    );
    Ok(())
}

// ---------------------------------------------------------------------------
// Fixtures
// ---------------------------------------------------------------------------

/// What the scripted adapter's publish answers.
#[derive(Clone, Copy, Debug)]
enum Script {
    /// A successful publish whose outcome carries this durable position.
    Outcome(Option<u64>),
    /// A publish whose append outcome is not known.
    Unresolved,
    /// A definite publish failure.
    ListenerAccept,
}

/// A connection services double whose publish answers its script; every other
/// operation is refused, since none of these tests reaches one.
#[derive(Debug)]
struct ScriptedPublish {
    script: Script,
}

impl ConnectionServices for ScriptedPublish {
    fn publish(
        &self,
        _connection: u64,
        _channel: &str,
        _envelope: &MessageEnvelope,
        _idempotency_key: Option<&str>,
    ) -> Result<PublishOutcome, ServerError> {
        match self.script {
            Script::Outcome(durable_position) => Ok(PublishOutcome {
                message_id: 42,
                delivered: false,
                durable_position,
                reach: crate::server::connection::services::Reach::NoSubscriber,
            }),
            Script::Unresolved => Err(ServerError::PublishUnresolved {
                channel: "orders".to_owned(),
                idempotency_key: Some("k1".to_owned()),
                sequence: Some(0),
                message: "append outcome unknown".to_owned(),
            }),
            Script::ListenerAccept => Err(ServerError::ListenerAccept {
                message: "m".to_owned(),
            }),
        }
    }

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

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

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

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

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

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

fn unused(operation: &str) -> ServerError {
    ServerError::ListenerAccept {
        message: format!("{operation} is not scripted in the publish-outcome tests"),
    }
}

/// One connection over a scripted adapter, carrying its state across frames.
struct Connection {
    runtime: ConnectionRuntime,
    state: ConnectionProcessState,
}

impl Connection {
    fn new(script: Script) -> Self {
        let services: Arc<dyn ConnectionServices> = Arc::new(ScriptedPublish { script });
        Self {
            runtime: ConnectionRuntime::for_tests(services),
            state: ConnectionProcessState::default(),
        }
    }

    fn apply(&mut self, frame: Frame) -> FrameAction {
        apply_frame(TEST_PID, &self.runtime, &mut self.state, frame)
    }

    /// Sends a `Connect` offering 1.0 through `max_version` and requires the
    /// server to select `max_version`.
    fn handshake(&mut self, max_version: ProtocolVersion) -> Result<(), Box<dyn Error>> {
        let action = self.apply(Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version,
            auth_token: Vec::new(),
        });
        let FrameAction::Respond(Frame::ConnectAck {
            selected_version, ..
        }) = action
        else {
            return Err(format!("expected a ConnectAck, got {action:?}").into());
        };
        assert_eq!(selected_version, max_version);
        Ok(())
    }
}

fn envelope() -> MessageEnvelope {
    MessageEnvelope::new(
        SchemaId::new([0_u8; SchemaId::WIRE_LEN]),
        CausalContext::independent(),
        br#"{"order":1}"#.to_vec(),
    )
}

fn publish_frame() -> Frame {
    Frame::Publish {
        flags: 0,
        stream_id: 3,
        channel: "orders".to_owned(),
        envelope: envelope(),
        idempotency_key: Some("k1".to_owned()),
    }
}

/// The channel-free config the roster starts from, with a cap declared so a
/// runtime registration is admitted.
fn config() -> ServerConfig {
    ServerConfig {
        listen_address: std::net::SocketAddr::from(([127, 0, 0, 1], 0)),
        extra_listen_addresses: Vec::new(),
        health_listen_address: std::net::SocketAddr::from(([127, 0, 0, 1], 0)),
        drain_timeout_ms: 30_000,
        channels: Vec::new(),
        routing_rules: Vec::new(),
        persistence_path: None,
        cluster: None,
        auth: None,
        services: crate::config::types::ServicesConfig::default(),
        limits: LimitsConfig {
            max_channels: Some(4),
            ..LimitsConfig::default()
        },
        participant: None,
        websocket: None,
    }
}