use super::*;
use crate::storage::warp_table::WarpTable;
use alloc::vec::Vec;
#[derive(Clone, Debug, Eq, PartialEq, WarpTableRow)]
#[index(primary=u32)]
struct Keys {
#[index(secondary, range)]
room: u32,
#[index(unique)]
code: u32,
}
impl PersistentRow for Keys {
fn own(row: KeysRef<'_>) -> Self {
Self {
room: *row.room,
code: *row.code,
}
}
fn retained_bytes(&self) -> usize {
8
}
}
fn located(room: u32, code: u32, offset: u64) -> Located<Keys> {
Located {
keys: Keys { room, code },
location: RecordLocation {
generation: 1,
offset,
bytes: 64,
},
}
}
#[test]
fn index_membership_and_atomic_unique_swaps_do_not_require_record_payloads() -> Result<(), Error> {
let mut limits = limits();
limits.row_bytes = 64;
let mut authority = Authority::<Located<Keys>>::recover(
Recovery {
history: HistoryId([1; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![(1, located(3, 10, 0)), (2, located(4, 20, 64))],
outcomes: vec![],
},
1024,
limits,
Replication::Disabled,
)?;
let changes = vec![
Change::Replace {
id: 1,
row: located(4, 20, 128),
},
Change::Replace {
id: 2,
row: located(3, 10, 192),
},
];
authority.validate_changes(&changes)?;
let inspect =
|authority: &Authority<Located<Keys>>, guarantee: Guarantee| -> Result<Vec<u32>, Error> {
match authority.read(ReadRequest { guarantee }) {
ReadOutcome::Ready { table, .. } => Ok(IndexView::new(table)
.select(|query| query.find_room(3))
.ids()
.collect()),
ReadOutcome::Pending => Err(Error::Busy),
ReadOutcome::Refused(error) => Err(error),
}
};
assert_eq!(inspect(&authority, Guarantee::Accepted)?, [1]);
let session = authority.open_session()?;
authority
.step(Submit {
receipt: None,
operation: authority.operation(session, 1),
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes,
})
.map_err(|(_request, error)| error)?;
assert_eq!(inspect(&authority, Guarantee::Accepted)?, [2]);
assert_eq!(inspect(&authority, Guarantee::LocallyDurable)?, [1]);
let attempt = authority
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
authority.step(Complete {
attempt,
result: PhysicalResult::Durable,
})?;
assert_eq!(inspect(&authority, Guarantee::LocallyDurable)?, [2]);
assert_eq!(
authority.validate_changes(&[Change::Replace {
id: 1,
row: located(3, 10, 256)
}]),
Err(Error::UniqueConflict)
);
assert_eq!(inspect(&authority, Guarantee::Accepted)?, [2]);
Ok(())
}
#[test]
fn bounded_target_and_change_sets_remember_a_discarded_refusal() -> Result<(), Error> {
let mut targets = Targets::new(2)?;
targets.push(1)?;
targets.push(1)?;
targets.push(2)?;
assert_eq!(targets.push(3), Err(Error::Full));
targets.push(1)?;
assert_eq!(targets.finish(), Err(Error::Full));
let mut changes = ChangeSet::<Keys>::new(1)?;
assert!(changes.push(Change::Remove { id: 1 }).is_ok());
assert!(changes.push(Change::Remove { id: 2 }).is_err());
assert!(matches!(changes.finish(), Err(Error::Full)));
assert!(matches!(
ChangeSet::<Keys>::new(0),
Err(Error::InvalidBounds)
));
assert!(matches!(Targets::<u32>::new(0), Err(Error::InvalidBounds)));
Ok(())
}
#[test]
fn located_columns_keep_keys_locations_and_indexes_aligned() -> Result<(), Error> {
let mut table =
WarpTable::<Located<Keys>>::try_new(limits().rows).map_err(|_error| Error::Allocation)?;
assert!(table.query().is_empty());
assert_eq!(table.query().len(), 0);
assert!(table.query().get(1).is_none());
super::super::changes::insert(&mut table, 1, located(3, 10, 0))?;
super::super::changes::insert(&mut table, 2, located(4, 20, 64))?;
assert!(!table.query().is_empty());
assert_eq!(table.query().len(), 2);
assert_eq!(table.query().ids(), [1, 2]);
assert_eq!(*table.query().get(1).ok_or(Error::Absent)?.code, 10);
assert!(matches!(
table.replace(1, located(4, 20, 128)),
crate::storage::warp_table::WarpTableReplaceOutcome::UniqueConflict { .. }
));
assert!(matches!(
table.replace(1, located(9, 30, 128)),
crate::storage::warp_table::WarpTableReplaceOutcome::Replaced { .. }
));
assert!(
table
.query()
.select(|query| query.find_room(3))
.ids()
.next()
.is_none()
);
assert_eq!(
table
.query()
.select(|query| query.find_room(9))
.ids()
.collect::<Vec<_>>(),
[1]
);
match table.remove(1) {
crate::storage::warp_table::WarpTableRemoveOutcome::Removed { removed, .. } => {
assert_eq!(removed.location.offset, 128)
}
crate::storage::warp_table::WarpTableRemoveOutcome::Absent { .. } => {
return Err(Error::Absent);
}
}
match table.get(2) {
WarpTableGetOutcome::Present { row } => {
assert_eq!(*row.keys.room, 4);
assert_eq!(row.location.offset, 64);
}
WarpTableGetOutcome::Absent { .. } => return Err(Error::Absent),
}
let ids: Vec<_> = table
.query()
.select(|query| query.find_room(4))
.ids()
.collect();
assert_eq!(ids, [2]);
Ok(())
}
#[test]
fn resident_byte_admission_counts_record_locations_as_well_as_keys() {
let mut limits = limits();
limits.row_bytes = 8;
assert!(matches!(
Authority::<Located<Keys>>::recover(
Recovery {
history: HistoryId([1; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![(1, located(3, 10, 0))],
outcomes: vec![]
},
1024,
limits,
Replication::Disabled
),
Err(Error::InvalidBounds)
));
}