extern crate std;
use super::*;
use crate::storage::query_workspace::QueryWorkspaceError;
use crate::storage::warp_table::{
Capacity, WarpTableInsertWithIdOutcome, WarpTableRemoveOutcome, WarpTableRow,
};
use alloc::{boxed::Box, vec, vec::Vec};
use core::num::NonZeroU32;
#[derive(WarpTableRow)]
#[index(primary = u32)]
struct Task {
#[index(secondary)]
status: u32,
#[index(secondary)]
assignee: u8,
payload: u32,
}
type Workspace = BulkUpdateWorkspace<
Vec<QueryChunk>,
Box<QueryChunk>,
Box<QueryChunk>,
Box<QueryChunk>,
Vec<u16>,
>;
fn workspace(capacity: usize, slots: usize) -> Result<Workspace, &'static str> {
let chunks = (0..capacity.div_ceil(65536))
.map(|_| QueryChunk::EMPTY)
.collect();
let query =
QueryWorkspace::new(capacity, chunks, Box::new(QueryChunk::EMPTY)).map_err(|_| "query")?;
Ok(BulkUpdateWorkspace::new(
query,
Box::new(QueryChunk::EMPTY),
Box::new(QueryChunk::EMPTY),
vec![0; slots],
))
}
fn table(statuses: impl IntoIterator<Item = u32>) -> Result<WarpTable<Task>, &'static str> {
let values: Vec<_> = statuses.into_iter().collect();
let mut table = WarpTable::try_new(Capacity::new(
NonZeroU32::new(values.len() as u32).ok_or("capacity")?,
))
.map_err(|_| "table")?;
for (id, status) in values.into_iter().enumerate() {
if !matches!(
table.insert_with_id(
id as u32,
Task {
status,
assignee: (id % 2) as u8,
payload: id as u32
}
),
WarpTableInsertWithIdOutcome::Inserted { .. }
) {
return Err("insert");
}
}
Ok(table)
}
fn verify(table: &WarpTable<Task>, model: &[(u32, u8)]) {
for row in table.rows() {
assert_eq!(
Some(&(*row.status, *row.assignee)),
model.get(*row.id as usize)
);
assert_eq!(*row.payload, *row.id);
assert!(
table
.query()
.find_status(*row.status)
.rows()
.any(|indexed| indexed.id == row.id)
);
}
for key in model.iter().map(|(status, _)| *status).chain([999_999]) {
assert_eq!(
table.query_memberships().find_status(key).len(),
table.rows().filter(|r| *r.status == key).count()
);
}
}
#[test]
fn setters_keep_selection_fixed_and_are_idempotent() -> Result<(), &'static str> {
let mut table = table([0, 0, 1, 2, 1, 2])?;
let mut work = workspace(6, 6)?;
let outcomes = table
.bulk_update_in(
&mut work,
|query| query.find_assignee(0),
|mut selected| {
let status = selected.set_status(1)?;
let repeated = selected.set_status(1)?;
let assignee = selected.set_assignee(1)?;
let second_status = selected.set_status(2)?;
Ok((status, repeated, assignee, second_status))
},
)
.map_err(|_| "set")?;
assert_eq!(
outcomes,
(
BulkSetOutcome::Updated {
selected: 3,
changed: 1
},
BulkSetOutcome::Unchanged { selected: 3 },
BulkSetOutcome::Updated {
selected: 3,
changed: 3
},
BulkSetOutcome::Updated {
selected: 3,
changed: 3
},
)
);
verify(&table, &[(2, 1), (0, 1), (2, 1), (2, 1), (2, 1), (2, 1)]);
Ok(())
}
#[test]
fn scratch_refusal_precedes_every_chunk_and_satisfied_rows_need_none() -> Result<(), &'static str> {
let mut table = table((0..100_000).map(|id| u32::from(id < 65536)))?;
let mut work = workspace(100_000, 0)?;
let error = table.bulk_update_in(
&mut work,
|query| query.find_assignee(0),
|mut selected| selected.set_status(1),
);
assert!(matches!(
error,
Err(BulkUpdateError::InsufficientStorage {
required: 17232,
available: 0
})
));
assert_eq!(table.query_memberships().find_status(0).len(), 34464);
let result = table
.bulk_update_in(
&mut work,
|query| query.find_status(1),
|mut selected| selected.set_status(1),
)
.map_err(|_| "noop")?;
assert_eq!(result, BulkSetOutcome::Unchanged { selected: 65536 });
let result = table
.bulk_update_in(
&mut work,
|query| query.find_status(77),
|mut selected| selected.set_status(1),
)
.map_err(|_| "empty")?;
assert_eq!(result, BulkSetOutcome::Unchanged { selected: 0 });
Ok(())
}
#[test]
fn cross_chunk_assignment_tracks_dense_relocation() -> Result<(), &'static str> {
let mut table = table((0..100_000).map(|id| id % 3))?;
assert!(matches!(
table.remove(1),
WarpTableRemoveOutcome::Removed { .. }
));
let mut work = workspace(100_000, 32768)?;
for target in [1, 1, 7, 0] {
let changed = table
.rows()
.filter(|r| *r.assignee == 0 && *r.status != target)
.count() as u64;
let outcome = table
.bulk_update_in(
&mut work,
|query| query.find_assignee(0),
|mut selected| selected.set_status(target),
)
.map_err(|_| "set")?;
assert_eq!(
outcome,
if changed == 0 {
BulkSetOutcome::Unchanged { selected: 50_000 }
} else {
BulkSetOutcome::Updated {
selected: 50_000,
changed,
}
}
);
for row in table.rows() {
assert_eq!(
*row.status,
if *row.assignee == 0 {
target
} else {
*row.id % 3
}
);
assert_eq!(*row.payload, *row.id);
}
assert_eq!(
table.query_memberships().find_status(target).len(),
table.rows().filter(|r| *r.status == target).count()
);
}
Ok(())
}
#[test]
fn source_fragmentation_and_payload_boundaries_match_rows() -> Result<(), &'static str> {
for (rows, keys) in [
(1, 1),
(64, 3),
(65, 1),
(128, 2),
(8192, 8),
(8194, 1),
(10_000, 5000),
(65536, 1),
] {
let mut table = table((0..rows).map(|id| id % keys))?;
let mut work = workspace(rows as usize, 65536)?;
for target in [999_999, 1] {
let selected = table.rows().filter(|r| *r.assignee == 0).count() as u64;
let changed = table
.rows()
.filter(|r| *r.assignee == 0 && *r.status != target)
.count() as u64;
let result = table
.bulk_update_in(
&mut work,
|query| query.find_assignee(0),
|mut selected| selected.set_status(target),
)
.map_err(|_| "set")?;
assert_eq!(result, BulkSetOutcome::Updated { selected, changed });
for row in table.rows() {
assert_eq!(
*row.status,
if *row.assignee == 0 {
target
} else {
*row.id % keys
}
);
}
assert_eq!(
table.query_memberships().find_status(target).len(),
table.rows().filter(|r| *r.status == target).count()
);
}
}
Ok(())
}
#[test]
fn foreign_tables_and_capacity_mismatches_are_refused() -> Result<(), &'static str> {
let foreign: &'static WarpTable<Task> =
alloc::boxed::Box::leak(alloc::boxed::Box::new(table([0])?));
let mut table = table([0])?;
let mut work = workspace(1, 1)?;
assert!(matches!(
table.bulk_update_in(
&mut work,
|query| foreign
.query_memberships()
.find_status(0)
.in_workspace(query.find_status(0)?.into_workspace()),
|mut selected| selected.set_status(1)
),
Err(BulkUpdateError::Selection(
WorkspaceSelectionError::DifferentTables
))
));
let mut work = workspace(2, 1)?;
assert!(matches!(
table.bulk_update_in(
&mut work,
|query| query.find_status(0),
|mut selected| selected.set_status(1)
),
Err(BulkUpdateError::Selection(
WorkspaceSelectionError::Workspace(QueryWorkspaceError::DifferentCapacities {
left: 2,
right: 1
})
))
));
assert_eq!(table.query_memberships().find_status(0).len(), 1);
Ok(())
}
proptest::proptest! {
#![proptest_config(proptest::test_runner::Config::with_cases(48))]
#[test]
fn bulk_sequences_match_independent_row_assignment(
statuses in proptest::collection::vec(0_u32..16,1..400),
operations in proptest::collection::vec((0_u8..2,0_u32..20),1..30),
) {
let mut model:Vec<_>=statuses.iter().enumerate().map(|(id,status)|(*status,(id%2)as u8)).collect();
let mut table=table(statuses).map_err(proptest::test_runner::TestCaseError::fail)?;
let mut work=workspace(model.len(),65536).map_err(proptest::test_runner::TestCaseError::fail)?;
for (assignee,target) in operations {
let selected=model.iter().filter(|(_,a)|*a==assignee).count()as u64;
let changed=model.iter().filter(|(s,a)|*a==assignee&&*s!=target).count()as u64;
let result=table.bulk_update_in(&mut work,
|query|query.find_assignee(assignee),
|mut selected|selected.set_status(target)).map_err(|e|proptest::test_runner::TestCaseError::fail(alloc::format!("{e:?}")))?;
proptest::prop_assert_eq!(result,if changed==0{BulkSetOutcome::Unchanged{selected}}else{BulkSetOutcome::Updated{selected,changed}});
for (status,a) in &mut model { if *a==assignee {*status=target;} }
verify(&table,&model);
}
}
}
#[test]
fn sparse_selected_overlap_uses_bounded_direct_work() -> Result<(), &'static str> {
for prefix in [2, 256] {
let mut table =
table((0..10_000).map(|id| if id < prefix || id % 2 == 1 { 0 } else { id }))?;
let mut work = workspace(10_000, 5000)?;
let outcome = table
.bulk_update_in(
&mut work,
|query| query.find_assignee(0),
|mut selected| selected.set_status(999_999),
)
.map_err(|_| "set")?;
assert_eq!(
outcome,
BulkSetOutcome::Updated {
selected: 5000,
changed: 5000
}
);
for row in table.rows() {
assert_eq!(*row.status, if *row.assignee == 0 { 999_999 } else { 0 });
}
assert_eq!(table.query_memberships().find_status(999_999).len(), 5000);
assert_eq!(table.query_memberships().find_status(0).len(), 5000);
}
Ok(())
}
#[test]
fn satisfied_large_membership_never_resolves_or_accesses_rows() -> Result<(), &'static str> {
use crate::storage::warp_table::secondary_index::{SecondaryIndex, StableSlot, StableSlots};
for (capacity, scratch_slots) in [(128, 0), (128, 128)] {
let mut index = SecondaryIndex::try_new(capacity).map_err(|_| "index")?;
for slot in 0..capacity as u32 {
let _outcome = index
.try_prepare_insert(7_u8, StableSlot::new(slot))
.map_err(|_| "insert")?
.commit();
}
let slots = StableSlots::try_new(capacity).map_err(|_| "slots")?;
let mut work = workspace(capacity, scratch_slots)?;
let mut selected = work.query.membership().map_err(|_| "members")?;
for slot in 0..capacity as u32 {
let _outcome = selected.insert(slot).map_err(|_| "select")?;
}
let mut selection = BulkSelection {
selected: &selected,
pending: work.pending.as_mut(),
group: work.group.as_mut(),
slots: &mut work.slots,
};
assert_eq!(
selection
.set_secondary(&mut index, &mut [], &slots, 7)
.map_err(|_| "no-op")?,
BulkSetOutcome::Unchanged {
selected: capacity as u64
}
);
}
Ok(())
}
#[test]
fn scratch_bounds_only_changes_and_each_setter_preserves_prior_success() -> Result<(), &'static str>
{
let mut table = table((0..200).map(|id| u32::from(id >= 100)))?;
let mut work = workspace(200, 50)?;
let result = table.bulk_update_in(
&mut work,
|query| query.find_assignee(0),
|mut selected| {
assert_eq!(
selected.set_status(0)?,
BulkSetOutcome::Updated {
selected: 100,
changed: 50
}
);
selected.set_status(2)
},
);
assert!(matches!(
result,
Err(BulkUpdateError::InsufficientStorage {
required: 100,
available: 50
})
));
assert_eq!(table.query_memberships().find_status(0).len(), 150);
assert_eq!(table.query_memberships().find_status(1).len(), 50);
assert_eq!(table.query_memberships().find_status(2).len(), 0);
Ok(())
}
#[test]
fn chunk_cutoff_preserves_assignment_and_scratch_bounds() -> Result<(), &'static str> {
for (first_count, scratch_slots) in [(64, 0), (65, 65)] {
let mut table = table((0..65537_u32).map(|id| u32::from(id < first_count || id == 65536)))?;
let mut work = workspace(65537, scratch_slots)?;
let outcome = table
.bulk_update_in(
&mut work,
|query| query.find_status(1),
|mut selected| selected.set_status(2),
)
.map_err(|_| "set")?;
let count = u64::from(first_count).checked_add(1).ok_or("count")?;
assert_eq!(
outcome,
BulkSetOutcome::Updated {
selected: count,
changed: count
}
);
assert_eq!(table.query_memberships().find_status(1).len(), 0);
assert_eq!(
table.query_memberships().find_status(2).len(),
count as usize
);
for row in table.rows() {
assert_eq!(
*row.status,
if *row.id < first_count || *row.id == 65536 {
2
} else {
0
}
);
}
}
Ok(())
}