pipecircuit 0.0.21

#![cfg(feature = "storage")]
use core::num::NonZeroU32;
use pipecircuit::{
    StateMachine,
    storage::warp_table::{Capacity, WarpTableRow, observed::*, persistent::*},
};

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

#[derive(Clone, Eq, PartialEq, WarpTableRow)]
#[index(primary = u32)]
struct Stock {
    quantity: u32,
}
impl PersistentRow for Stock {
    fn own(row: StockRef<'_>) -> Self {
        Self {
            quantity: *row.quantity,
        }
    }
    fn retained_bytes(&self) -> usize {
        4
    }
}
#[derive(Clone, Eq, PartialEq)]
struct Quantities;
impl ViewDefinition<Stock> for Quantities {
    type Value = u32;
    fn project(&self, row: StockRef<'_>) -> Option<u32> {
        Some(*row.quantity)
    }
}
fn capacity(value: u32) -> Result<Capacity, Error> {
    Ok(Capacity::new(
        NonZeroU32::new(value).ok_or(Error::InvalidBounds)?,
    ))
}

#[test]
fn borrowed_continuous_fanout_reuses_storage_even_after_subscriber_history_overflow()
-> Result<(), String> {
    fn debug(error: impl core::fmt::Debug) -> String {
        format!("{error:?}")
    }
    for (rows, subscribers) in [(4, 1), (4, 16), (1024, 1), (1024, 16)] {
        let limits = Limits {
            rows: capacity(rows).map_err(debug)?,
            operations: capacity(1).map_err(debug)?,
            sessions: capacity(1).map_err(debug)?,
            changes: 1,
            row_bytes: 4,
            receipt_bytes: 0,
        };
        let mut owner = Authority::recover(
            Recovery {
                history: HistoryId([7; 16]),
                revision: Revision(0),
                reserved_through: 0,
                rows: (0..rows).map(|id| (id, Stock { quantity: 0 })).collect(),
                outcomes: vec![],
            },
            1024,
            limits,
            Replication::Disabled,
        )
        .map_err(debug)?;
        let session = owner.open_session().map_err(debug)?;
        let mut views = Subscriptions::new(SubscriptionLimits {
            views: capacity(1).map_err(debug)?,
            subscribers: capacity(subscribers).map_err(debug)?,
            retained_changes: capacity(1).map_err(debug)?,
        })
        .map_err(debug)?;
        let ids: Vec<_> = (0..subscribers)
            .map(|_| views.subscribe(&owner, Quantities, Guarantee::Accepted))
            .collect::<Result<_, _>>()
            .map_err(debug)?;
        views.refresh(&owner).map_err(debug)?;
        for id in &ids {
            views
                .transmit(*id, |_| Ok::<_, String>(ViewAdmission::Admitted))
                .map_err(debug)?
                .ok_or("baseline")?
                .map_err(debug)?;
        }
        let mut changes = Vec::with_capacity(1);
        let profiler = dhat::Profiler::builder().testing().build();
        for turn in 1_u32..=128 {
            changes.push(Change::Replace {
                id: 0,
                row: Stock { quantity: turn },
            });
            let operation = owner.operation(session, u64::from(turn));
            owner
                .step(Submit {
                    receipt: None,
                    operation,
                    expected: owner.revision(),
                    guarantee: Guarantee::Accepted,
                    changes,
                })
                .map_err(|(_, error)| debug(error))?;
            views.refresh(&owner).map_err(debug)?;
            let index = usize::try_from(turn % subscribers).map_err(debug)?;
            let id = *ids.get(index).ok_or("subscriber")?;
            let mut observed = None;
            views
                .transmit(id, |delivery| {
                    match delivery {
                        ViewDelivery::Baseline { rows, .. } => {
                            for row in rows {
                                if row.id == 0 {
                                    observed = Some(*row.value);
                                }
                            }
                        }
                        ViewDelivery::Changes { changes, .. } => {
                            for change in changes {
                                match change {
                                    ViewChange::Upsert { id, value, .. } if *id == 0 => {
                                        observed = Some(*value)
                                    }
                                    ViewChange::Upsert { .. } | ViewChange::Remove { .. } => {
                                        return Err("unexpected change".to_owned());
                                    }
                                }
                            }
                        }
                        ViewDelivery::Current { .. } => return Err("unpublished edit".to_owned()),
                    }
                    Ok::<_, String>(ViewAdmission::Admitted)
                })
                .map_err(debug)?
                .ok_or("publication")?
                .map_err(debug)?;
            assert_eq!(observed, Some(turn));
            let attempt = owner
                .prepare(Guarantee::LocallyDurable)
                .map_err(debug)?
                .ok_or("attempt")?;
            owner
                .step(Complete {
                    attempt,
                    result: PhysicalResult::Durable,
                })
                .map_err(debug)?;
            changes = owner.retire(operation).map_err(debug)?.changes;
            changes.clear();
        }
        let stats = dhat::HeapStats::get();
        drop(profiler);
        assert_eq!(
            stats.total_blocks, 0,
            "rows={rows} subscribers={subscribers}"
        );
        assert_eq!(
            stats.total_bytes, 0,
            "rows={rows} subscribers={subscribers}"
        );
    }
    Ok(())
}