pipecircuit 0.0.20

#![cfg(feature = "storage")]
use core::num::{NonZeroU32, NonZeroU64};
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 debug(error: impl core::fmt::Debug) -> String {
    format!("{error:?}")
}
type Request = ContinuousRequest<(), u32, ()>;
type Reply = ContinuousReply<(), (), (), u32, u32>;

#[test]
fn consumer_deltas_and_receipt_tracking_add_no_steady_state_allocations() -> Result<(), String> {
    let capacity = Capacity::new(NonZeroU32::MIN);
    let authority = Authority::recover(
        Recovery {
            history: HistoryId([7; 16]),
            revision: Revision(0),
            reserved_through: 0,
            rows: vec![(1, Stock { quantity: 0 })],
            outcomes: vec![],
        },
        1024,
        Limits {
            rows: capacity,
            operations: capacity,
            sessions: capacity,
            changes: 1,
            row_bytes: 4,
            receipt_bytes: 0,
        },
        Replication::Disabled,
    )
    .map_err(debug)?;
    let mut subscriptions = Subscriptions::new(SubscriptionLimits {
        views: capacity,
        subscribers: capacity,
        retained_changes: capacity,
    })
    .map_err(debug)?;
    let subscription = subscriptions
        .subscribe(&authority, Quantities, Guarantee::Accepted)
        .map_err(debug)?;
    let id = RequestId(NonZeroU64::MIN);
    let mut copy = SubscribedView::new(ContinuousView::new(capacity));
    let SubscriptionStart {
        session,
        request: _request,
    }: SubscriptionStart<(), u32, ()> =
        copy.subscribe(id, (), Guarantee::Accepted).map_err(debug)?;
    let subscribed: Reply = Reply::Subscribed { id, subscription };
    copy.step(ReceiveSubscription {
        session,
        reply: &subscribed,
    })
    .map_err(debug)?;
    let stamp = |revision| ViewRevision {
        source: SourceId::new(NonZeroU64::MIN),
        view: ViewId::new(NonZeroU64::MIN),
        epoch: 1,
        revision,
    };
    let baseline: Reply = Reply::View {
        subscription,
        update: ViewUpdate::Baseline {
            through: stamp(1),
            rows: vec![ViewChange::Upsert {
                id: 1,
                incarnation: 1,
                value: 0,
            }],
        },
    };
    copy.step(ReceiveSubscription {
        session,
        reply: &baseline,
    })
    .map_err(debug)?;
    // Owned wire publications are deliberately built before measurement. This
    // test measures consumer application, not codec/transport capture costs.
    let updates: Vec<Reply> = (1_u32..=128)
        .map(|value| {
            let base = u64::from(value);
            Reply::View {
                subscription,
                update: ViewUpdate::Changes {
                    base: stamp(base),
                    through: stamp(base.saturating_add(1)),
                    changes: vec![ViewChange::Upsert {
                        id: 1,
                        incarnation: 1,
                        value,
                    }],
                },
            }
        })
        .collect();
    let profiler = dhat::Profiler::builder().testing().build();
    for (index, update) in updates.iter().enumerate() {
        assert_eq!(
            copy.step(ReceiveSubscription {
                session,
                reply: update
            })
            .map_err(debug)?,
            SubscriptionProgress::Updated(ViewApplyOutcome::Applied)
        );
        let receipt = Receipt::from_operation(OperationId {
            history: HistoryId([7; 16]),
            epoch: 1,
            session: SessionId(1),
            sequence: u64::try_from(index)
                .map_err(debug)?
                .checked_add(1)
                .ok_or("sequence overflow")?,
        });
        let mut pending = PendingMutation::new(TableRequest::<(), _>::Mutate {
            receipt,
            command: 5_u32,
            guarantee: Guarantee::LocallyDurable,
        })
        .map_err(debug)?;
        let _request: Request = pending.retry(id).map_err(debug)?;
        assert_eq!(pending.disconnect(), MutationStatus::Uncertain);
        let _inquiry: Request = pending.inquire(id).map_err(debug)?;
        let reply = Ok(TableReply::<(), (), ()>::Mutation(
            MutationAnswer::Confirmed {
                receipt,
                outcome: Ok(()),
                established: Guarantee::LocallyDurable,
            },
        ));
        assert_eq!(
            pending
                .step(ReceiveMutation { id, reply: &reply })
                .map_err(debug)?,
            MutationStatus::Confirmed
        );
    }
    let stats = dhat::HeapStats::get();
    drop(profiler);
    assert_eq!(copy.view().get(1), Some(&128));
    assert_eq!(stats.total_blocks, 0);
    assert_eq!(stats.total_bytes, 0);
    Ok(())
}