inspection.rs
This Kit feature is not available yet.
In rust/crates/mkit-server/src/indexed/At c3d562fCopy the commit ID
Preview
Code38.4 KiB//! Bounded, metadata-only enumeration of added-pack files for launch inspection.
use std::collections::{BTreeMap, BTreeSet};
use mkit_core::hash::Hash;
use mkit_core::object::ObjectType;
use crate::ServerError;
use crate::store::{BlobBody, BlobKey, BlobStore, ByteRange, codec::TicketV1};
use futures::StreamExt as _;
/// Launch inspection kinds come directly from verified object types.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Kind {
/// Any blob, including one used only as a chunk.
Blob,
/// A `ChunkedBlob` manifest.
ChunkedFile,
}
/// One inspected object's metadata, without its bytes.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InspectObject {
/// Verified content identity.
pub id: Hash,
/// Canonical decoded object length.
pub size: u64,
/// Blob or chunked file.
pub kind: Kind,
}
/// The sorted, deduplicated file-typed entries of the advance's added packs.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InspectionSet {
limit: usize,
objects: BTreeMap<Hash, InspectObject>,
added_entries: u64,
raw_packs: BTreeSet<Hash>,
pending: Option<Added>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct NativeEntry {
pub id: Hash,
pub size: u64,
pub object_type: u8,
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum Added {
Native(Vec<NativeEntry>),
Scheduled(crate::Partition, Vec<ScheduledPack>),
}
/// Request-local proof of the exact verified job accepted during ticket checking.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct ScheduledPack {
pub pack: Hash,
pub job: crate::Value,
pub verification: crate::Value,
pub decoded_bytes: u64,
}
/// The established bounded-index refusal; request size is never unavailability.
#[must_use]
pub fn limit_error() -> ServerError {
ServerError::invalid_argument("object index limit exceeded")
}
impl InspectionSet {
/// Create a whole-advance collector with its configured object limit.
#[must_use]
pub fn new(limit: usize) -> Self {
Self {
limit,
objects: BTreeMap::new(),
added_entries: 0,
raw_packs: BTreeSet::new(),
pending: None,
}
}
/// Refuse the conservative count before decoding or enumeration.
///
/// # Errors
/// The existing index-limit error if the bound exceeds the launch limit.
pub fn preflight(&self, count: u64) -> Result<(), ServerError> {
if count > self.limit as u64 {
return Err(limit_error());
}
Ok(())
}
/// Reserve the conservative count of all added-pack entries.
///
/// # Errors
/// The existing index-limit refusal when the header/job bound is oversized.
pub fn reserve_added_count(&mut self, count: u64) -> Result<(), ServerError> {
self.preflight(count)?;
self.added_entries = count;
Ok(())
}
pub(super) fn add_raw_pack(&mut self, id: Hash) {
self.raw_packs.insert(id);
}
/// Verified raw additions; excludes packlist nodes without new byte reads.
#[cfg(feature = "remote-hooks")]
pub(crate) fn raw_packs(&self) -> &BTreeSet<Hash> {
&self.raw_packs
}
pub(super) fn defer_native(&mut self, entries: Vec<NativeEntry>) {
self.pending = Some(Added::Native(entries));
}
pub(super) fn defer_scheduled(&mut self, packs: Vec<ScheduledPack>, source: crate::Partition) {
self.raw_packs.extend(packs.iter().map(|pack| pack.pack));
self.pending = Some(Added::Scheduled(source, packs));
}
/// Collect added entries after header/job preflight, without reading bytes.
///
/// # Errors
/// Existing indexed storage, metadata decoding and bounded-enumeration refusals.
pub async fn complete_added<S: crate::NamespaceStore>(
&mut self,
store: &S,
repo: &crate::RepoId,
) -> Result<(), ServerError> {
self.preflight(self.added_entries)?;
match self.pending.take() {
Some(Added::Native(entries)) => {
for entry in entries {
self.entry(entry.id, entry.size, entry.object_type)?;
}
}
Some(Added::Scheduled(source, packs)) => {
let added =
scheduled_entries(store, repo, &source, &packs, self.limit, self.added_entries)
.await?;
self.objects = added.objects;
}
None => {}
}
Ok(())
}
/// Insert verified metadata, ignoring non-file object types.
///
/// # Errors
/// Index-limit refusal or inconsistent immutable metadata.
pub fn entry(&mut self, id: Hash, size: u64, object_type: u8) -> Result<(), ServerError> {
if !(ObjectType::Blob as u8..=ObjectType::Tag as u8).contains(&object_type) {
return Err(ServerError::unavailable(
"verified object metadata inconsistency",
));
}
let kind = if object_type == ObjectType::Blob as u8 {
Kind::Blob
} else if object_type == ObjectType::ChunkedBlob as u8 {
Kind::ChunkedFile
} else {
return Ok(());
};
if let Some(old) = self.objects.get(&id) {
if old.size != size || old.kind != kind {
return Err(ServerError::unavailable(
"verified object metadata inconsistency",
));
}
return Ok(());
}
if self.objects.len() >= self.limit {
return Err(limit_error());
}
self.objects.insert(id, InspectObject { id, size, kind });
Ok(())
}
/// Final metadata in stable object-id order.
#[must_use]
pub fn finalize(self) -> Vec<InspectObject> {
self.objects.into_values().collect()
}
}
/// Read only fixed-size headers before native whole-pack allocation/decoding.
pub(super) async fn preflight_native<B: BlobStore>(
blobs: &B,
tickets: &[TicketV1],
limit: usize,
) -> Result<u64, ServerError> {
let mut count = 0_u64;
let mut packs = BTreeSet::new();
for ticket in tickets {
if !packs.insert(ticket.pack_id) {
continue;
}
let body = blobs
.get(
&BlobKey::pack(ticket.pack_id),
Some(ByteRange {
start: 0,
end_inclusive: 11,
}),
)
.await
.map_err(|_| ServerError::unavailable("object storage request failed"))?
.ok_or_else(|| ServerError::unavailable("object storage request failed"))?;
let mut prefix = Vec::with_capacity(12);
match body {
BlobBody::Bytes(bytes) => {
if bytes.len() != 12 {
return Err(ServerError::invalid_argument("object hash mismatch"));
}
prefix.extend_from_slice(&bytes);
}
BlobBody::Stream { len, mut stream } => {
if len != 12 {
return Err(ServerError::unavailable("object storage request failed"));
}
while let Some(bytes) = stream.next().await {
let bytes = bytes
.map_err(|_| ServerError::unavailable("object storage request failed"))?;
if prefix.len().saturating_add(bytes.len()) > 12 {
return Err(ServerError::unavailable("object storage request failed"));
}
prefix.extend_from_slice(&bytes);
}
}
}
if prefix.len() != 12 {
return Err(ServerError::invalid_argument("object hash mismatch"));
}
if super::classify::classify(&prefix)? == super::classify::UploadType::Pack {
let entries = u32::from_le_bytes(prefix[8..12].try_into().map_err(|_| limit_error())?);
count = count.saturating_add(u64::from(entries));
if count > limit as u64 {
return Err(limit_error());
}
}
}
Ok(count)
}
fn metadata_error(error: &crate::StoreError) -> ServerError {
if super::budget::is_exhausted(error) {
limit_error()
} else {
ServerError::unavailable("object storage request failed")
}
}
async fn guard_jobs<S: crate::NamespaceStore>(
store: &S,
repo: &crate::RepoId,
source: &crate::Partition,
packs: &[ScheduledPack],
) -> Result<(), ServerError> {
if packs.is_empty() {
return Ok(());
}
let keys = packs
.iter()
.flat_map(|pack| {
[
crate::store::keys::verify_job(&repo.name, &pack.pack),
crate::store::keys::verification(&repo.name, &pack.pack),
]
})
.collect::<Vec<_>>();
let current = store
.get_many(source, &keys)
.await
.map_err(|error| metadata_error(&error))?;
if current.len() != keys.len() {
return Err(ServerError::unavailable("object storage request failed"));
}
if packs.iter().enumerate().any(|(i, pack)| {
current[2 * i].as_ref() != Some(&pack.job)
|| current[2 * i + 1].as_ref() != Some(&pack.verification)
}) {
return Err(super::pending(1000));
}
Ok(())
}
/// Page verified first-occurrence frame rows, without reference scans or R2 reads.
pub(super) async fn scheduled_entries<S: crate::NamespaceStore>(
store: &S,
repo: &crate::RepoId,
source: &crate::Partition,
additions: &[ScheduledPack],
limit: usize,
added_count: u64,
) -> Result<InspectionSet, ServerError> {
use crate::store::keys;
let failed = || ServerError::unavailable("object storage request failed");
let mut set = InspectionSet::new(limit);
set.reserve_added_count(added_count)?;
guard_jobs(store, repo, source, additions).await?;
let mut entries = 0_u64;
for pack in additions {
let mut decoded_bytes = 0_u64;
let (start, end) = keys::verify_range(&repo.name, &pack.pack, Some(keys::VC_FRAME));
let mut after = None;
loop {
let page = store
.scan(source, &start, &end, after.as_ref(), 1000)
.await
.map_err(|error| metadata_error(&error))?;
if page.entries.len() > 1000 {
return Err(failed());
}
for (key, raw) in page.entries {
let Some(keys::ParsedKey::VerifyCursor {
repo: found,
pack_id,
sub,
id: Some(id),
}) = keys::parse(&key)
else {
return Err(failed());
};
if found != repo.name || pack_id != pack.pack || sub != keys::VC_FRAME {
return Err(failed());
}
entries = entries.saturating_add(1);
if entries > added_count {
return Err(failed());
}
let frame = super::checkpoint::decode_frame(&id, &raw).map_err(|_| failed())?;
decoded_bytes = decoded_bytes
.checked_add(frame.value.decoded_size)
.ok_or_else(failed)?;
set.entry(id, frame.value.decoded_size, frame.object_type)?;
}
match page.next {
Some(next) if after.as_ref() != Some(&next) => after = Some(next),
Some(_) => return Err(failed()),
None => break,
}
}
if decoded_bytes != pack.decoded_bytes {
return Err(super::pending(1000));
}
}
guard_jobs(store, repo, source, additions).await?;
Ok(set)
}
#[cfg(all(test, feature = "memory"))]
mod tests {
use super::*;
use crate::indexed::checkpoint::{FrameRow, encode_frame};
use crate::indexed::tests::{NOW, repo, source, ticket, upload};
use crate::memory::{MemoryBlobStore, MemoryKv};
use crate::pipeline::SinglePartition;
use crate::rt::ManualClock;
use crate::store::{Batch, NamespaceStore, index::IndexValue, keys};
use crate::telemetry::NoopMetrics;
use futures_executor::block_on;
use mkit_core::hash::hash;
use mkit_core::object::{
Blob, ChunkedBlob, Commit, EntryMode, Identity, Object, Tree, TreeEntry,
};
use mkit_core::pack::{DecodeLimits, NoExternalBases, PackWriter, decode_entries_with};
use mkit_core::serialize::serialize;
use mkit_core::sign::{KeyPair, sign_commit};
use std::sync::Arc;
fn verified_pack(
store: &MemoryKv,
repo: &crate::RepoId,
pack: Hash,
entries: u64,
decoded_bytes: u64,
) -> ScheduledPack {
let mut job = super::super::checkpoint::VerifyJobV1::new(hash(&pack), NOW as u64, 1, 4096);
job.kind = super::super::checkpoint::Kind::Pack;
job.phase = super::super::checkpoint::Phase::Watch;
job.entries = entries;
job.in_pack_bytes = decoded_bytes;
let snapshot = ScheduledPack {
pack,
job: super::super::checkpoint::encode_job(&job),
verification: super::super::state::encode(
&super::super::state::VerificationV1::Verified {
pack_len: 1,
verified_at_ms: NOW as u64,
publication: None,
},
),
decoded_bytes,
};
block_on(
store.apply(
&source(repo),
Batch::new()
.put(keys::verify_job(&repo.name, &pack), snapshot.job.clone())
.put(
keys::verification(&repo.name, &pack),
snapshot.verification.clone(),
),
),
)
.expect("seed deterministic verified-pack metadata");
snapshot
}
#[allow(clippy::unwrap_used)] // Deterministic valid objects used only by these tests.
fn fixture() -> (Vec<u8>, Hash, Vec<(Hash, Kind)>) {
let chunk = Object::Blob(Blob {
data: b"one".to_vec(),
});
let dual = Object::Blob(Blob {
data: b"two".to_vec(),
});
let extra = Object::Blob(Blob {
data: b"surplus".to_vec(),
});
let manifest = Object::ChunkedBlob(ChunkedBlob {
total_size: 6,
chunk_size: 0,
chunks: vec![chunk.id().unwrap(), dual.id().unwrap()],
});
let tree = Object::Tree(Tree {
entries: vec![
TreeEntry {
name: b"direct".to_vec(),
mode: EntryMode::Blob,
object_hash: dual.id().unwrap(),
},
TreeEntry {
name: b"manifest".to_vec(),
mode: EntryMode::Blob,
object_hash: manifest.id().unwrap(),
},
],
});
let key = KeyPair::from_seed([7; 32]);
let mut commit = Commit::new_unannotated(
tree.id().unwrap(),
vec![],
Identity::ed25519(key.public.0),
key.public.0,
b"inspection".to_vec(),
42,
[0; 64],
);
commit.signature = sign_commit(&commit, &key).unwrap().0;
let commit = Object::Commit(commit);
let mut writer = PackWriter::new_raw_only();
for object in [&chunk, &dual, &extra, &manifest, &tree, &commit, &chunk] {
writer
.push_raw(object.id().unwrap(), &serialize(object).unwrap())
.unwrap();
}
let mut expected = vec![
(chunk.id().unwrap(), Kind::Blob),
(dual.id().unwrap(), Kind::Blob),
(extra.id().unwrap(), Kind::Blob),
(manifest.id().unwrap(), Kind::ChunkedFile),
];
expected.sort_by_key(|e| e.0);
(writer.finish().unwrap(), commit.id().unwrap(), expected)
}
#[test]
#[allow(clippy::too_many_lines)] // One native/scheduled parity scenario and its preflight failure.
fn native_and_worker_enumerate_surplus_manifests_chunks_dual_uses_and_duplicates() {
let (pack, head, expected) = fixture();
let repo = repo("inspection-union");
let blobs = MemoryBlobStore::default();
upload(&blobs, &pack);
let clock = Arc::new(ManualClock::new(NOW));
let store = MemoryKv::with_clock(clock.clone());
let ticket = ticket(&repo, &pack, NOW as u64);
let mut native = block_on(super::super::verify::verify_ticketed_inspected(
&blobs,
&store,
&SinglePartition,
&repo,
&source(&repo),
std::slice::from_ref(&ticket),
&[hash(&ticket.pack_id)],
head,
super::super::IndexedConfig::default(),
true,
clock.as_ref(),
&NoopMetrics,
10,
))
.unwrap()
.inspection
.unwrap();
assert!(native.objects.is_empty());
block_on(native.complete_added(&store, &repo)).unwrap();
let native = native.finalize();
assert_eq!(
native.iter().map(|e| (e.id, e.kind)).collect::<Vec<_>>(),
expected
);
let mut frames = BTreeMap::new();
decode_entries_with(
&pack,
&mut NoExternalBases,
DecodeLimits::default(),
|entry| {
let row = FrameRow {
object_type: entry.object.object_type() as u8,
external: None,
value: IndexValue {
frame_offset: entry.frame_offset,
frame_length: entry.frame_length,
wire_type: entry.wire_type,
decoded_size: entry.bytes.len() as u64,
chain_depth: 0,
delta_base: None,
},
};
frames
.entry(entry.id)
.or_insert_with(|| encode_frame(&entry.id, &row).unwrap());
Ok(())
},
)
.unwrap();
let in_pack_bytes = frames
.iter()
.map(|(id, raw)| {
super::super::checkpoint::decode_frame(id, raw)
.unwrap()
.value
.decoded_size
})
.sum();
let mut batch = Batch::new();
for (id, raw) in frames {
batch = batch.put(
keys::verify_row(&repo.name, &ticket.pack_id, keys::VC_FRAME, Some(&id)),
raw,
);
}
block_on(store.apply(&source(&repo), batch)).unwrap();
let mut job = super::super::checkpoint::VerifyJobV1::new(
hash(&ticket.pack_id),
NOW as u64,
pack.len() as u64,
4096,
);
job.kind = super::super::checkpoint::Kind::Pack;
job.phase = super::super::checkpoint::Phase::Watch;
job.entries = 7;
job.in_pack_bytes = in_pack_bytes;
let ready = Batch::new()
.put(
keys::verify_job(&repo.name, &ticket.pack_id),
super::super::checkpoint::encode_job(&job),
)
.put(
keys::verification(&repo.name, &ticket.pack_id),
super::super::state::encode(&super::super::state::VerificationV1::Verified {
pack_len: pack.len() as u64,
verified_at_ms: NOW as u64,
publication: None,
}),
);
block_on(store.apply(&source(&repo), ready)).unwrap();
let accepted = ScheduledPack {
pack: ticket.pack_id,
job: super::super::checkpoint::encode_job(&job),
verification: super::super::state::encode(
&super::super::state::VerificationV1::Verified {
pack_len: pack.len() as u64,
verified_at_ms: NOW as u64,
publication: None,
},
),
decoded_bytes: job.in_pack_bytes,
};
let worker = block_on(scheduled_entries(
&store,
&repo,
&source(&repo),
&[accepted],
10,
7,
))
.unwrap()
.finalize();
assert_eq!(native, worker);
let mut checked = block_on(super::super::scheduled::check_inspected(
&blobs,
&store,
&SinglePartition,
&repo,
&source(&repo),
std::slice::from_ref(&ticket),
&[job.ticket_id],
head,
super::super::IndexedConfig::default(),
clock.as_ref(),
&NoopMetrics,
10,
))
.unwrap()
.inspection
.unwrap();
assert!(checked.objects.is_empty());
block_on(checked.complete_added(&store, &repo)).unwrap();
assert_eq!(checked.finalize(), native);
let error = block_on(super::super::scheduled::check_inspected(
&blobs,
&store,
&SinglePartition,
&repo,
&source(&repo),
&[ticket],
&[job.ticket_id],
head,
super::super::IndexedConfig::default(),
clock.as_ref(),
&NoopMetrics,
6,
))
.unwrap_err();
assert_eq!(error.public_message(), "object index limit exceeded");
}
#[test]
fn native_preflight_counts_repeated_pack_once_at_the_entry_limit() {
let (pack, _, _) = fixture();
let repo = repo("inspection-repeated-pack");
let blobs = MemoryBlobStore::default();
upload(&blobs, &pack);
let mut first = ticket(&repo, &pack, NOW as u64);
first.reservation_id = "s:duplicate-first".into();
let mut second = first.clone();
second.reservation_id = "s:duplicate-second".into();
assert_ne!(
crate::store::tickets::ticket_id(&first.reservation_id),
crate::store::tickets::ticket_id(&second.reservation_id),
);
assert_eq!(first.pack_id, second.pack_id);
let calls = super::super::budget::SliceBudget::new(4);
let counted = super::super::budget::Budgeted::new(&blobs, &calls);
assert_eq!(
block_on(preflight_native(&counted, &[first, second], 7))
.map_err(|error| (error.code(), error.public_message().to_owned())),
Ok(7),
);
// One header range reserves both metadata and byte requests.
assert_eq!(calls.used(), 2);
}
#[test]
fn oversize_native_header_refuses_before_verification_writes() {
let (pack, head, _) = fixture();
let repo = repo("inspection-size");
let blobs = MemoryBlobStore::default();
upload(&blobs, &pack);
let clock = Arc::new(ManualClock::new(NOW));
let store = MemoryKv::with_clock(clock.clone());
let ticket = ticket(&repo, &pack, NOW as u64);
let error = block_on(super::super::verify::verify_ticketed_inspected(
&blobs,
&store,
&SinglePartition,
&repo,
&source(&repo),
&[ticket],
&[[1; 32]],
head,
super::super::IndexedConfig::default(),
true,
clock.as_ref(),
&NoopMetrics,
6,
))
.unwrap_err();
assert_eq!(error.code(), crate::Code::InvalidArgument);
assert_eq!(error.public_message(), "object index limit exceeded");
assert_eq!(block_on(store.stats(&source(&repo))).unwrap().keys, Some(0));
}
#[test]
fn frame_scan_budget_exhaustion_is_an_index_limit_refusal() {
let repo = repo("inspection-scan-budget");
let store = MemoryKv::default();
let accepted = verified_pack(&store, &repo, [1; 32], 1, 11);
let budget = super::super::budget::SliceBudget::new(1);
let bounded = super::super::budget::Budgeted::new(&store, &budget);
let error = block_on(scheduled_entries(
&bounded,
&repo,
&source(&repo),
&[accepted],
10_000,
1,
))
.unwrap_err();
assert_eq!(error.code(), crate::Code::InvalidArgument);
assert_eq!(error.public_message(), "object index limit exceeded");
assert_eq!(budget.used(), 1);
}
#[test]
fn conservative_added_count_is_independent_of_deduplication() {
let mut set = InspectionSet::new(4);
set.reserve_added_count(4).unwrap();
set.entry([1; 32], 11, ObjectType::Blob as u8).unwrap();
set.entry([1; 32], 11, ObjectType::Blob as u8).unwrap();
assert!(set.reserve_added_count(5).is_err());
assert_eq!(set.finalize().len(), 1);
}
#[test]
fn unknown_checkpoint_object_types_fail_closed() {
let mut set = InspectionSet::new(4);
for tag in [0, 8, 255] {
assert_eq!(
set.entry([tag; 32], 11, tag).unwrap_err().code(),
crate::Code::Unavailable
);
}
assert!(set.finalize().is_empty());
}
#[test]
#[allow(clippy::too_many_lines)] // One real 400-entry pack compares native and Worker enumeration.
fn small_valid_manifest_pack_accepts_with_one_scan_two_guards_and_native_parity() {
let repo = repo("manifest-budget-reproduction");
let mut writer = PackWriter::new_raw_only();
for n in 0_u8..200 {
let blob = Object::Blob(Blob { data: vec![n] });
let manifest = Object::ChunkedBlob(ChunkedBlob {
total_size: 1,
chunk_size: 0,
chunks: vec![blob.id().unwrap()],
});
for object in [&blob, &manifest] {
writer
.push_raw(object.id().unwrap(), &serialize(object).unwrap())
.unwrap();
}
}
let pack = writer.finish().unwrap();
let pack_id = hash(&pack);
let blobs = MemoryBlobStore::default();
upload(&blobs, &pack);
let store = MemoryKv::default();
let mut frames = Vec::new();
let mut entries = Vec::new();
decode_entries_with(
&pack,
&mut NoExternalBases,
DecodeLimits::default(),
|entry| {
let object_type = entry.object.object_type() as u8;
let row = FrameRow {
object_type,
external: None,
value: IndexValue {
frame_offset: entry.frame_offset,
frame_length: entry.frame_length,
wire_type: entry.wire_type,
decoded_size: entry.bytes.len() as u64,
chain_depth: 0,
delta_base: None,
},
};
frames.push((
keys::verify_row(&repo.name, &pack_id, keys::VC_FRAME, Some(&entry.id)),
encode_frame(&entry.id, &row).unwrap(),
));
entries.push(NativeEntry {
id: entry.id,
size: entry.bytes.len() as u64,
object_type,
});
Ok(())
},
)
.unwrap();
assert_eq!(entries.len(), 400);
for rows in frames.chunks(90) {
let batch = rows.iter().fold(Batch::new(), |batch, (key, raw)| {
batch.put(key.clone(), raw.clone())
});
block_on(store.apply(&source(&repo), batch)).unwrap();
}
let mut native = InspectionSet::new(10_000);
native.reserve_added_count(400).unwrap();
native.defer_native(entries);
block_on(native.complete_added(&store, &repo)).unwrap();
let native = native.finalize();
assert_eq!(native.len(), 400);
let metadata = super::super::budget::SliceBudget::new(256);
let mut worker = InspectionSet::new(10_000);
worker.reserve_added_count(400).unwrap();
let accepted = verified_pack(
&store,
&repo,
pack_id,
400,
native.iter().map(|e| e.size).sum(),
);
worker.defer_scheduled(vec![accepted], source(&repo));
block_on(worker.complete_added(
&super::super::budget::Budgeted::new(&store, &metadata),
&repo,
))
.unwrap();
assert_eq!(worker.finalize(), native);
assert_eq!(metadata.used(), 3);
}
#[test]
fn worker_enumeration_resumes_frame_pages_within_the_shared_call_budget() {
let repo = repo("inspection-pages");
let store = MemoryKv::default();
let pack = [9; 32];
let mut decoded_bytes = 0_u64;
for first in (0_u64..1500).step_by(90) {
let mut batch = Batch::new();
for n in first..(first + 90).min(1500) {
let bytes = serialize(&Object::Blob(Blob {
data: n.to_le_bytes().to_vec(),
}))
.unwrap();
decoded_bytes += bytes.len() as u64;
let id = hash(&bytes);
let frame = FrameRow {
object_type: ObjectType::Blob as u8,
external: None,
value: IndexValue {
frame_offset: 12 + n * 23,
frame_length: 23,
wire_type: 0,
decoded_size: bytes.len() as u64,
chain_depth: 0,
delta_base: None,
},
};
batch = batch.put(
keys::verify_row(&repo.name, &pack, keys::VC_FRAME, Some(&id)),
encode_frame(&id, &frame).unwrap(),
);
}
block_on(store.apply(&source(&repo), batch)).unwrap();
}
let accepted = verified_pack(&store, &repo, pack, 1500, decoded_bytes);
let budget = super::super::budget::SliceBudget::new(256);
let bounded = super::super::budget::Budgeted::new(&store, &budget);
let entries = block_on(scheduled_entries(
&bounded,
&repo,
&source(&repo),
&[accepted],
1500,
1500,
))
.unwrap()
.finalize();
assert_eq!(entries.len(), 1500);
assert_eq!(budget.used(), 4);
assert!(entries.windows(2).all(|w| w[0].id < w[1].id));
}
#[test]
#[allow(clippy::too_many_lines)] // Worst-case page rounding across seven consumed packs.
fn cap_across_seven_packs_uses_sixteen_scans_two_guards_and_matches_native() {
let repo = repo("inspection-seven-pack-cap");
let store = MemoryKv::default();
let counts = [1001_u64, 1001, 1001, 1001, 1001, 1001, 3994];
let mut packs = Vec::new();
let mut entries = Vec::new();
let mut sequence = 0_u64;
for (pack_number, count) in counts.into_iter().enumerate() {
let pack = [u8::try_from(pack_number + 1).unwrap(); 32];
let mut decoded_bytes = 0_u64;
for first in (0..count).step_by(90) {
let mut batch = Batch::new();
for n in first..(first + 90).min(count) {
let bytes = serialize(&Object::Blob(Blob {
data: sequence.to_le_bytes().to_vec(),
}))
.unwrap();
sequence += 1;
decoded_bytes += bytes.len() as u64;
let id = hash(&bytes);
let row = FrameRow {
object_type: ObjectType::Blob as u8,
external: None,
value: IndexValue {
frame_offset: 12 + n * 23,
frame_length: 23,
wire_type: 0,
decoded_size: bytes.len() as u64,
chain_depth: 0,
delta_base: None,
},
};
entries.push(NativeEntry {
id,
size: bytes.len() as u64,
object_type: row.object_type,
});
batch = batch.put(
keys::verify_row(&repo.name, &pack, keys::VC_FRAME, Some(&id)),
encode_frame(&id, &row).unwrap(),
);
}
block_on(store.apply(&source(&repo), batch)).unwrap();
}
packs.push(verified_pack(&store, &repo, pack, count, decoded_bytes));
}
assert_eq!(sequence, 10_000);
let metadata = super::super::budget::SliceBudget::new(18);
let bounded_store = super::super::budget::Budgeted::new(&store, &metadata);
let mut native = InspectionSet::new(10_000);
native.reserve_added_count(10_000).unwrap();
native.defer_native(entries);
block_on(native.complete_added(&bounded_store, &repo)).unwrap();
assert_eq!(metadata.used(), 0);
let mut worker = InspectionSet::new(10_000);
worker.reserve_added_count(10_000).unwrap();
worker.defer_scheduled(packs, source(&repo));
block_on(worker.complete_added(&bounded_store, &repo)).unwrap();
let native = native.finalize();
assert_eq!(native.len(), 10_000);
assert_eq!(worker.finalize(), native);
assert_eq!(metadata.used(), 18);
}
#[test]
fn missing_verified_frames_are_pending_before_inspection() {
let repo = repo("inspection-missing-frames");
let store = MemoryKv::default();
let pack = [7; 32];
let accepted = verified_pack(&store, &repo, pack, 1, 11);
let mut worker = InspectionSet::new(10_000);
worker.reserve_added_count(1).unwrap();
worker.defer_scheduled(vec![accepted], source(&repo));
let error = block_on(worker.complete_added(&store, &repo)).unwrap_err();
assert_eq!(error.public_message(), "pack verification pending");
}
struct ReplaceAfterScan<'a> {
inner: &'a MemoryKv,
job_key: crate::Key,
replacement: crate::Value,
}
impl NamespaceStore for ReplaceAfterScan<'_> {
fn capabilities(&self) -> crate::StoreCapabilities {
self.inner.capabilities()
}
async fn get(
&self,
p: &crate::Partition,
key: &crate::Key,
) -> Result<Option<crate::Value>, crate::StoreError> {
self.inner.get(p, key).await
}
async fn get_many(
&self,
p: &crate::Partition,
keys: &[crate::Key],
) -> Result<Vec<Option<crate::Value>>, crate::StoreError> {
self.inner.get_many(p, keys).await
}
async fn scan(
&self,
p: &crate::Partition,
start: &crate::Key,
end: &crate::Key,
after: Option<&crate::store::Cursor>,
limit: u32,
) -> Result<crate::store::ScanPage, crate::StoreError> {
let page = self.inner.scan(p, start, end, after, limit).await?;
self.inner
.apply(
p,
Batch::new().put(self.job_key.clone(), self.replacement.clone()),
)
.await?;
Ok(page)
}
async fn apply(
&self,
p: &crate::Partition,
batch: Batch,
) -> Result<crate::BatchOutcome, crate::StoreError> {
self.inner.apply(p, batch).await
}
async fn stats(
&self,
p: &crate::Partition,
) -> Result<crate::PartitionStats, crate::StoreError> {
self.inner.stats(p).await
}
async fn probe(&self) -> Result<(), crate::StoreError> {
self.inner.probe().await
}
}
#[test]
fn replacement_verification_job_is_pending_before_and_during_enumeration() {
for during_scan in [false, true] {
let repo = repo("inspection-replaced-job");
let store = MemoryKv::default();
let pack = [7; 32];
let accepted = verified_pack(&store, &repo, pack, 1, 11);
let frame = FrameRow {
object_type: ObjectType::Blob as u8,
external: None,
value: IndexValue {
frame_offset: 12,
frame_length: 23,
wire_type: 0,
decoded_size: 11,
chain_depth: 0,
delta_base: None,
},
};
block_on(store.apply(
&source(&repo),
Batch::new().put(
keys::verify_row(&repo.name, &pack, keys::VC_FRAME, Some(&[9; 32])),
encode_frame(&[9; 32], &frame).unwrap(),
),
))
.unwrap();
let mut replacement = super::super::checkpoint::decode_job(&accepted.job).unwrap();
replacement.ticket_id = [99; 32];
replacement.phase = super::super::checkpoint::Phase::Decode;
let replacement = super::super::checkpoint::encode_job(&replacement);
let job_key = keys::verify_job(&repo.name, &pack);
let budget = super::super::budget::SliceBudget::new(3);
let mut worker = InspectionSet::new(10_000);
worker.reserve_added_count(1).unwrap();
worker.defer_scheduled(vec![accepted], source(&repo));
let error =
if during_scan {
let racing = ReplaceAfterScan {
inner: &store,
job_key,
replacement,
};
block_on(worker.complete_added(
&super::super::budget::Budgeted::new(&racing, &budget),
&repo,
))
.unwrap_err()
} else {
block_on(store.apply(&source(&repo), Batch::new().put(job_key, replacement)))
.unwrap();
block_on(worker.complete_added(
&super::super::budget::Budgeted::new(&store, &budget),
&repo,
))
.unwrap_err()
};
assert_eq!(error.public_message(), "pack verification pending");
assert_eq!(budget.used(), if during_scan { 3 } else { 1 });
}
}
}