extern crate std;
use super::*;
use crate::storage::query_workspace::QueryWorkspaceError;
use crate::storage::warp_table::{Capacity, WarpTableInsertWithIdOutcome, WarpTableRow};
use alloc::boxed::Box;
use core::num::NonZeroU32;
#[derive(WarpTableRow)]
#[index(primary = u32)]
struct Task {
#[index(secondary)]
status: u8,
#[index(secondary)]
assignee: u8,
}
fn tasks(capacity: u32) -> Result<WarpTable<Task>, &'static str> {
let mut tasks = WarpTable::try_new(Capacity::new(NonZeroU32::new(capacity).ok_or("capacity")?))
.map_err(|_| "table")?;
for id in 0..capacity {
assert!(matches!(
tasks.insert_with_id(
id,
Task {
status: (id % 3) as u8,
assignee: (id % 2) as u8,
}
),
WarpTableInsertWithIdOutcome::Inserted { .. }
));
}
Ok(tasks)
}
fn storage_identity(
tasks: &WarpTable<Task>,
) -> Result<(*const AllocatedBulkUpdateWorkspace, *const u16), &'static str> {
let workspace = tasks
.bulk_workspace
.as_deref()
.ok_or("reserved workspace")?;
Ok((core::ptr::from_ref(workspace), workspace.slots.as_ptr()))
}
#[test]
fn allocated_calls_reuse_storage_and_keep_the_original_selection() -> Result<(), &'static str> {
for capacity in [1, 64, 65, 65535, 65536, 65537, 100_000] {
let mut tasks = tasks(capacity)?;
assert!(tasks.bulk_workspace.is_none());
let selected = tasks.rows().filter(|row| *row.assignee == 0).count() as u64;
let changed = tasks
.rows()
.filter(|row| *row.assignee == 0 && *row.status != 1)
.count() as u64;
let outcome = tasks
.bulk_update(
|query| query.find_assignee(0),
|mut selected| {
let outcome = selected.set_status(1)?;
let _assignee = selected.set_assignee(1)?;
assert!(matches!(
selected.set_status(1)?,
BulkSetOutcome::Unchanged { .. }
));
let _final = selected.set_status(2)?;
Ok(outcome)
},
)
.map_err(|_| "bulk update")?;
assert_eq!(outcome, BulkSetOutcome::Updated { selected, changed });
let workspace = tasks.bulk_workspace.as_mut().ok_or("reserved workspace")?;
assert_eq!(
workspace
.query
.membership()
.map_err(|_| "membership")?
.capacity(),
capacity as usize
);
assert_eq!(workspace.slots.len(), (capacity as usize).min(65536));
let identity = storage_identity(&tasks)?;
tasks.reserve_bulk_updates().map_err(|_| "reservation")?;
let outcome = tasks
.bulk_update(
|query| {
query
.find_status(2)?
.and(|query| query.find_assignee(1))?
.or(|query| query.find_status(99))?
.and_not(|query| query.find_assignee(99))?
.xor(|query| query.find_status(99))
},
|mut selected| selected.set_status(2),
)
.map_err(|_| "algebra")?;
assert_eq!(
outcome,
BulkSetOutcome::Unchanged {
selected: tasks
.rows()
.filter(|row| *row.status == 2 && *row.assignee == 1)
.count() as u64
}
);
assert_eq!(identity, storage_identity(&tasks)?);
for row in tasks.rows() {
assert_eq!(
*row.status,
if *row.id % 2 == 0 {
2
} else {
(*row.id % 3) as u8
}
);
assert_eq!(*row.assignee, 1);
}
}
Ok(())
}
#[test]
fn errors_and_unwinding_return_reserved_storage() -> Result<(), &'static str> {
let mut tasks = tasks(128)?;
tasks.reserve_bulk_updates().map_err(|_| "reserve")?;
let identity = storage_identity(&tasks)?;
let error = tasks.bulk_update(
|query| {
let _first = query.find_status(0)?;
query.find_status(1)
},
|mut selected| selected.set_status(2),
);
assert!(matches!(
error,
Err(AllocatedBulkUpdateError::Update(
BulkUpdateError::Selection(WorkspaceSelectionError::SelectionAlreadyStarted)
))
));
assert_eq!(tasks.query_memberships().find_status(2).len(), 42);
let error = tasks.bulk_update(
|query| query.find_assignee(0),
|mut selected| {
let outcome = selected.set_status(7)?;
assert_eq!(
outcome,
BulkSetOutcome::Updated {
selected: 64,
changed: 64
}
);
Err::<(), _>(BulkUpdateError::InsufficientStorage {
required: 8,
available: 0,
})
},
);
assert!(matches!(
error,
Err(AllocatedBulkUpdateError::Update(
BulkUpdateError::InsufficientStorage {
required: 8,
available: 0
}
))
));
assert_eq!(tasks.query_memberships().find_status(7).len(), 64);
for during_selection in [true, false] {
let unwind = std::panic::catch_unwind(core::panic::AssertUnwindSafe(|| {
tasks.bulk_update(
|query| {
if during_selection {
std::panic::resume_unwind(Box::new("selection"));
}
query.find_assignee(0)
},
|mut selected| {
let _outcome = selected.set_status(8)?;
std::panic::resume_unwind(Box::new("mutation"))
},
) as Result<(), AllocatedBulkUpdateError>
}));
assert!(unwind.is_err());
assert_eq!(identity, storage_identity(&tasks)?);
}
assert_eq!(
tasks
.bulk_update(|q| q.find_status(8), |mut selected| selected.set_status(9))
.map_err(|_| "reuse")?,
BulkSetOutcome::Updated {
selected: 64,
changed: 64
}
);
assert_eq!(tasks.query_memberships().find_status(9).len(), 64);
assert_eq!(identity, storage_identity(&tasks)?);
Ok(())
}
#[test]
fn storage_refusals_retain_the_exact_cause() -> Result<(), &'static str> {
#[cfg(target_pointer_width = "64")]
assert!(matches!(
AllocatedBulkUpdateWorkspace::try_new(usize::MAX),
Err(AllocatedBulkUpdateError::Update(
BulkUpdateError::Selection(WorkspaceSelectionError::Workspace(
QueryWorkspaceError::CapacityOutsideSlotUniverse { slots: usize::MAX }
))
))
));
let error = Vec::<u16>::new()
.try_reserve_exact(usize::MAX)
.err()
.ok_or("allocation succeeded")?;
let message = alloc::format!("{error}");
let AllocatedBulkUpdateError::Reservation(error) = AllocatedBulkUpdateError::from(error) else {
return Err("cause changed");
};
assert_eq!(alloc::format!("{error}"), message);
Ok(())
}