liminal-server 0.19.1

//! One-time offline rewrite of every participant conversation log into the
//! current row grammar.
//!
//! Rows written before schema v4 carry the complete retained-charge list, so a
//! conversation's log grows with the square of its length and its replay
//! decodes all of it. The rewrite replays each conversation once, re-encodes
//! every row in log order with its retained-charge audit recorded as the change
//! the row made, and writes the result as the conversation's next log
//! generation. Nothing is authoritative until the new generation replays to
//! exactly the state the old one does; only then does one compare-and-swap of
//! the conversation's generation key make it the log every reader uses.
//!
//! Per conversation the rewrite is atomic (the switch is one compare-and-swap),
//! idempotent (a conversation already on current rows is reported and left
//! alone; a new generation left part-written by an interrupted run is checked
//! row for row and completed), and it never skips a row: a row that does not
//! replay, re-encode, or match what an earlier run wrote refuses the migration
//! by name with the conversation's generation unchanged.
//!
//! It runs offline, over a store no server has open.

use std::sync::Arc;

use liminal::durability::DurableStore;
use liminal::durability::bridge::block_on;

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

use super::log::{GenerationSwitchFailure, OperationLog};
use super::ops_session_replay::LogRewrite;
use super::outbox::ConversationOutboxLimits;
use super::outbox_log::OutboxLog;
use super::outbox_replay::RestoreError;
use super::registry::ConversationRegistry;
use super::state::ConversationAuthority;

/// Rows appended to a new generation per store write.
const WRITE_BATCH_ROWS: usize = 256;

/// What the migration did to every registered conversation, in registry order.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ParticipantLogMigration {
    /// One entry per registered conversation.
    pub conversations: Vec<ConversationLogMigration>,
}

/// What the migration did to one conversation.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ConversationLogMigration {
    /// The conversation.
    pub conversation_id: u64,
    /// The result for it.
    pub outcome: ConversationLogOutcome,
}

/// The result of migrating one conversation.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ConversationLogOutcome {
    /// The conversation has no log rows: its registry row was written and its
    /// genesis never was.
    Empty,
    /// The authoritative log already holds only current rows.
    Current {
        /// The authoritative generation, unchanged.
        generation: u64,
        /// Rows in it.
        rows: u64,
    },
    /// The log was rewritten into a new generation that is now authoritative.
    Migrated {
        /// The generation that was authoritative before.
        from_generation: u64,
        /// The generation that is authoritative now.
        to_generation: u64,
        /// Rows rewritten, every row of the log.
        rows: u64,
        /// Rows whose complete retained-charge list became a change.
        converted_rows: u64,
        /// Encoded bytes of the new generation.
        bytes: u64,
    },
}

/// Why the migration stopped. Every refusal names the conversation, and no
/// conversation's authoritative generation is changed by a refused run.
#[derive(Debug, thiserror::Error)]
pub enum ParticipantLogMigrationError {
    /// The participant configuration's outbox limits are invalid.
    #[error("participant outbox limit configuration failed: {0}")]
    Configuration(String),
    /// The conversation registry could not be read.
    #[error("the participant conversation registry could not be read: {0}")]
    Registry(String),
    /// A conversation's log did not replay, so its rows cannot be rewritten.
    #[error(
        "participant conversation {conversation_id} did not replay and was not migrated: {reason}"
    )]
    Replay {
        /// The conversation.
        conversation_id: u64,
        /// The replay's refusal.
        reason: String,
    },
    /// A conversation's new generation could not be read or written.
    #[error(
        "participant conversation {conversation_id}: generation {generation} could not be written: {reason}"
    )]
    Write {
        /// The conversation.
        conversation_id: u64,
        /// The generation being written.
        generation: u64,
        /// The store's refusal.
        reason: String,
    },
    /// A new generation left by an earlier run holds a row that differs from
    /// the row this run derives at the same sequence, or more rows than the
    /// log has.
    #[error(
        "participant conversation {conversation_id}: generation {generation} already holds a different row at sequence {sequence}"
    )]
    GenerationDrift {
        /// The conversation.
        conversation_id: u64,
        /// The generation holding the stray row.
        generation: u64,
        /// The first sequence that differs.
        sequence: u64,
    },
    /// The rewritten generation replays to a different state than the log it
    /// was rewritten from.
    #[error(
        "participant conversation {conversation_id}: generation {generation} replays to a different state than generation {source_generation}; not switched"
    )]
    StateDrift {
        /// The conversation.
        conversation_id: u64,
        /// The generation it was rewritten from.
        source_generation: u64,
        /// The rewritten generation.
        generation: u64,
    },
    /// The generation key moved while the migration ran, so another writer is
    /// using the store.
    #[error(
        "participant conversation {conversation_id}: the switch of the authoritative generation from {expected} to {to} was refused, so generation {expected} is unchanged by this run: {reason}"
    )]
    Switch {
        /// The conversation.
        conversation_id: u64,
        /// The generation the migration read.
        expected: u64,
        /// The generation the migration wrote.
        to: u64,
        /// The store's refusal.
        reason: String,
    },
    /// The authoritative generation was swapped but the flush that makes the
    /// swap durable failed, so which generation a restart finds is not known.
    /// Both generations replay to the same state; running the migration again
    /// reads the authoritative generation and reports or completes the
    /// conversation.
    #[error(
        "participant conversation {conversation_id}: the authoritative generation was swapped from {expected} to {to} but the flush failed, so it is not known which generation a restart finds; run the migration again, which reads the authoritative generation and completes or reports the conversation: {reason}"
    )]
    SwitchUnconfirmed {
        /// The conversation.
        conversation_id: u64,
        /// The generation the migration read.
        expected: u64,
        /// The generation the migration wrote.
        to: u64,
        /// The store's refusal.
        reason: String,
    },
}

