#![cfg(feature = "storage")]
use core::{
cell::Cell,
num::{NonZeroU32, NonZeroU64},
};
extern crate alloc;
use alloc::rc::Rc;
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)
}
}
#[derive(Debug)]
struct Payload {
value: u32,
clones: Rc<Cell<usize>>,
}
impl Clone for Payload {
fn clone(&self) -> Self {
self.clones.set(self.clones.get().saturating_add(1));
Self {
value: self.value,
clones: self.clones.clone(),
}
}
}
impl PartialEq for Payload {
fn eq(&self, other: &Self) -> bool {
self.value == other.value
}
}
impl Eq for Payload {}
fn debug(error: impl core::fmt::Debug) -> String {
format!("{error:?}")
}
#[test]
fn reports_reuse_storage_without_cloning_payloads_beyond_existing_replica_staging()
-> Result<(), String> {
let capacity = Capacity::new(NonZeroU32::MIN.saturating_add(3));
let owner = Authority::<Stock>::recover(
Recovery {
history: HistoryId([1; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![],
outcomes: vec![],
},
16,
Limits {
rows: capacity,
operations: capacity,
sessions: capacity,
changes: 4,
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(&owner, Quantities, Guarantee::Accepted)
.map_err(debug)?;
let mut client = SubscribedView::new(ContinuousView::<u32, Payload>::new(capacity));
let mut report = AppliedViewChanges::try_new(capacity).map_err(debug)?;
let request = RequestId(NonZeroU64::MIN);
let start: SubscriptionStart<(), (), ()> = client
.subscribe(request, (), Guarantee::Accepted)
.map_err(|(_, e)| debug(e))?;
type Reply = ContinuousReply<(), (), (), u32, Payload>;
let subscribed: Reply = Reply::Subscribed {
id: request,
subscription,
};
assert_eq!(
client
.step(
ReceiveSubscription {
session: start.session,
reply: &subscribed
}
.with_changes(&mut report)
)
.map_err(debug)?,
SubscriptionChanges::Subscribed
);
let clones = Rc::new(Cell::new(0));
let stamp = |revision| ViewRevision {
source: SourceId::new(NonZeroU64::MIN),
view: ViewId::new(NonZeroU64::MIN),
epoch: 1,
revision,
};
let mut baseline: Reply = Reply::View {
subscription,
update: ViewUpdate::Baseline {
through: stamp(0),
rows: vec![ViewChange::Upsert {
id: 1,
incarnation: 1,
value: Payload {
value: 0,
clones: clones.clone(),
},
}],
},
};
assert_eq!(
client
.step(
ReceiveSubscription {
session: start.session,
reply: &baseline
}
.with_changes(&mut report)
)
.map_err(debug)?,
SubscriptionChanges::Applied {
changes: &[AppliedViewChange::Entered { id: 1 }]
}
);
let mut delta: Reply = Reply::View {
subscription,
update: ViewUpdate::Changes {
base: stamp(0),
through: stamp(1),
changes: vec![ViewChange::Upsert {
id: 1,
incarnation: 1,
value: Payload {
value: 0,
clones: clones.clone(),
},
}],
},
};
clones.set(0);
let profiler = dhat::Profiler::builder().testing().build();
for turn in 1_u32..=128 {
let reply = if turn % 2 == 0 {
&mut baseline
} else {
&mut delta
};
match reply {
Reply::View {
update: ViewUpdate::Baseline { through, rows },
..
} => {
*through = stamp(u64::from(turn));
let Some(ViewChange::Upsert { value, .. }) = rows.first_mut() else {
return Err("baseline row".into());
};
value.value = turn;
}
Reply::View {
update:
ViewUpdate::Changes {
base,
through,
changes,
},
..
} => {
*base = stamp(u64::from(turn).checked_sub(1).ok_or("turn")?);
*through = stamp(u64::from(turn));
let Some(ViewChange::Upsert { value, .. }) = changes.first_mut() else {
return Err("delta row".into());
};
value.value = turn;
}
Reply::Command { .. }
| Reply::Subscribed { .. }
| Reply::Unsubscribed { .. }
| Reply::Resynchronizing { .. }
| Reply::SubscriptionRefused { .. }
| Reply::ReplicationInterrupted { .. } => return Err("publication".into()),
}
assert_eq!(
client
.step(
ReceiveSubscription {
session: start.session,
reply
}
.with_changes(&mut report)
)
.map_err(debug)?,
SubscriptionChanges::Applied {
changes: &[AppliedViewChange::Updated { id: 1 }]
}
);
assert_eq!(client.view().get(1).map(|value| value.value), Some(turn));
}
let stats = dhat::HeapStats::get();
drop(profiler);
assert_eq!(stats.total_blocks, 0);
assert_eq!(stats.total_bytes, 0);
assert_eq!(
clones.get(),
128,
"one staging clone per publication; no report payload clones"
);
Ok(())
}