#![cfg(feature = "storage")]
use core::num::NonZeroU32;
use pipecircuit::{
StateMachine,
storage::warp_table::{Capacity, WarpTableRow, 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
}
}
#[test]
fn fixed_row_semantic_turns_reuse_storage_without_allocation() -> Result<(), Error> {
for rows in [1, 32, 1024] {
let limits = Limits {
receipt_bytes: 1024,
rows: Capacity::new(NonZeroU32::new(rows).ok_or(Error::InvalidBounds)?),
operations: Capacity::new(NonZeroU32::MIN),
sessions: Capacity::new(NonZeroU32::MIN),
changes: 1,
row_bytes: 4,
};
let recovery = Recovery {
history: HistoryId([1; 16]),
revision: Revision(0),
reserved_through: 0,
rows: (0..rows).map(|id| (id, Stock { quantity: 0 })).collect(),
outcomes: vec![],
};
let mut authority = Authority::recover(recovery, 1024, limits, Replication::Required)?;
let session = authority.open_session()?;
let mut changes = Vec::with_capacity(1);
let profiler = dhat::Profiler::builder().testing().build();
for sequence in 1..=128 {
changes.push(Change::Replace {
id: 0,
row: Stock {
quantity: sequence as u32,
},
});
let operation = authority.operation(session, sequence);
authority
.step(Submit {
receipt: None,
operation,
expected: authority.revision(),
guarantee: Guarantee::Accepted,
changes,
})
.map_err(|(_, error)| error)?;
for guarantee in [Guarantee::LocallyDurable, Guarantee::ReplicatedDurable] {
let attempt = authority.prepare(guarantee)?.ok_or(Error::Busy)?;
authority.step(Complete {
attempt,
result: PhysicalResult::Durable,
})?;
}
changes = authority.retire(operation)?.changes;
changes.clear();
}
let stats = dhat::HeapStats::get();
drop(profiler);
assert_eq!(stats.total_blocks, 0, "rows={rows}");
assert_eq!(stats.total_bytes, 0, "rows={rows}");
}
Ok(())
}