/// Rewrites every registered conversation's log into the current row grammar.
///
/// # Errors
///
/// Returns [`ParticipantLogMigrationError`] at the first conversation that
/// cannot be migrated; conversations migrated before it stay migrated, and
/// every later one keeps its authoritative generation. The refused one keeps
/// its generation too, except after [`ParticipantLogMigrationError::SwitchUnconfirmed`],
/// where either generation may be authoritative (both replay to the same
/// state). Running again resumes where the refused run stopped.
///
/// The migration is forward-only: a binary older than the change-per-row
/// grammar cannot read a migrated conversation, so a migrated store is never
/// run under an older binary.
pub fn migrate_participant_logs(
    store: &Arc<dyn DurableStore>,
    config: &ParticipantConfig,
) -> Result<ParticipantLogMigration, ParticipantLogMigrationError> {
    let limits =
        ConversationOutboxLimits::try_new(config.max_retained_record_rows, config.identity_slots)
            .map_err(|error| ParticipantLogMigrationError::Configuration(error.to_string()))?;
    let conversation_ids = ConversationRegistry::new(Arc::clone(store))
        .restore()
        .map_err(|error| ParticipantLogMigrationError::Registry(error.to_string()))?;
    let mut migration = ParticipantLogMigration::default();
    for conversation_id in conversation_ids {
        let outcome = migrate_conversation(store, config, limits, conversation_id)?;
        match &outcome {
            ConversationLogOutcome::Empty => tracing::info!(
                conversation_id,
                "participant log migration: conversation has no rows; nothing to rewrite"
            ),
            ConversationLogOutcome::Current { generation, rows } => tracing::info!(
                conversation_id,
                generation,
                rows,
                "participant log migration: conversation already holds only current rows"
            ),
            ConversationLogOutcome::Migrated {
                from_generation,
                to_generation,
                rows,
                converted_rows,
                bytes,
            } => tracing::info!(
                conversation_id,
                from_generation,
                to_generation,
                rows,
                converted_rows,
                bytes,
                "participant log migration: conversation rewritten and switched to its new generation"
            ),
        }
        migration.conversations.push(ConversationLogMigration {
            conversation_id,
            outcome,
        });
    }
    Ok(migration)
}

