liminal-server 0.20.1

//! The offline participant log migration: a migrated store replays to the
//! state the unmigrated store did, a second run changes nothing, an
//! interrupted run is checked and completed, and a row that cannot be
//! converted refuses the migration by name with the log left authoritative.

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

use liminal::durability::bridge::block_on;
use liminal::durability::{DurableStore, open_ephemeral};
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
    ParticipantAck, RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use crate::config::types::ParticipantConfig;

use super::ProductionParticipantHandler;
use super::log::{
    GENERATION_KEY_PREFIX, OperationLog, SCHEMA_VERSION, SCHEMA_VERSION_V3, legacy_rows,
};
use super::log_migration::{
    ConversationLogOutcome, ParticipantLogMigrationError, migrate_participant_logs,
};
use super::ops_session_replay::LogRewrite;
use super::outbox::ConversationOutboxLimits;
use super::outbox_log::OutboxLog;
use super::restore_checkpoint::CheckpointSchema;
use super::state::ConversationAuthority;
use super::tests::{dispatch, test_participant_config};
use super::tests_marker_ack_fixture::{marker_fixture_config, prepare_marker_fixture_with_store};
use super::tests_outcome_unknown::{UnknownOutcomeStore, Write};

const RECORDS: u64 = 0x316;
const CYCLES: u64 = 40;

