pipecircuit 0.0.21

#![cfg(feature = "storage")]

use core::num::{NonZeroU32, NonZeroU64};
use pipecircuit::storage::warp_table::observed::{
    SourceId, ViewApplyOutcome, ViewDefinition, ViewDelivery, ViewId, ViewPublication, ViewReplica,
};
use pipecircuit::storage::warp_table::{
    Capacity, WarpTable, WarpTableInsertWithIdOutcome, WarpTableRow, WarpTableUpdateOutcome,
};

#[global_allocator]
static ALLOCATOR: dhat::Alloc = dhat::Alloc;

#[derive(WarpTableRow)]
#[index(primary = u32)]
struct Position {
    x: u32,
}
struct Positions;
impl ViewDefinition<Position> for Positions {
    type Value = u32;
    fn project(&self, row: PositionRef<'_>) -> Option<u32> {
        Some(*row.x)
    }
}

#[test]
fn fixed_payload_incremental_publication_and_application_do_not_allocate()
-> Result<(), &'static str> {
    for count in [1_u32, 32, 1024] {
        let capacity = Capacity::new(NonZeroU32::new(count).ok_or("capacity")?);
        let table = WarpTable::<Position>::try_new(capacity).map_err(|_| "table")?;
        let source_id = SourceId::new(NonZeroU64::MIN);
        let view_id = ViewId::new(NonZeroU64::MIN);
        let mut source = table.observe(source_id).map_err(|_| "observation")?;
        for id in 0..count {
            let outcome = source
                .insert_with_id(id, Position { x: 0 })
                .map_err(|_| "insert sequence")?;
            assert!(matches!(
                outcome,
                WarpTableInsertWithIdOutcome::Inserted { .. }
            ));
        }
        let mut view = source
            .maintain_with_retention(
                view_id,
                Positions,
                Capacity::new(
                    NonZeroU32::new(count.checked_mul(4).ok_or("history overflow")?)
                        .ok_or("history")?,
                ),
            )
            .map_err(|_| "view")?;
        let mut replica =
            ViewReplica::try_new(source_id, view_id, capacity).map_err(|_| "replica")?;
        let ViewPublication::Baseline { through, rows } = view.publication() else {
            return Err("baseline");
        };
        assert_eq!(
            replica.begin_baseline(through, rows.len()),
            Ok(ViewApplyOutcome::BaselineRequired)
        );
        replica
            .baseline_chunk(through, rows)
            .map_err(|_| "baseline rows")?;
        assert_eq!(
            replica.finish_baseline(through),
            Ok(ViewApplyOutcome::Applied)
        );
        let mut cursor = through;
        let mut buffer = view.delivery_buffer().map_err(|_| "delivery buffer")?;
        let profiler = dhat::Profiler::builder().testing().build();
        for frame in 1..65_u32 {
            for id in 0..count {
                let outcome = source
                    .update(id, |mut row| *row.x_mut() = frame)
                    .map_err(|_| "update sequence")?;
                assert!(matches!(outcome, WarpTableUpdateOutcome::Updated { .. }));
            }
            view.refresh(&source).map_err(|_| "refresh")?;
            if frame % 4 == 0 {
                let ViewDelivery::Changes {
                    base,
                    through,
                    changes,
                } = view
                    .delivery(Some(cursor), &mut buffer)
                    .map_err(|_| "delivery")?
                else {
                    return Err("coalesced changes");
                };
                assert_eq!(
                    replica.apply_changes(base, through, changes),
                    Ok(ViewApplyOutcome::Applied)
                );
                cursor = through;
            }
        }
        let stats = dhat::HeapStats::get();
        drop(profiler);
        assert_eq!(stats.total_blocks, 0, "rows={count}");
        assert_eq!(stats.total_bytes, 0, "rows={count}");
        assert!(replica.rows().all(|row| *row.value == 64));
    }
    Ok(())
}