pipecircuit 0.0.23

extern crate std;

use super::*;
use alloc::{string::String, vec};
use core::num::{NonZeroU32, NonZeroU64};
use core::panic::AssertUnwindSafe;
use std::panic::catch_unwind;

fn fresh() -> Result<(ViewReplica<u32, u32>, ViewRevision), &'static str> {
    let source = SourceId::new(NonZeroU64::MIN);
    let view = ViewId::new(NonZeroU64::MIN);
    let replica = ViewReplica::try_new(source, view, Capacity::new(NonZeroU32::MIN))
        .map_err(|_| "replica storage unavailable")?;
    Ok((
        replica,
        ViewRevision {
            source,
            view,
            epoch: 1,
            revision: 0,
        },
    ))
}

fn assert_invariant(run: impl FnOnce(), reason: &str) -> Result<(), &'static str> {
    let Err(payload) = catch_unwind(AssertUnwindSafe(run)) else {
        return Err("corrupt state did not stop publication");
    };
    let message = payload
        .downcast_ref::<String>()
        .ok_or("unexpected invariant payload")?;
    assert_eq!(
        message,
        &std::format!("WarpTable invariant violated: {reason:?}")
    );
    Ok(())
}

#[test]
fn corrupted_staging_bounds_refuse_before_visible_publication() -> Result<(), &'static str> {
    let (mut replica, through) = fresh()?;
    let changes = [ViewChange::Upsert {
        id: 7,
        incarnation: 1,
        value: 42,
    }];
    replica.staging_rows.maximum_rows = 0;
    assert_eq!(
        replica.begin_baseline(through, 1),
        Ok(ViewApplyOutcome::BaselineRequired)
    );
    assert_eq!(
        replica.baseline_chunk(through, &changes),
        Err(ViewApplyError::CapacityExceeded)
    );
    assert!(replica.is_empty());
    assert!(replica.staging_rows.is_empty());
    assert!(matches!(replica.staging, Staging::Empty));

    replica.seen.maximum_rows = 0;
    assert_eq!(
        replica.stage_changes(&changes),
        Err(ViewApplyError::CapacityExceeded)
    );
    assert!(replica.is_empty());
    assert!(replica.pending.is_empty());
    Ok(())
}

#[test]
fn corrupted_prepared_baseline_cannot_publish_stale_rows() -> Result<(), &'static str> {
    let (mut replica, through) = fresh()?;
    assert_eq!(
        replica.begin_baseline(through, 0),
        Ok(ViewApplyOutcome::BaselineRequired)
    );
    let ViewBaselinePreparation::Ready(prepared) = replica
        .prepare_baseline(through)
        .map_err(|_| "baseline refused")?
    else {
        return Err("baseline unexpectedly stale");
    };
    prepared.replica.sync.require_snapshot_through((1, 1));
    assert_invariant(
        || {
            prepared.publish();
        },
        "prepared baseline became stale under exclusive guard",
    )?;
    assert!(replica.is_empty());
    assert!(matches!(replica.staging, Staging::Empty));
    Ok(())
}

#[test]
fn corrupted_prepared_delta_cannot_cross_a_revision_gap() -> Result<(), &'static str> {
    let (mut replica, base) = fresh()?;
    replica
        .begin_baseline(base, 0)
        .map_err(|_| "baseline refused")?;
    assert_eq!(replica.finish_baseline(base), Ok(ViewApplyOutcome::Applied));
    let through = ViewRevision {
        revision: 1,
        ..base
    };
    let changes = [ViewChange::Upsert {
        id: 7,
        incarnation: 1,
        value: 42,
    }];
    let ViewPreparation::Ready(prepared) = replica
        .prepare_changes(base, through, &changes)
        .map_err(|_| "changes refused")?
    else {
        return Err("changes unexpectedly need recovery");
    };
    prepared.replica.sync.require_snapshot_through((1, 2));
    assert_invariant(
        || {
            prepared.publish();
        },
        "prepared changes lost continuity under exclusive guard",
    )?;
    assert!(replica.is_empty());
    assert!(replica.pending.is_empty());
    assert_eq!(
        replica.status(),
        ViewReplicaStatus::BaselineRequired {
            through: base,
            required: ViewRevision {
                revision: 2,
                ..base
            },
        }
    );
    Ok(())
}

#[test]
fn corrupted_commit_inputs_preserve_exact_invariant_diagnostics() -> Result<(), &'static str> {
    let (mut replica, _) = fresh()?;
    let mut pending = vec![ViewChange::Remove {
        id: 7,
        incarnation: 1,
    }];
    assert_invariant(
        || commit(&mut replica.rows, &mut pending),
        "validated view removal absent",
    )?;
    assert!(replica.is_empty());

    replica.rows.maximum_rows = 0;
    let mut pending = vec![ViewChange::Upsert {
        id: 7,
        incarnation: 1,
        value: 42,
    }];
    assert_invariant(
        || commit(&mut replica.rows, &mut pending),
        "validated view capacity exceeded",
    )?;
    assert!(replica.is_empty());
    Ok(())
}