fn enroll(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    token: u8,
) -> Result<u64, Box<dyn Error>> {
    let enrolled = dispatch(
        handler,
        connection,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: RECORDS,
            enrollment_token: EnrollmentToken::new([token; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(receipt) = enrolled else {
        return Err(format!("enrollment {token:#x} did not bind: {enrolled:?}").into());
    };
    Ok(receipt.participant_id())
}

fn ack(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    participant_id: u64,
    through_seq: u64,
) -> Result<(), Box<dyn Error>> {
    let answered = dispatch(
        handler,
        connection,
        ClientRequest::ParticipantAck(ParticipantAck {
            conversation_id: RECORDS,
            participant_id,
            capability_generation: Generation::ONE,
            through_seq,
        }),
    )?;
    if matches!(answered, ServerValue::AckCommitted(_)) {
        Ok(())
    } else {
        Err(format!("ack through {through_seq} did not commit: {answered:?}").into())
    }
}

fn admit(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    sender: u64,
    cycle: u64,
) -> Result<u64, Box<dyn Error>> {
    let mut token = [0_u8; 16];
    token[..8].copy_from_slice(&cycle.to_be_bytes());
    let committed = dispatch(
        handler,
        connection,
        ClientRequest::RecordAdmission(RecordAdmission {
            conversation_id: RECORDS,
            participant_id: sender,
            capability_generation: Generation::ONE,
            record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
            payload: vec![0x3C; 48],
        }),
    )?;
    let ServerValue::RecordCommitted(committed) = committed else {
        return Err(format!("record {cycle} did not commit: {committed:?}").into());
    };
    Ok(committed.delivery_seq())
}

struct Members {
    recipient_connection: ConnectionIncarnation,
    sender_connection: ConnectionIncarnation,
    recipient: u64,
    sender: u64,
}

/// A records-and-acks conversation written as a store from before schema v4.
fn legacy_records_store() -> Result<(Arc<dyn DurableStore>, Members), Box<dyn Error>> {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let members = Members {
        recipient_connection: ConnectionIncarnation::new(316, 1),
        sender_connection: ConnectionIncarnation::new(316, 2),
        recipient: 0,
        sender: 0,
    };
    let legacy = legacy_rows::arm();
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let recipient = enroll(&handler, members.recipient_connection, 0x31)?;
    let sender = enroll(&handler, members.sender_connection, 0x32)?;
    ack(&handler, members.recipient_connection, recipient, 2)?;
    for cycle in 0..CYCLES {
        let delivered = admit(&handler, members.sender_connection, sender, cycle)?;
        ack(&handler, members.recipient_connection, recipient, delivered)?;
    }
    drop(handler);
    drop(legacy);
    Ok((
        store,
        Members {
            recipient,
            sender,
            ..members
        },
    ))
}

/// Schema version of every row of the conversation's authoritative log.
fn row_versions(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<Vec<u64>, Box<dyn Error>> {
    let log = OperationLog::new(Arc::clone(store), conversation_id);
    let mut versions = Vec::new();
    loop {
        let page = block_on(log.read_raw_page(u64::try_from(versions.len())?))??;
        let short = page.len() < super::log::READ_BATCH_SIZE;
        for entry in page {
            let json: serde_json::Value = serde_json::from_slice(&entry.payload)?;
            versions.push(
                json["schema_version"]
                    .as_u64()
                    .ok_or("row carries no schema version")?,
            );
        }
        if short {
            return Ok(versions);
        }
    }
}

/// One cold-booted owner, read for a comparison across the migration: the
/// conversation state, and apart from it the checkpoint publication the owner
/// holds, with the prefix witness on its own. The witness (the checkpoint's
/// schema phase and its digest over the stored row bytes) is the one part of
/// an owner the migration is meant to change: it rewrites every v3 row as a v4
/// row, so a rebuild over the migrated log reads other bytes under another
/// phase, while the identity history root is captured from the state alone.
#[derive(Debug, PartialEq, Eq)]
struct BootedOwner {
    /// Every field of the authority but the checkpoint publication.
    state: String,
    /// The history root the owner holds, and the metadata's root and merged
    /// boundaries; the metadata's positions follow from the state, these and
    /// the schema phase.
    publication: String,
    /// The metadata's schema phase and prefix digest.
    witness: Option<(CheckpointSchema, [u8; 32])>,
}

/// Every loaded conversation's complete owner, after a cold boot.
fn booted_states(
    store: &Arc<dyn DurableStore>,
    config: ParticipantConfig,
) -> Result<BTreeMap<u64, BootedOwner>, Box<dyn Error>> {
    let handler = ProductionParticipantHandler::new(Arc::clone(store), config)?;
    assert!(
        handler.unloadable_conversations().is_empty(),
        "a conversation failed to load: {:?}",
        handler.unloadable_conversations()
    );
    let mut states = BTreeMap::new();
    for conversation_id in handler.registered_conversation_ids()? {
        let cell = handler.cell(conversation_id)?;
        let mut owner = cell.lock().map_err(|_| "conversation lock poisoned")?;
        if let Some(authority) = owner.as_mut() {
            let metadata = authority.checkpoint_metadata.take();
            let root = authority.checkpoint_history_root.take();
            let state = format!("{authority:?}");
            let published = metadata.as_ref().map(|metadata| {
                (
                    *metadata.history_root(),
                    metadata.merged_base_sequence(),
                    metadata.merged_extension_sequence(),
                )
            });
            let publication = format!("{root:?}\n{published:?}");
            let witness = metadata
                .as_ref()
                .map(|metadata| (metadata.schema(), metadata.prefix_digest()));
            authority.checkpoint_metadata = metadata;
            authority.checkpoint_history_root = root;
            states.insert(
                conversation_id,
                BootedOwner {
                    state,
                    publication,
                    witness,
                },
            );
        }
        drop(owner);
    }
    Ok(states)
}

/// The migrated store's owners hold the unmigrated store's conversation state
/// and checkpoint publication exactly. Their prefix witnesses are equal when
/// the migrated boot restored the checkpoint the unmigrated boot published;
/// when it rebuilt instead, its witness is over the rewritten rows, so it
/// must read the v4 phase, never an old one.
fn assert_same_replayed_owners(
    unmigrated: &BTreeMap<u64, BootedOwner>,
    migrated: &BTreeMap<u64, BootedOwner>,
) {
    assert_eq!(
        unmigrated.keys().collect::<Vec<_>>(),
        migrated.keys().collect::<Vec<_>>(),
        "the migrated store loads a different set of conversations"
    );
    for ((conversation_id, before), (_, after)) in unmigrated.iter().zip(migrated) {
        assert_eq!(
            before.state, after.state,
            "conversation {conversation_id}: the migrated store replays to a different state"
        );
        assert_eq!(
            before.publication, after.publication,
            "conversation {conversation_id}: the migrated store holds a different checkpoint \
             publication"
        );
        if before.witness != after.witness {
            assert!(
                matches!(after.witness, Some((CheckpointSchema::V4Suffix, _))),
                "conversation {conversation_id}: the migrated boot's prefix witness differs \
                 from the unmigrated boot's and is not over v4 rows: before {:?} after {:?}",
                before.witness,
                after.witness
            );
        }
    }
}

fn assert_migration_preserves_state(
    store: &Arc<dyn DurableStore>,
    config: ParticipantConfig,
) -> Result<(), Box<dyn Error>> {
    let before = booted_states(store, config)?;
    assert!(!before.is_empty(), "the fixture loaded no conversation");
    for conversation_id in before.keys() {
        let versions = row_versions(store, *conversation_id)?;
        assert!(
            versions.contains(&u64::from(SCHEMA_VERSION_V3)),
            "conversation {conversation_id} was not written as a pre-v4 store: {versions:?}"
        );
    }

    let migration = migrate_participant_logs(store, &config)?;
    for conversation in &migration.conversations {
        let ConversationLogOutcome::Migrated {
            from_generation,
            to_generation,
            rows,
            converted_rows,
            ..
        } = conversation.outcome
        else {
            return Err(format!("conversation was not migrated: {conversation:?}").into());
        };
        assert_eq!((from_generation, to_generation), (0, 1));
        assert!(rows > 0 && converted_rows > 0, "{conversation:?}");
        let versions = row_versions(store, conversation.conversation_id)?;
        assert!(
            versions
                .iter()
                .all(|version| *version == u64::from(SCHEMA_VERSION)),
            "a migrated log kept an old row: {versions:?}"
        );
    }

    let after = booted_states(store, config)?;
    assert_same_replayed_owners(&before, &after);

    let again = migrate_participant_logs(store, &config)?;
    for conversation in &again.conversations {
        assert!(
            matches!(
                conversation.outcome,
                ConversationLogOutcome::Current { generation: 1, .. }
            ),
            "a second run changed a migrated conversation: {conversation:?}"
        );
    }
    Ok(())
}

/// The field case of the native-head branch (Waffles, 10 Oct 01:1x): a store
/// seeded flat by a pre-pruning handler, upgraded by the boot migration, and
/// then driven by a fresh handler whose first live operation commits routed
/// at the native head, with no sequence conflict and no head mismatch.
#[test]
fn a_migrated_records_store_replays_to_the_state_the_unmigrated_store_did()
-> Result<(), Box<dyn Error>> {
    let (store, members) = legacy_records_store()?;
    assert_migration_preserves_state(&store, test_participant_config())?;

    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let delivered = admit(&handler, members.sender_connection, members.sender, CYCLES)?;
    ack(
        &handler,
        members.recipient_connection,
        members.recipient,
        delivered,
    )?;
    Ok(())
}

/// The marker geometry seeded as a store from before schema v4: under the
/// legacy-rows arm the handler and the fixture's own appender both write flat
/// complete-list rows, so the marker drain's record and marker rows sit in the
/// flat log the migration walks and are converted, not left routed behind it.
#[test]
fn a_migrated_marker_store_replays_to_the_state_the_unmigrated_store_did()
-> Result<(), Box<dyn Error>> {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    {
        let _legacy = legacy_rows::arm();
        let fixture = prepare_marker_fixture_with_store(Arc::clone(&store))?;
        drop(fixture);
    }
    assert_migration_preserves_state(&store, marker_fixture_config())
}

/// The rows the migration derives for the records conversation, without
/// writing them.
fn derived_rows(store: &Arc<dyn DurableStore>) -> Result<Vec<Vec<u8>>, Box<dyn Error>> {
    let config = test_participant_config();
    let limits =
        ConversationOutboxLimits::try_new(config.max_retained_record_rows, config.identity_slots)?;
    let mut rewrite = LogRewrite::default();
    let log = OperationLog::new(Arc::clone(store), RECORDS);
    let outbox_log = OutboxLog::new(Arc::clone(store), RECORDS);
    block_on(ConversationAuthority::replay_rewriting(
        RECORDS,
        &log,
        &outbox_log,
        &config,
        limits,
        Some(&mut rewrite),
    ))?
    .map_err(|error| format!("replay refused: {error}"))?;
    Ok(rewrite.rows)
}

#[test]
fn an_interrupted_migration_is_checked_row_for_row_and_completed() -> Result<(), Box<dyn Error>> {
    let (store, _) = legacy_records_store()?;
    let rows = derived_rows(&store)?;
    let partial = rows
        .get(..10)
        .ok_or("the fixture log is too short")?
        .to_vec();
    let target = OperationLog::at_generation(Arc::clone(&store), RECORDS, 1);
    block_on(target.append_encoded(partial, 0))??;
    assert_eq!(
        block_on(OperationLog::new(Arc::clone(&store), RECORDS).generation())??,
        0,
        "a part-written generation must not be authoritative"
    );

    assert_migration_preserves_state_after_resume(&store, rows.len())
}

fn assert_migration_preserves_state_after_resume(
    store: &Arc<dyn DurableStore>,
    rows: usize,
) -> Result<(), Box<dyn Error>> {
    let before = booted_states(store, test_participant_config())?;
    let migration = migrate_participant_logs(store, &test_participant_config())?;
    assert!(matches!(
        migration.conversations.as_slice(),
        [conversation] if matches!(
            conversation.outcome,
            ConversationLogOutcome::Migrated { rows: migrated, to_generation: 1, .. }
                if usize::try_from(migrated).is_ok_and(|migrated| migrated == rows)
        )
    ));
    assert_eq!(before, booted_states(store, test_participant_config())?);
    Ok(())
}

#[test]
fn a_generation_holding_a_different_row_refuses_by_name() -> Result<(), Box<dyn Error>> {
    let (store, _) = legacy_records_store()?;
    let target = OperationLog::at_generation(Arc::clone(&store), RECORDS, 1);
    block_on(target.append_encoded(vec![b"a row no migration wrote".to_vec()], 0))??;
    let refused = migrate_participant_logs(&store, &test_participant_config());
    assert!(
        matches!(
            refused,
            Err(ParticipantLogMigrationError::GenerationDrift {
                conversation_id: RECORDS,
                generation: 1,
                sequence: 0,
            })
        ),
        "{refused:?}"
    );
    assert_eq!(
        block_on(OperationLog::new(Arc::clone(&store), RECORDS).generation())??,
        0
    );
    Ok(())
}

#[test]
fn a_row_that_does_not_replay_refuses_the_migration_and_keeps_the_log() -> Result<(), Box<dyn Error>>
{
    let (store, _) = legacy_records_store()?;
    let rows = row_versions(&store, RECORDS)?.len();
    let key = format!("{}{RECORDS}", super::log::STREAM_PREFIX);
    block_on(store.append(&key, b"not a row".to_vec(), u64::try_from(rows)?))??;
    let refused = migrate_participant_logs(&store, &test_participant_config());
    assert!(
        matches!(
            refused,
            Err(ParticipantLogMigrationError::Replay {
                conversation_id: RECORDS,
                ..
            })
        ),
        "{refused:?}"
    );
    assert_eq!(
        block_on(OperationLog::new(Arc::clone(&store), RECORDS).generation())??,
        0
    );
    Ok(())
}

/// A store whose next flush after a generation compare-and-swap fails, with
/// the swap itself already made.
#[derive(Debug)]
struct FlushFailsAfterSwap {
    inner: Arc<dyn DurableStore>,
    armed: std::sync::atomic::AtomicBool,
}

#[async_trait::async_trait]
impl DurableStore for FlushFailsAfterSwap {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        self.inner.apply_routed_batch(route_key, items).await
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, liminal::durability::DurabilityError> {
        self.inner.get_routed(route_key, key).await
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, liminal::durability::DurabilityError> {
        self.inner.append(stream_key, payload, expected_seq).await
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, liminal::durability::DurabilityError> {
        self.inner
            .append_all(stream_key, payloads, expected_seq)
            .await
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<liminal::durability::StoredEntry>, liminal::durability::DurabilityError> {
        self.inner.read_from(stream_key, offset, limit).await
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<liminal::durability::StoredEntry>, liminal::durability::DurabilityError>
    {
        self.inner.read_at(stream_key, sequence).await
    }

    async fn cas(
        &self,
        key: &str,
        old_value: u64,
        new_value: u64,
    ) -> Result<(), liminal::durability::DurabilityError> {
        self.inner.cas(key, old_value, new_value).await?;
        self.armed.store(true, std::sync::atomic::Ordering::SeqCst);
        Ok(())
    }

    async fn read_stream_next_seq(
        &self,
        stream_key: &str,
    ) -> Result<Option<u64>, liminal::durability::DurabilityError> {
        self.inner.read_stream_next_seq(stream_key).await
    }

    async fn read_value(
        &self,
        key: &str,
    ) -> Result<Option<u64>, liminal::durability::DurabilityError> {
        self.inner.read_value(key).await
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<liminal::durability::ScanCursor>,
        limit: usize,
    ) -> Result<liminal::durability::ScanPage, liminal::durability::DurabilityError> {
        self.inner.scan_page(prefix, cursor, limit).await
    }

    async fn flush(&self) -> Result<(), liminal::durability::DurabilityError> {
        if self.armed.swap(false, std::sync::atomic::Ordering::SeqCst) {
            return Err(liminal::durability::DurabilityError::ConfigError(
                "the test failed the flush after the generation swap".to_owned(),
            ));
        }
        self.inner.flush().await
    }
}

/// A flush that fails after the generation swap is reported as unconfirmed,
/// never as moved or unchanged; reading the generation back and running the
/// migration again leaves the conversation current, replaying to the state it
/// had before the migration.
#[test]
fn a_switch_whose_flush_failed_is_unconfirmed_and_a_second_run_resumes()
-> Result<(), Box<dyn Error>> {
    let (store, _) = legacy_records_store()?;
    let before = booted_states(&store, test_participant_config())?;
    let failing: Arc<dyn DurableStore> = Arc::new(FlushFailsAfterSwap {
        inner: Arc::clone(&store),
        armed: std::sync::atomic::AtomicBool::new(false),
    });
    let refused = migrate_participant_logs(&failing, &test_participant_config());
    let Err(error) = refused else {
        return Err(format!("the migration reported success: {refused:?}").into());
    };
    assert!(
        matches!(
            error,
            ParticipantLogMigrationError::SwitchUnconfirmed {
                conversation_id: RECORDS,
                expected: 0,
                to: 1,
                ..
            }
        ),
        "{error:?}"
    );
    let text = error.to_string();
    assert!(
        text.contains("not known which generation") && !text.contains("unchanged"),
        "the refusal claims to know where the generation stands: {text}"
    );

    let authoritative = block_on(OperationLog::new(Arc::clone(&store), RECORDS).generation())??;
    assert!(
        authoritative == 0 || authoritative == 1,
        "the generation key reads {authoritative}"
    );
    let again = migrate_participant_logs(&store, &test_participant_config())?;
    assert!(
        matches!(
            again.conversations.as_slice(),
            [conversation] if matches!(
                conversation.outcome,
                ConversationLogOutcome::Current { generation: 1, .. }
                    | ConversationLogOutcome::Migrated { to_generation: 1, .. }
            )
        ),
        "the second run did not leave the conversation on generation 1: {again:?}"
    );
    assert_eq!(
        block_on(OperationLog::new(Arc::clone(&store), RECORDS).generation())??,
        1
    );
    assert_eq!(before, booted_states(&store, test_participant_config())?);
    Ok(())
}

/// `OutcomeUnknown` class 9: a generation switch the store applied but answered
/// `OutcomeUnknown` for is settled by reading the key, never reported as not
/// swapped.
#[test]
fn a_switch_answered_outcome_unknown_is_settled_not_reported_unswapped()
-> Result<(), Box<dyn Error>> {
    let (store, _) = legacy_records_store()?;
    let fault = Arc::new(UnknownOutcomeStore::new(
        &store,
        Write::Cas,
        GENERATION_KEY_PREFIX,
    ));
    fault.arm();
    let faulty: Arc<dyn DurableStore> = fault.clone();
    let outcome = migrate_participant_logs(&faulty, &test_participant_config());
    assert!(fault.struck(), "the migration switched no generation");
    assert_eq!(
        block_on(OperationLog::new(Arc::clone(&store), RECORDS).generation())??,
        1,
        "the switch did not land"
    );
    assert!(
        matches!(
            &outcome,
            Ok(migration) if matches!(
                migration.conversations.as_slice(),
                [conversation] if matches!(
                    conversation.outcome,
                    ConversationLogOutcome::Migrated { to_generation: 1, .. }
                )
            )
        ),
        "a switch the store holds was reported {outcome:?}"
    );
    Ok(())
}