fn migrate_conversation(
    store: &Arc<dyn DurableStore>,
    config: &ParticipantConfig,
    limits: ConversationOutboxLimits,
    conversation_id: u64,
) -> Result<ConversationLogOutcome, ParticipantLogMigrationError> {
    let write_error = |generation: u64, reason: String| ParticipantLogMigrationError::Write {
        conversation_id,
        generation,
        reason,
    };
    let source = OperationLog::new(Arc::clone(store), conversation_id);
    let source_generation = block_on(source.generation())
        .map_err(|error| write_error(0, error.to_string()))?
        .map_err(|error| write_error(0, error.to_string()))?;
    let source = OperationLog::at_generation(Arc::clone(store), conversation_id, source_generation);
    let mut rewrite = LogRewrite::default();
    let source_state = replay(store, config, limits, &source, Some(&mut rewrite))?;
    let rows = u64::try_from(rewrite.rows.len())
        .map_err(|_| write_error(source_generation, "row count exceeds u64".to_owned()))?;
    if rows == 0 {
        return Ok(ConversationLogOutcome::Empty);
    }
    if rewrite.legacy_rows == 0 {
        return Ok(ConversationLogOutcome::Current {
            generation: source_generation,
            rows,
        });
    }
    let generation = source_generation
        .checked_add(1)
        .ok_or_else(|| write_error(source_generation, "generation overflow".to_owned()))?;
    let target = OperationLog::at_generation(Arc::clone(store), conversation_id, generation);
    let bytes = write_generation(&target, &rewrite.rows)?;
    let target_state = replay(store, config, limits, &target, None)?;
    if format!("{source_state:?}") != format!("{target_state:?}") {
        return Err(ParticipantLogMigrationError::StateDrift {
            conversation_id,
            source_generation,
            generation,
        });
    }
    let switched =
        block_on(source.switch_generation(source_generation, generation)).map_err(|error| {
            ParticipantLogMigrationError::SwitchUnconfirmed {
                conversation_id,
                expected: source_generation,
                to: generation,
                reason: error.to_string(),
            }
        })?;
    switched.map_err(|failure| match failure {
        GenerationSwitchFailure::NotSwapped(error) => ParticipantLogMigrationError::Switch {
            conversation_id,
            expected: source_generation,
            to: generation,
            reason: error.to_string(),
        },
        GenerationSwitchFailure::SwappedUnconfirmed(error) => {
            ParticipantLogMigrationError::SwitchUnconfirmed {
                conversation_id,
                expected: source_generation,
                to: generation,
                reason: error.to_string(),
            }
        }
    })?;
    Ok(ConversationLogOutcome::Migrated {
        from_generation: source_generation,
        to_generation: generation,
        rows,
        converted_rows: rewrite.converted_rows,
        bytes,
    })
}

fn replay(
    store: &Arc<dyn DurableStore>,
    config: &ParticipantConfig,
    limits: ConversationOutboxLimits,
    log: &OperationLog,
    rewrite: Option<&mut LogRewrite>,
) -> Result<ConversationAuthority, ParticipantLogMigrationError> {
    let conversation_id = log.conversation_id();
    let refused = |reason: String| ParticipantLogMigrationError::Replay {
        conversation_id,
        reason,
    };
    let outbox_log = OutboxLog::new(Arc::clone(store), conversation_id);
    block_on(ConversationAuthority::replay_rewriting(
        conversation_id,
        log,
        &outbox_log,
        config,
        limits,
        rewrite,
    ))
    .map_err(|error| refused(error.to_string()))?
    .map_err(|error| {
        refused(match error {
            RestoreError::Extension(error) => error.to_string(),
            RestoreError::Semantic(error) => error.to_string(),
        })
    })
}

/// Writes `rows` as the whole of `target`'s generation, completing a
/// generation an interrupted run left part-written after checking every row it
/// already holds. Returns the generation's encoded size.
fn write_generation(
    target: &OperationLog,
    rows: &[Vec<u8>],
) -> Result<u64, ParticipantLogMigrationError> {
    let conversation_id = target.conversation_id();
    let generation = block_on(target.generation())
        .map_err(|error| error.to_string())
        .and_then(|read| read.map_err(|error| error.to_string()))
        .map_err(|reason| ParticipantLogMigrationError::Write {
            conversation_id,
            generation: 0,
            reason,
        })?;
    let write_error = |reason: String| ParticipantLogMigrationError::Write {
        conversation_id,
        generation,
        reason,
    };
    let mut present = 0_u64;
    let mut expected = rows.iter();
    loop {
        let page = block_on(target.read_raw_page(present))
            .map_err(|error| write_error(error.to_string()))?
            .map_err(|error| write_error(error.to_string()))?;
        let short = page.len() < super::log::READ_BATCH_SIZE;
        for entry in page {
            let matches = expected.next().is_some_and(|row| *row == entry.payload);
            if entry.sequence != present || !matches {
                return Err(ParticipantLogMigrationError::GenerationDrift {
                    conversation_id,
                    generation,
                    sequence: entry.sequence,
                });
            }
            present = present
                .checked_add(1)
                .ok_or_else(|| write_error("row count overflow".to_owned()))?;
        }
        if short {
            break;
        }
    }
    if present > 0 {
        tracing::warn!(
            conversation_id,
            generation,
            present,
            "participant log migration: completing a generation an interrupted run left part-written"
        );
    }
    let remaining: Vec<Vec<u8>> = expected.cloned().collect();
    let mut next = present;
    for chunk in remaining.chunks(WRITE_BATCH_ROWS) {
        next = block_on(target.append_encoded(chunk.to_vec(), next))
            .map_err(|error| write_error(error.to_string()))?
            .map_err(|error| write_error(error.to_string()))?;
    }
    rows.iter().try_fold(0_u64, |total, row| {
        u64::try_from(row.len())
            .ok()
            .and_then(|len| total.checked_add(len))
            .ok_or_else(|| write_error("generation size exceeds u64".to_owned()))
    })
}