mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-07 20:46:11 +00:00
chore(scanner): stage Scanner/Heal follow-up slices (#7374)
* fix(scanner): remove unused digest import Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * feat(scanner): add raw page owner index (#7375) * feat(scanner): add raw page owner index Add a serializable raw enumeration page owner index for scanner resume work. The index exposes unsupported, building, and ready states, validates committed page identity by recomputing digests, and uses generation checks for CAS-style page commits. Focused tests cover small-budget restart progress, page digest/source drift rejection, corrupt deserialized state, CAS failure, precommit crash, empty sources, and invalid entry boundaries. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * feat(scanner): persist raw page owner resume state (#7379) Wire the scanner raw enumeration partial-cache writer to the raw page owner index so interrupted bucket walks can retain validated page-builder state across scanner restarts. Keep complete owner sources terminal-only, add partial-source ingestion for in-progress raw directory reads, and validate the persisted page index through bucket checkpoint preparation. Co-authored-by: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com> * test(scanner): fence segment producer observations (#7381) Require the segment observation fixture to carry source, incarnation, key-format, baseline, process epoch, generation-window, gap, overflow, and producer-coverage proof before accepting a narrowed proposal. Keep the diagnostic path fixture-only and remove its ordinary stderr output. Co-authored-by: zhi22915 <qiuzgang@gmail.com> * fix(ecstore): isolate pool metadata read probes (#7367) Co-authored-by: zhi22915 <qiuzgang@gmail.com> * test(heal): cover MRF crash successor matrix (#7369) * test(heal): cover MRF crash successor matrix Add process-boundary MRF replay coverage for the successor snapshot window after a retained startup journal is flushed but before cleanup deletes it. Extend the mixed authoritative/legacy reader fixture with a scoped v2 journal epoch to pin the no-merge contract. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(heal): cover service-kill MRF replay (#7380) Add a Unix process fixture that waits after publishing the pending MRF successor snapshot, then is terminated by the parent before restart replay. Co-authored-by: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com> * test(heal): cover transport-lost start receipts (#7371) Add gRPC transport fault fixtures for heal-control start admission. The tests distinguish pre-admission transport loss from post-admission response loss, then verify exact envelope retries reuse the canonical receipt while fresh forceStart requests create distinct tasks. Co-authored-by: zhi22915 <qiuzgang@gmail.com> * test(scanner): add crash-restart heal evidence case (#7370) * test(scanner): add crash-restart heal evidence case Add a distinct W21 background target crash case to the scanner/heal evidence registry and oracle path. Keep the existing restart lane on graceful process restart, keep the crash lane on hard kill, and make the wiring checker reject evidence/oracle mismatches. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(scanner): support older Python wiring checks Let the scanner/heal evidence wiring checker run under Python 3.9/3.10 by falling back to tomli and chunked SHA-256 hashing when the Python 3.11 standard APIs are unavailable. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com> * fix(scanner): reject stale raw page source seeds (#7382) Do not prefill a resumed raw page owner with previously indexed entries when starting a new raw directory observation pass. The next pass must observe the same prefix again before the page index can advance; otherwise the index is discarded fail-closed. Co-authored-by: zhi22915 <qiuzgang@gmail.com> * fix(scanner): defer raw page revalidation until observed (#7384) A resumed raw page owner index must not prefill entries from older cache state, but it also must not discard a valid multi-entry index before the current raw directory pass has observed enough entries to prove identity. Track the persisted index floor and only run the strict owner identity check once the current pass reaches that floor. Co-authored-by: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
@@ -38,6 +38,7 @@ use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
|
||||
use tokio::time::{Duration, Instant, sleep, timeout};
|
||||
use tracing::{debug, warn};
|
||||
|
||||
use crate::raw_page_index::{RawEnumerationPageIndex, RawEnumerationPageOwnerStatus};
|
||||
use crate::storage_api::owner::HTTPPreconditions;
|
||||
use crate::{
|
||||
BUCKET_META_PREFIX, EcstoreError as Error, EcstoreResult as StorageResult, RUSTFS_META_BUCKET, ReplicationConfig,
|
||||
@@ -601,6 +602,8 @@ pub struct DataUsageCacheInfo {
|
||||
pub scan_checkpoint: Option<DataUsageScanCheckpoint>,
|
||||
#[serde(default)]
|
||||
pub scan_raw_enumeration_cursor: Option<DataUsageRawEnumerationCursor>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub scan_raw_enumeration_page_index: Option<RawEnumerationPageIndex>,
|
||||
#[serde(default)]
|
||||
pub scan_identity: Option<DataUsageScanIdentity>,
|
||||
#[serde(default)]
|
||||
@@ -661,6 +664,7 @@ impl Serialize for DataUsageCacheInfo {
|
||||
// appended by newer scanner versions during rolling upgrades.
|
||||
let field_count = 16
|
||||
+ usize::from(self.scan_raw_enumeration_cursor.is_some())
|
||||
+ usize::from(self.scan_raw_enumeration_page_index.is_some())
|
||||
+ usize::from(self.scan_identity.is_some())
|
||||
+ usize::from(self.scan_progress.is_some())
|
||||
+ usize::from(self.scan_coverage_receipt.is_some())
|
||||
@@ -687,6 +691,9 @@ impl Serialize for DataUsageCacheInfo {
|
||||
if let Some(cursor) = &self.scan_raw_enumeration_cursor {
|
||||
state.serialize_entry("scan_raw_enumeration_cursor", cursor)?;
|
||||
}
|
||||
if let Some(index) = &self.scan_raw_enumeration_page_index {
|
||||
state.serialize_entry("scan_raw_enumeration_page_index", index)?;
|
||||
}
|
||||
if let Some(identity) = self.scan_identity {
|
||||
state.serialize_entry("scan_identity", &identity)?;
|
||||
}
|
||||
@@ -895,6 +902,7 @@ impl DataUsageCache {
|
||||
&& self.info.scan_progress.is_none()
|
||||
&& self.info.scan_checkpoint.is_none()
|
||||
&& self.info.scan_raw_enumeration_cursor.is_none()
|
||||
&& self.info.scan_raw_enumeration_page_index.is_none()
|
||||
&& self.info.scan_resume_after.is_none()
|
||||
&& self.info.scan_coverage_receipt.is_none()
|
||||
&& self.info.scan_plan_digest == Some(scan_plan_digest)
|
||||
@@ -922,12 +930,17 @@ impl DataUsageCache {
|
||||
if self.validated_raw_enumeration_cursor().is_none() {
|
||||
self.info.scan_raw_enumeration_cursor = None;
|
||||
}
|
||||
if self.validated_raw_enumeration_page_index().is_none() {
|
||||
self.info.scan_raw_enumeration_page_index = None;
|
||||
}
|
||||
let cursor_is_valid = (self.info.scan_checkpoint.is_none()
|
||||
&& self.info.scan_raw_enumeration_cursor.is_none()
|
||||
&& self.info.scan_raw_enumeration_page_index.is_none()
|
||||
&& self.info.scan_resume_after.is_none()
|
||||
&& self.info.scan_coverage_receipt.is_none())
|
||||
|| self.validated_scan_frontier().is_some()
|
||||
|| self.info.scan_raw_enumeration_cursor.is_some();
|
||||
|| self.info.scan_raw_enumeration_cursor.is_some()
|
||||
|| self.info.scan_raw_enumeration_page_index.is_some();
|
||||
if !cursor_is_valid {
|
||||
self.info.scan_progress = None;
|
||||
}
|
||||
@@ -949,6 +962,7 @@ impl DataUsageCache {
|
||||
self.info.scan_resume_after = None;
|
||||
self.info.scan_checkpoint = None;
|
||||
self.info.scan_raw_enumeration_cursor = None;
|
||||
self.info.scan_raw_enumeration_page_index = None;
|
||||
self.info.scan_coverage_receipt = None;
|
||||
}
|
||||
// Old readers do not understand coverage sweeps. An absent plan makes
|
||||
@@ -1026,6 +1040,25 @@ impl DataUsageCache {
|
||||
.then_some(cursor)
|
||||
}
|
||||
|
||||
pub(crate) fn validated_raw_enumeration_page_index(&self) -> Option<&RawEnumerationPageIndex> {
|
||||
let index = self.info.scan_raw_enumeration_page_index.as_ref()?;
|
||||
if self.info.scan_progress.is_none()
|
||||
|| !self.info.scan_identity.is_some_and(|identity| identity.is_valid())
|
||||
|| self.info.source.is_none()
|
||||
|| index.committed_entries().is_err()
|
||||
|| index.indexed_entries().is_err()
|
||||
{
|
||||
return None;
|
||||
}
|
||||
let parent = match index.status() {
|
||||
RawEnumerationPageOwnerStatus::Unsupported => return None,
|
||||
RawEnumerationPageOwnerStatus::Building { parent, .. } | RawEnumerationPageOwnerStatus::Ready { parent, .. } => {
|
||||
parent
|
||||
}
|
||||
};
|
||||
path_is_in_bucket_scope(&self.info.name, &parent).then_some(index)
|
||||
}
|
||||
|
||||
/// Seal only the frontier supplied by completed traversal, never a restored cursor.
|
||||
pub(crate) fn seal_scan_frontier(&mut self, frontier: Option<&str>) -> Result<(), serde_json::Error> {
|
||||
if self.info.scan_progress.is_none() {
|
||||
|
||||
@@ -1179,6 +1179,7 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() {
|
||||
7,
|
||||
[7; 32],
|
||||
)),
|
||||
scan_raw_enumeration_page_index: Some(raw_page_index_fixture("bucket/prefix", &["entry-a"], false)),
|
||||
snapshot_complete: true,
|
||||
scan_plan_digest: Some(TEST_PLAN_DIGEST),
|
||||
scan_execution_digest: Some(DataUsageScanPlanDigest([42; 32])),
|
||||
@@ -1207,6 +1208,7 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() {
|
||||
.map(|cursor| cursor.last_entry.as_deref()),
|
||||
Some(Some("last-object"))
|
||||
);
|
||||
assert!(current.info.scan_raw_enumeration_page_index.is_some());
|
||||
assert!(current.info.snapshot_complete);
|
||||
assert_eq!(current.info.scan_plan_digest, Some(TEST_PLAN_DIGEST));
|
||||
assert_eq!(current.info.scan_execution_digest, Some(DataUsageScanPlanDigest([42; 32])));
|
||||
@@ -1260,6 +1262,24 @@ fn cache_with_raw_cursor(cursor: DataUsageRawEnumerationCursor) -> DataUsageCach
|
||||
}
|
||||
}
|
||||
|
||||
fn raw_page_index_fixture(parent: &str, entries: &[&str], complete: bool) -> RawEnumerationPageIndex {
|
||||
let mut index = RawEnumerationPageIndex::new(parent, 2).expect("raw page index should initialize");
|
||||
let generation = index.generation().expect("raw page index should expose generation");
|
||||
let outcome = if complete {
|
||||
index.ingest_owner_entries(entries.iter().map(|entry| (*entry).to_string()), entries.len().max(1), generation)
|
||||
} else {
|
||||
index.ingest_partial_owner_entries(entries.iter().map(|entry| (*entry).to_string()), entries.len().max(1), generation)
|
||||
}
|
||||
.expect("raw page index fixture should ingest entries");
|
||||
if outcome.ready_to_commit {
|
||||
let generation = index.generation().expect("raw page index should expose commit generation");
|
||||
index
|
||||
.commit_building_page(generation)
|
||||
.expect("raw page index fixture should commit ready page");
|
||||
}
|
||||
index
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn raw_enumeration_cursor_validation_requires_bucket_identity_and_bounded_marker() {
|
||||
let valid = DataUsageRawEnumerationCursor::new("bucket/raw".to_string(), Some("entry-001".to_string()), 1, [8; 32]);
|
||||
@@ -1346,6 +1366,44 @@ fn prepare_bucket_checkpoint_preserves_only_valid_raw_enumeration_cursor() {
|
||||
assert!(cache.info.scan_progress.is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prepare_bucket_checkpoint_preserves_only_valid_raw_page_index() {
|
||||
let identity = valid_scan_identity();
|
||||
let source = DataUsageCacheSource::new(1, 2);
|
||||
let page_index = raw_page_index_fixture("bucket/raw", &["entry-001"], false);
|
||||
let mut cache = DataUsageCache {
|
||||
info: DataUsageCacheInfo {
|
||||
name: "bucket".to_string(),
|
||||
leader_epoch: 1,
|
||||
source: Some(source),
|
||||
cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
|
||||
scan_identity: Some(identity),
|
||||
tier_registry_generation: Some(9),
|
||||
scan_progress: Some(DataUsageScanProgress {
|
||||
started_plan: TEST_PLAN_DIGEST,
|
||||
requested_plan: TEST_PLAN_DIGEST,
|
||||
}),
|
||||
scan_raw_enumeration_page_index: Some(page_index.clone()),
|
||||
..Default::default()
|
||||
},
|
||||
..Default::default()
|
||||
};
|
||||
assert_eq!(
|
||||
cache.prepare_bucket_checkpoint("bucket", 1, 1, source, TEST_PLAN_DIGEST, identity),
|
||||
DataUsageCachePrepareOutcome::Reused
|
||||
);
|
||||
assert_eq!(cache.info.scan_raw_enumeration_page_index, Some(page_index));
|
||||
|
||||
let invalid = raw_page_index_fixture("other/raw", &["entry-001"], false);
|
||||
cache.info.scan_raw_enumeration_page_index = Some(invalid);
|
||||
assert_eq!(
|
||||
cache.prepare_bucket_checkpoint("bucket", 1, 1, source, TEST_PLAN_DIGEST, identity),
|
||||
DataUsageCachePrepareOutcome::Reused
|
||||
);
|
||||
assert!(cache.info.scan_raw_enumeration_page_index.is_none());
|
||||
assert!(cache.info.scan_progress.is_some());
|
||||
}
|
||||
|
||||
/// Deterministic, fully populated cache used to pin the persisted
|
||||
/// `.usage-cache.bin` wire bytes. Every map/set holds at most one element
|
||||
/// so the map-encoded `marshal_msg` output is byte-stable.
|
||||
|
||||
@@ -60,6 +60,7 @@ use uuid::Uuid;
|
||||
pub mod data_usage_define;
|
||||
pub mod error;
|
||||
pub mod prefix_usage;
|
||||
pub mod raw_page_index;
|
||||
mod remote_scanner;
|
||||
pub mod runtime_config;
|
||||
pub mod scanner;
|
||||
|
||||
@@ -0,0 +1,945 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sha2::{Digest as _, Sha256};
|
||||
use thiserror::Error;
|
||||
|
||||
const RAW_PAGE_INDEX_VERSION: u16 = 1;
|
||||
const RAW_PAGE_ENTRY_MAX_BYTES: usize = 16 * 1024;
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct RawEnumerationPageIndex {
|
||||
state: RawEnumerationPageIndexState,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
|
||||
enum RawEnumerationPageIndexState {
|
||||
Unsupported,
|
||||
Supported(RawEnumerationPageIndexInner),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct RawEnumerationPageIndexInner {
|
||||
generation: u64,
|
||||
parent: String,
|
||||
page_entry_limit: usize,
|
||||
pages: Vec<RawEnumerationPage>,
|
||||
building: Option<RawEnumerationPageBuilder>,
|
||||
complete: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct RawEnumerationPage {
|
||||
version: u16,
|
||||
parent: String,
|
||||
page_index: u64,
|
||||
entries_start: u64,
|
||||
entries: Vec<String>,
|
||||
terminal: bool,
|
||||
digest: [u8; 32],
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct RawEnumerationPageBuilder {
|
||||
page_index: u64,
|
||||
entries_start: u64,
|
||||
entries: Vec<String>,
|
||||
terminal: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct RawEnumerationPageBuildOutcome {
|
||||
pub status: RawEnumerationPageOwnerStatus,
|
||||
pub ready_to_commit: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub enum RawEnumerationPageOwnerStatus {
|
||||
Unsupported,
|
||||
Building {
|
||||
generation: u64,
|
||||
parent: String,
|
||||
page_index: u64,
|
||||
indexed_entries: u64,
|
||||
buffered_entries: usize,
|
||||
},
|
||||
Ready {
|
||||
generation: u64,
|
||||
parent: String,
|
||||
committed_pages: usize,
|
||||
indexed_entries: u64,
|
||||
complete: bool,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Debug, Error, PartialEq, Eq)]
|
||||
pub enum RawEnumerationPageIndexError {
|
||||
#[error("raw enumeration page owner is unsupported")]
|
||||
Unsupported,
|
||||
#[error("raw enumeration page owner generation changed")]
|
||||
StaleGeneration,
|
||||
#[error("raw enumeration page identity changed before committed coverage")]
|
||||
IdentityMismatch,
|
||||
#[error("raw enumeration page owner requires a non-empty parent")]
|
||||
EmptyParent,
|
||||
#[error("raw enumeration page entry limit must be non-zero")]
|
||||
EmptyPage,
|
||||
#[error("raw enumeration page build budget must be non-zero")]
|
||||
EmptyBudget,
|
||||
#[error("raw enumeration page entry is invalid")]
|
||||
InvalidEntry,
|
||||
#[error("raw enumeration page has no staged entries")]
|
||||
EmptyCommit,
|
||||
#[error("raw enumeration page index is corrupt")]
|
||||
CorruptIndex,
|
||||
}
|
||||
|
||||
impl RawEnumerationPageIndex {
|
||||
pub fn unsupported() -> Self {
|
||||
Self {
|
||||
state: RawEnumerationPageIndexState::Unsupported,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new(parent: impl Into<String>, page_entry_limit: usize) -> Result<Self, RawEnumerationPageIndexError> {
|
||||
let parent = parent.into();
|
||||
if parent.is_empty() {
|
||||
return Err(RawEnumerationPageIndexError::EmptyParent);
|
||||
}
|
||||
if page_entry_limit == 0 {
|
||||
return Err(RawEnumerationPageIndexError::EmptyPage);
|
||||
}
|
||||
Ok(Self {
|
||||
state: RawEnumerationPageIndexState::Supported(RawEnumerationPageIndexInner {
|
||||
generation: 0,
|
||||
parent,
|
||||
page_entry_limit,
|
||||
pages: Vec::new(),
|
||||
building: None,
|
||||
complete: false,
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn generation(&self) -> Option<u64> {
|
||||
match &self.state {
|
||||
RawEnumerationPageIndexState::Unsupported => None,
|
||||
RawEnumerationPageIndexState::Supported(inner) => Some(inner.generation),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn status(&self) -> RawEnumerationPageOwnerStatus {
|
||||
match &self.state {
|
||||
RawEnumerationPageIndexState::Unsupported => RawEnumerationPageOwnerStatus::Unsupported,
|
||||
RawEnumerationPageIndexState::Supported(inner) => inner.status(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn ingest_owner_entries<I>(
|
||||
&mut self,
|
||||
entries: I,
|
||||
max_new_entries: usize,
|
||||
expected_generation: u64,
|
||||
) -> Result<RawEnumerationPageBuildOutcome, RawEnumerationPageIndexError>
|
||||
where
|
||||
I: IntoIterator<Item = String>,
|
||||
{
|
||||
self.ingest_owner_entries_inner(entries, max_new_entries, expected_generation, true)
|
||||
}
|
||||
|
||||
pub fn ingest_partial_owner_entries<I>(
|
||||
&mut self,
|
||||
entries: I,
|
||||
max_new_entries: usize,
|
||||
expected_generation: u64,
|
||||
) -> Result<RawEnumerationPageBuildOutcome, RawEnumerationPageIndexError>
|
||||
where
|
||||
I: IntoIterator<Item = String>,
|
||||
{
|
||||
self.ingest_owner_entries_inner(entries, max_new_entries, expected_generation, false)
|
||||
}
|
||||
|
||||
fn ingest_owner_entries_inner<I>(
|
||||
&mut self,
|
||||
entries: I,
|
||||
max_new_entries: usize,
|
||||
expected_generation: u64,
|
||||
source_complete: bool,
|
||||
) -> Result<RawEnumerationPageBuildOutcome, RawEnumerationPageIndexError>
|
||||
where
|
||||
I: IntoIterator<Item = String>,
|
||||
{
|
||||
if max_new_entries == 0 {
|
||||
return Err(RawEnumerationPageIndexError::EmptyBudget);
|
||||
}
|
||||
let RawEnumerationPageIndexState::Supported(inner) = &mut self.state else {
|
||||
return Err(RawEnumerationPageIndexError::Unsupported);
|
||||
};
|
||||
if inner.generation != expected_generation {
|
||||
return Err(RawEnumerationPageIndexError::StaleGeneration);
|
||||
}
|
||||
let mut entries = normalize_owner_entries(entries)?;
|
||||
let committed_entries = inner.validated_committed_entries()?;
|
||||
if inner.complete {
|
||||
if entries != committed_entries {
|
||||
return Err(RawEnumerationPageIndexError::IdentityMismatch);
|
||||
}
|
||||
return Ok(RawEnumerationPageBuildOutcome {
|
||||
status: inner.status(),
|
||||
ready_to_commit: false,
|
||||
});
|
||||
}
|
||||
|
||||
if !entries.starts_with(&committed_entries) {
|
||||
return Err(RawEnumerationPageIndexError::IdentityMismatch);
|
||||
}
|
||||
|
||||
let mut indexed_entries = committed_entries.len();
|
||||
if let Some(building) = &inner.building {
|
||||
building.validate(
|
||||
u64::try_from(inner.pages.len()).unwrap_or(u64::MAX),
|
||||
u64::try_from(committed_entries.len()).unwrap_or(u64::MAX),
|
||||
inner.page_entry_limit,
|
||||
)?;
|
||||
if !entries[committed_entries.len()..].starts_with(&building.entries) {
|
||||
return Err(RawEnumerationPageIndexError::IdentityMismatch);
|
||||
}
|
||||
indexed_entries = indexed_entries.saturating_add(building.entries.len());
|
||||
}
|
||||
if source_complete && indexed_entries == entries.len() && inner.building.is_none() {
|
||||
inner.complete = true;
|
||||
inner.generation = inner.generation.saturating_add(1);
|
||||
return Ok(RawEnumerationPageBuildOutcome {
|
||||
status: inner.status(),
|
||||
ready_to_commit: false,
|
||||
});
|
||||
}
|
||||
|
||||
let ready_to_commit = {
|
||||
let page_index = u64::try_from(inner.pages.len()).unwrap_or(u64::MAX);
|
||||
let entries_start = u64::try_from(committed_entries.len()).unwrap_or(u64::MAX);
|
||||
let building = inner.building.get_or_insert_with(|| RawEnumerationPageBuilder {
|
||||
page_index,
|
||||
entries_start,
|
||||
entries: Vec::new(),
|
||||
terminal: false,
|
||||
});
|
||||
if building.entries.len() >= inner.page_entry_limit {
|
||||
true
|
||||
} else {
|
||||
let remaining_page_slots = inner.page_entry_limit.saturating_sub(building.entries.len());
|
||||
let append_count = max_new_entries
|
||||
.min(remaining_page_slots)
|
||||
.min(entries.len().saturating_sub(indexed_entries));
|
||||
if append_count == 0 {
|
||||
if source_complete && !building.terminal {
|
||||
building.terminal = true;
|
||||
inner.generation = inner.generation.saturating_add(1);
|
||||
}
|
||||
} else {
|
||||
let source_entries = entries.len();
|
||||
building
|
||||
.entries
|
||||
.extend(entries.drain(indexed_entries..indexed_entries + append_count));
|
||||
building.terminal = source_complete && indexed_entries.saturating_add(append_count) == source_entries;
|
||||
inner.generation = inner.generation.saturating_add(1);
|
||||
}
|
||||
!building.entries.is_empty() && (building.terminal || building.entries.len() >= inner.page_entry_limit)
|
||||
}
|
||||
};
|
||||
|
||||
Ok(RawEnumerationPageBuildOutcome {
|
||||
status: inner.status(),
|
||||
ready_to_commit,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn commit_building_page(&mut self, expected_generation: u64) -> Result<RawEnumerationPage, RawEnumerationPageIndexError> {
|
||||
let RawEnumerationPageIndexState::Supported(inner) = &mut self.state else {
|
||||
return Err(RawEnumerationPageIndexError::Unsupported);
|
||||
};
|
||||
if inner.generation != expected_generation {
|
||||
return Err(RawEnumerationPageIndexError::StaleGeneration);
|
||||
}
|
||||
let entries_start = inner.validated_committed_entries()?.len();
|
||||
let Some(building) = inner.building.as_ref() else {
|
||||
return Err(RawEnumerationPageIndexError::EmptyCommit);
|
||||
};
|
||||
if building.entries.is_empty() {
|
||||
return Err(RawEnumerationPageIndexError::EmptyCommit);
|
||||
}
|
||||
building.validate(
|
||||
u64::try_from(inner.pages.len()).unwrap_or(u64::MAX),
|
||||
u64::try_from(entries_start).unwrap_or(u64::MAX),
|
||||
inner.page_entry_limit,
|
||||
)?;
|
||||
let Some(building) = inner.building.take() else {
|
||||
return Err(RawEnumerationPageIndexError::EmptyCommit);
|
||||
};
|
||||
let page = RawEnumerationPage::new(&inner.parent, building);
|
||||
inner.complete = page.terminal;
|
||||
inner.pages.push(page.clone());
|
||||
inner.generation = inner.generation.saturating_add(1);
|
||||
Ok(page)
|
||||
}
|
||||
|
||||
pub fn page(&self, page_index: u64, expected_digest: [u8; 32]) -> Result<&RawEnumerationPage, RawEnumerationPageIndexError> {
|
||||
let RawEnumerationPageIndexState::Supported(inner) = &self.state else {
|
||||
return Err(RawEnumerationPageIndexError::Unsupported);
|
||||
};
|
||||
let Some(page_index) = usize::try_from(page_index).ok() else {
|
||||
return Err(RawEnumerationPageIndexError::IdentityMismatch);
|
||||
};
|
||||
let Some(page) = inner.pages.get(page_index) else {
|
||||
return Err(RawEnumerationPageIndexError::IdentityMismatch);
|
||||
};
|
||||
page.validate(&inner.parent, u64::try_from(page_index).unwrap_or(u64::MAX), page.entries_start)?;
|
||||
if page.digest != expected_digest {
|
||||
return Err(RawEnumerationPageIndexError::IdentityMismatch);
|
||||
}
|
||||
Ok(page)
|
||||
}
|
||||
|
||||
pub fn committed_entries(&self) -> Result<Vec<String>, RawEnumerationPageIndexError> {
|
||||
match &self.state {
|
||||
RawEnumerationPageIndexState::Unsupported => Ok(Vec::new()),
|
||||
RawEnumerationPageIndexState::Supported(inner) => inner.validated_committed_entries(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn indexed_entries(&self) -> Result<Vec<String>, RawEnumerationPageIndexError> {
|
||||
match &self.state {
|
||||
RawEnumerationPageIndexState::Unsupported => Ok(Vec::new()),
|
||||
RawEnumerationPageIndexState::Supported(inner) => inner.validated_indexed_entries(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl RawEnumerationPageIndexInner {
|
||||
fn status(&self) -> RawEnumerationPageOwnerStatus {
|
||||
let indexed_entries = u64::try_from(self.indexed_entries()).unwrap_or(u64::MAX);
|
||||
if let Some(building) = &self.building {
|
||||
RawEnumerationPageOwnerStatus::Building {
|
||||
generation: self.generation,
|
||||
parent: self.parent.clone(),
|
||||
page_index: building.page_index,
|
||||
indexed_entries,
|
||||
buffered_entries: building.entries.len(),
|
||||
}
|
||||
} else {
|
||||
RawEnumerationPageOwnerStatus::Ready {
|
||||
generation: self.generation,
|
||||
parent: self.parent.clone(),
|
||||
committed_pages: self.pages.len(),
|
||||
indexed_entries,
|
||||
complete: self.complete,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn validated_committed_entries(&self) -> Result<Vec<String>, RawEnumerationPageIndexError> {
|
||||
let mut entries = Vec::new();
|
||||
for (page_index, page) in self.pages.iter().enumerate() {
|
||||
page.validate(
|
||||
&self.parent,
|
||||
u64::try_from(page_index).unwrap_or(u64::MAX),
|
||||
u64::try_from(entries.len()).unwrap_or(u64::MAX),
|
||||
)?;
|
||||
if page.terminal && page_index + 1 != self.pages.len() {
|
||||
return Err(RawEnumerationPageIndexError::CorruptIndex);
|
||||
}
|
||||
entries.extend(page.entries.iter().cloned());
|
||||
}
|
||||
if self.complete && self.pages.last().is_some_and(|page| !page.terminal) {
|
||||
return Err(RawEnumerationPageIndexError::CorruptIndex);
|
||||
}
|
||||
Ok(entries)
|
||||
}
|
||||
|
||||
fn validated_indexed_entries(&self) -> Result<Vec<String>, RawEnumerationPageIndexError> {
|
||||
let mut entries = self.validated_committed_entries()?;
|
||||
if let Some(building) = &self.building {
|
||||
building.validate(
|
||||
u64::try_from(self.pages.len()).unwrap_or(u64::MAX),
|
||||
u64::try_from(entries.len()).unwrap_or(u64::MAX),
|
||||
self.page_entry_limit,
|
||||
)?;
|
||||
entries.extend(building.entries.iter().cloned());
|
||||
}
|
||||
Ok(entries)
|
||||
}
|
||||
|
||||
fn indexed_entries(&self) -> usize {
|
||||
self.pages.iter().map(|page| page.entries.len()).sum::<usize>()
|
||||
+ self.building.as_ref().map_or(0, |building| building.entries.len())
|
||||
}
|
||||
}
|
||||
|
||||
impl RawEnumerationPageBuilder {
|
||||
fn validate(&self, page_index: u64, entries_start: u64, page_entry_limit: usize) -> Result<(), RawEnumerationPageIndexError> {
|
||||
if self.page_index != page_index
|
||||
|| self.entries_start != entries_start
|
||||
|| self.entries.is_empty()
|
||||
|| self.entries.len() > page_entry_limit
|
||||
|| !entries_are_normalized(&self.entries)
|
||||
{
|
||||
return Err(RawEnumerationPageIndexError::CorruptIndex);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl RawEnumerationPage {
|
||||
pub fn version(&self) -> u16 {
|
||||
self.version
|
||||
}
|
||||
|
||||
pub fn parent(&self) -> &str {
|
||||
&self.parent
|
||||
}
|
||||
|
||||
pub fn page_index(&self) -> u64 {
|
||||
self.page_index
|
||||
}
|
||||
|
||||
pub fn entries_start(&self) -> u64 {
|
||||
self.entries_start
|
||||
}
|
||||
|
||||
pub fn entries(&self) -> &[String] {
|
||||
&self.entries
|
||||
}
|
||||
|
||||
pub fn terminal(&self) -> bool {
|
||||
self.terminal
|
||||
}
|
||||
|
||||
pub fn digest(&self) -> [u8; 32] {
|
||||
self.digest
|
||||
}
|
||||
|
||||
fn new(parent: &str, building: RawEnumerationPageBuilder) -> Self {
|
||||
let digest = raw_page_digest(parent, &building);
|
||||
Self {
|
||||
version: RAW_PAGE_INDEX_VERSION,
|
||||
parent: parent.to_string(),
|
||||
page_index: building.page_index,
|
||||
entries_start: building.entries_start,
|
||||
entries: building.entries,
|
||||
terminal: building.terminal,
|
||||
digest,
|
||||
}
|
||||
}
|
||||
|
||||
fn validate(&self, parent: &str, page_index: u64, entries_start: u64) -> Result<(), RawEnumerationPageIndexError> {
|
||||
let builder = RawEnumerationPageBuilder {
|
||||
page_index: self.page_index,
|
||||
entries_start: self.entries_start,
|
||||
entries: self.entries.clone(),
|
||||
terminal: self.terminal,
|
||||
};
|
||||
if self.version != RAW_PAGE_INDEX_VERSION
|
||||
|| self.parent != parent
|
||||
|| self.page_index != page_index
|
||||
|| self.entries_start != entries_start
|
||||
|| self.entries.is_empty()
|
||||
|| !entries_are_normalized(&self.entries)
|
||||
|| self.digest != raw_page_digest(parent, &builder)
|
||||
{
|
||||
return Err(RawEnumerationPageIndexError::CorruptIndex);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn normalize_owner_entries<I>(entries: I) -> Result<Vec<String>, RawEnumerationPageIndexError>
|
||||
where
|
||||
I: IntoIterator<Item = String>,
|
||||
{
|
||||
let mut entries = entries.into_iter().map(validate_owner_entry).collect::<Result<Vec<_>, _>>()?;
|
||||
entries.sort();
|
||||
entries.dedup();
|
||||
Ok(entries)
|
||||
}
|
||||
|
||||
fn validate_owner_entry(entry: String) -> Result<String, RawEnumerationPageIndexError> {
|
||||
if !owner_entry_is_valid(&entry) {
|
||||
return Err(RawEnumerationPageIndexError::InvalidEntry);
|
||||
}
|
||||
Ok(entry)
|
||||
}
|
||||
|
||||
fn owner_entry_is_valid(entry: &str) -> bool {
|
||||
!entry.is_empty() && entry != "." && entry != ".." && !entry.contains('/') && entry.len() <= RAW_PAGE_ENTRY_MAX_BYTES
|
||||
}
|
||||
|
||||
fn entries_are_normalized(entries: &[String]) -> bool {
|
||||
entries.iter().all(|entry| owner_entry_is_valid(entry))
|
||||
&& entries
|
||||
.windows(2)
|
||||
.all(|window| window.first().zip(window.get(1)).is_some_and(|(left, right)| left < right))
|
||||
}
|
||||
|
||||
fn raw_page_digest(parent: &str, building: &RawEnumerationPageBuilder) -> [u8; 32] {
|
||||
let mut digest = Sha256::new();
|
||||
update_digest(&mut digest, b"version", &RAW_PAGE_INDEX_VERSION.to_le_bytes());
|
||||
update_digest(&mut digest, b"parent", parent.as_bytes());
|
||||
update_digest(&mut digest, b"page_index", &building.page_index.to_le_bytes());
|
||||
update_digest(&mut digest, b"entries_start", &building.entries_start.to_le_bytes());
|
||||
update_digest(&mut digest, b"terminal", &[u8::from(building.terminal)]);
|
||||
for entry in &building.entries {
|
||||
update_digest(&mut digest, b"entry", entry.as_bytes());
|
||||
}
|
||||
digest.finalize().into()
|
||||
}
|
||||
|
||||
fn update_digest(digest: &mut Sha256, label: &[u8], value: &[u8]) {
|
||||
digest.update(label);
|
||||
digest.update(u64::try_from(value.len()).unwrap_or(u64::MAX).to_le_bytes());
|
||||
digest.update(value);
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn entries(values: &[&str]) -> Vec<String> {
|
||||
values.iter().map(|value| (*value).to_string()).collect()
|
||||
}
|
||||
|
||||
fn indexed_entries(status: &RawEnumerationPageOwnerStatus) -> u64 {
|
||||
match status {
|
||||
RawEnumerationPageOwnerStatus::Unsupported => 0,
|
||||
RawEnumerationPageOwnerStatus::Building { indexed_entries, .. }
|
||||
| RawEnumerationPageOwnerStatus::Ready { indexed_entries, .. } => *indexed_entries,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unsupported_owner_reports_unsupported_without_building_pages() {
|
||||
let mut owner = RawEnumerationPageIndex::unsupported();
|
||||
|
||||
assert_eq!(owner.status(), RawEnumerationPageOwnerStatus::Unsupported);
|
||||
assert_eq!(
|
||||
owner.ingest_owner_entries(entries(&["entry-a"]), 1, 0),
|
||||
Err(RawEnumerationPageIndexError::Unsupported)
|
||||
);
|
||||
assert_eq!(owner.commit_building_page(0), Err(RawEnumerationPageIndexError::Unsupported));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn owner_page_builds_monotonically_across_small_budget_restarts() {
|
||||
let source = entries(&["entry-c", "entry-a", "entry-e", "entry-b", "entry-d"]);
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");
|
||||
let mut last_indexed_entries = 0;
|
||||
|
||||
for _ in 0..8 {
|
||||
let mut restarted_owner = owner.clone();
|
||||
let generation = restarted_owner
|
||||
.generation()
|
||||
.expect("supported owner should expose generation");
|
||||
let outcome = restarted_owner
|
||||
.ingest_owner_entries(source.clone(), 1, generation)
|
||||
.expect("owner page build should accept stable source entries");
|
||||
let now_indexed_entries = indexed_entries(&outcome.status);
|
||||
assert!(
|
||||
now_indexed_entries >= last_indexed_entries,
|
||||
"owner-backed page build must not move coverage backward across restart"
|
||||
);
|
||||
last_indexed_entries = now_indexed_entries;
|
||||
if outcome.ready_to_commit {
|
||||
let generation = restarted_owner
|
||||
.generation()
|
||||
.expect("supported owner should expose generation");
|
||||
restarted_owner
|
||||
.commit_building_page(generation)
|
||||
.expect("ready page should commit under matching generation");
|
||||
}
|
||||
owner = restarted_owner;
|
||||
if let RawEnumerationPageOwnerStatus::Ready { complete: true, .. } = owner.status() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(
|
||||
owner.status(),
|
||||
RawEnumerationPageOwnerStatus::Ready {
|
||||
generation: 8,
|
||||
parent: "bucket".to_string(),
|
||||
committed_pages: 3,
|
||||
indexed_entries: 5,
|
||||
complete: true,
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
owner.committed_entries().expect("committed entries should validate"),
|
||||
entries(&["entry-a", "entry-b", "entry-c", "entry-d", "entry-e"])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn page_digest_identity_guards_consumption_and_source_drift() {
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");
|
||||
let source = entries(&["entry-a", "entry-b", "entry-c"]);
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let outcome = owner
|
||||
.ingest_owner_entries(source, 2, generation)
|
||||
.expect("first page build should succeed");
|
||||
assert!(outcome.ready_to_commit);
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let page = owner.commit_building_page(generation).expect("first page should commit");
|
||||
|
||||
assert_eq!(
|
||||
owner
|
||||
.page(page.page_index, page.digest)
|
||||
.expect("committed page should be readable"),
|
||||
&page
|
||||
);
|
||||
let mut wrong_digest = page.digest;
|
||||
wrong_digest[0] ^= 0xff;
|
||||
assert_eq!(
|
||||
owner.page(page.page_index, wrong_digest),
|
||||
Err(RawEnumerationPageIndexError::IdentityMismatch)
|
||||
);
|
||||
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
assert_eq!(
|
||||
owner.ingest_owner_entries(entries(&["entry-a", "entry-x", "entry-c"]), 1, generation),
|
||||
Err(RawEnumerationPageIndexError::IdentityMismatch)
|
||||
);
|
||||
assert_eq!(
|
||||
owner
|
||||
.committed_entries()
|
||||
.expect("committed entries should validate after source drift rejection"),
|
||||
entries(&["entry-a", "entry-b"])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn page_commit_cas_failure_and_precommit_crash_do_not_publish_coverage() {
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");
|
||||
let source = entries(&["entry-a", "entry-b", "entry-c"]);
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let outcome = owner
|
||||
.ingest_owner_entries(source, 2, generation)
|
||||
.expect("page build should stage entries");
|
||||
assert!(outcome.ready_to_commit);
|
||||
assert_eq!(
|
||||
owner
|
||||
.committed_entries()
|
||||
.expect("uncommitted staged entries should not publish coverage"),
|
||||
Vec::<String>::new()
|
||||
);
|
||||
|
||||
assert_eq!(owner.commit_building_page(generation), Err(RawEnumerationPageIndexError::StaleGeneration));
|
||||
assert_eq!(
|
||||
owner
|
||||
.committed_entries()
|
||||
.expect("failed CAS should not corrupt committed coverage"),
|
||||
Vec::<String>::new()
|
||||
);
|
||||
|
||||
let mut restarted_owner = owner.clone();
|
||||
let generation = restarted_owner
|
||||
.generation()
|
||||
.expect("supported owner should expose generation");
|
||||
restarted_owner
|
||||
.commit_building_page(generation)
|
||||
.expect("persisted staged page should commit after restart with fresh CAS generation");
|
||||
assert_eq!(
|
||||
restarted_owner
|
||||
.committed_entries()
|
||||
.expect("restarted committed entries should validate"),
|
||||
entries(&["entry-a", "entry-b"])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serialized_building_page_resumes_and_commits_after_restart() {
|
||||
let source = entries(&["entry-a", "entry-b", "entry-c"]);
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let outcome = owner
|
||||
.ingest_owner_entries(source.clone(), 1, generation)
|
||||
.expect("first budgeted page build should stage one entry");
|
||||
assert_eq!(
|
||||
outcome.status,
|
||||
RawEnumerationPageOwnerStatus::Building {
|
||||
generation: 1,
|
||||
parent: "bucket".to_string(),
|
||||
page_index: 0,
|
||||
indexed_entries: 1,
|
||||
buffered_entries: 1,
|
||||
}
|
||||
);
|
||||
assert!(!outcome.ready_to_commit);
|
||||
|
||||
let encoded = rmp_serde::to_vec(&owner).expect("building page index should encode");
|
||||
let mut decoded: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("building page index should decode");
|
||||
|
||||
let generation = decoded.generation().expect("decoded owner should expose generation");
|
||||
let outcome = decoded
|
||||
.ingest_owner_entries(source, 1, generation)
|
||||
.expect("decoded page owner should resume from staged coverage");
|
||||
assert!(outcome.ready_to_commit);
|
||||
let generation = decoded.generation().expect("decoded owner should expose generation");
|
||||
let page = decoded
|
||||
.commit_building_page(generation)
|
||||
.expect("decoded ready page should commit");
|
||||
assert_eq!(page.entries(), entries(&["entry-a", "entry-b"]));
|
||||
assert_eq!(
|
||||
decoded
|
||||
.page(page.page_index(), page.digest())
|
||||
.expect("committed decoded page should validate by digest")
|
||||
.entries(),
|
||||
entries(&["entry-a", "entry-b"])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn partial_owner_source_does_not_mark_terminal_before_completion() {
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let outcome = owner
|
||||
.ingest_partial_owner_entries(entries(&["entry-a"]), 1, generation)
|
||||
.expect("partial source should stage one entry");
|
||||
assert_eq!(
|
||||
outcome.status,
|
||||
RawEnumerationPageOwnerStatus::Building {
|
||||
generation: 1,
|
||||
parent: "bucket".to_string(),
|
||||
page_index: 0,
|
||||
indexed_entries: 1,
|
||||
buffered_entries: 1,
|
||||
}
|
||||
);
|
||||
assert!(!outcome.ready_to_commit);
|
||||
assert_eq!(
|
||||
owner.indexed_entries().expect("building page entries should validate"),
|
||||
entries(&["entry-a"])
|
||||
);
|
||||
|
||||
let encoded = rmp_serde::to_vec(&owner).expect("partial owner should encode");
|
||||
let mut restarted: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("partial owner should decode");
|
||||
let generation = restarted.generation().expect("restarted owner should expose generation");
|
||||
let outcome = restarted
|
||||
.ingest_owner_entries(entries(&["entry-a", "entry-b"]), 1, generation)
|
||||
.expect("complete source should finish resumed building page");
|
||||
assert!(outcome.ready_to_commit);
|
||||
let generation = restarted.generation().expect("finished owner should expose generation");
|
||||
let page = restarted
|
||||
.commit_building_page(generation)
|
||||
.expect("terminal resumed page should commit");
|
||||
assert!(page.terminal());
|
||||
assert_eq!(page.entries(), entries(&["entry-a", "entry-b"]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn terminal_marker_advances_generation_before_commit() {
|
||||
let initial_source = entries(&["entry-a", "entry-b", "entry-c"]);
|
||||
let current_source = entries(&["entry-a", "entry-b"]);
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 3).expect("page owner should initialize");
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let outcome = owner
|
||||
.ingest_owner_entries(initial_source, 2, generation)
|
||||
.expect("budgeted page build should stage all source entries");
|
||||
assert_eq!(owner.generation(), Some(1));
|
||||
assert!(!outcome.ready_to_commit);
|
||||
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let outcome = owner
|
||||
.ingest_owner_entries(current_source, 1, generation)
|
||||
.expect("terminal-only build step should complete the staged page");
|
||||
assert!(outcome.ready_to_commit);
|
||||
assert_eq!(
|
||||
owner.generation(),
|
||||
Some(2),
|
||||
"terminal marker is a persisted builder state change and must advance the CAS generation"
|
||||
);
|
||||
assert_eq!(owner.commit_building_page(generation), Err(RawEnumerationPageIndexError::StaleGeneration));
|
||||
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let page = owner
|
||||
.commit_building_page(generation)
|
||||
.expect("fresh generation should commit terminal page");
|
||||
assert!(page.terminal());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn deserialized_corrupt_page_digest_fails_closed() {
|
||||
let source = entries(&["entry-a", "entry-b", "entry-c"]);
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let outcome = owner
|
||||
.ingest_owner_entries(source.clone(), 2, generation)
|
||||
.expect("page build should stage entries");
|
||||
assert!(outcome.ready_to_commit);
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let page = owner.commit_building_page(generation).expect("ready page should commit");
|
||||
|
||||
let encoded = rmp_serde::to_vec(&owner).expect("committed page index should encode");
|
||||
let mut decoded: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("committed page index should decode");
|
||||
let RawEnumerationPageIndexState::Supported(inner) = &mut decoded.state else {
|
||||
panic!("decoded owner should be supported");
|
||||
};
|
||||
inner.pages[0].digest[0] ^= 0xff;
|
||||
|
||||
assert_eq!(decoded.committed_entries(), Err(RawEnumerationPageIndexError::CorruptIndex));
|
||||
assert_eq!(
|
||||
decoded.page(page.page_index(), page.digest()),
|
||||
Err(RawEnumerationPageIndexError::CorruptIndex)
|
||||
);
|
||||
assert_eq!(
|
||||
decoded.ingest_owner_entries(source, 1, decoded.generation().expect("decoded owner should expose generation")),
|
||||
Err(RawEnumerationPageIndexError::CorruptIndex)
|
||||
);
|
||||
|
||||
let mut complete_owner = RawEnumerationPageIndex::new("bucket", 2).expect("complete owner should initialize");
|
||||
let generation = complete_owner.generation().expect("complete owner should expose generation");
|
||||
let outcome = complete_owner
|
||||
.ingest_owner_entries(entries(&["entry-a", "entry-b"]), 2, generation)
|
||||
.expect("terminal page build should stage entries");
|
||||
assert!(outcome.ready_to_commit);
|
||||
let generation = complete_owner.generation().expect("complete owner should expose generation");
|
||||
complete_owner
|
||||
.commit_building_page(generation)
|
||||
.expect("terminal page should commit");
|
||||
let encoded = rmp_serde::to_vec(&complete_owner).expect("complete page index should encode");
|
||||
let mut decoded_complete: RawEnumerationPageIndex =
|
||||
rmp_serde::from_slice(&encoded).expect("complete page index should decode");
|
||||
let RawEnumerationPageIndexState::Supported(inner) = &mut decoded_complete.state else {
|
||||
panic!("decoded complete owner should be supported");
|
||||
};
|
||||
inner.pages[0].digest[0] ^= 0xff;
|
||||
assert_eq!(
|
||||
decoded_complete.ingest_owner_entries(
|
||||
entries(&["entry-a", "entry-b"]),
|
||||
1,
|
||||
decoded_complete
|
||||
.generation()
|
||||
.expect("decoded complete owner should expose generation")
|
||||
),
|
||||
Err(RawEnumerationPageIndexError::CorruptIndex)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn deserialized_corrupt_building_page_fails_closed_before_append() {
|
||||
let source = entries(&["entry-a", "entry-b", "entry-c"]);
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
owner
|
||||
.ingest_owner_entries(source.clone(), 1, generation)
|
||||
.expect("page build should stage one entry");
|
||||
|
||||
let encoded = rmp_serde::to_vec(&owner).expect("building page index should encode");
|
||||
let mut decoded: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("building page index should decode");
|
||||
let RawEnumerationPageIndexState::Supported(inner) = &mut decoded.state else {
|
||||
panic!("decoded building owner should be supported");
|
||||
};
|
||||
inner.building.as_mut().expect("building page should be present").page_index = 9;
|
||||
|
||||
assert_eq!(
|
||||
decoded.ingest_owner_entries(source, 1, decoded.generation().expect("decoded owner should expose generation")),
|
||||
Err(RawEnumerationPageIndexError::CorruptIndex)
|
||||
);
|
||||
assert_eq!(
|
||||
decoded.commit_building_page(decoded.generation().expect("decoded owner should expose generation")),
|
||||
Err(RawEnumerationPageIndexError::CorruptIndex)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_owner_source_becomes_ready_without_empty_page_commit() {
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let outcome = owner
|
||||
.ingest_owner_entries(Vec::new(), 1, generation)
|
||||
.expect("empty owner source should be a valid complete index");
|
||||
|
||||
assert_eq!(
|
||||
outcome.status,
|
||||
RawEnumerationPageOwnerStatus::Ready {
|
||||
generation: 1,
|
||||
parent: "bucket".to_string(),
|
||||
committed_pages: 0,
|
||||
indexed_entries: 0,
|
||||
complete: true,
|
||||
}
|
||||
);
|
||||
assert!(!outcome.ready_to_commit);
|
||||
assert_eq!(owner.commit_building_page(1), Err(RawEnumerationPageIndexError::EmptyCommit));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn complete_owner_rejects_source_drift_after_restart() {
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
let outcome = owner
|
||||
.ingest_owner_entries(entries(&["entry-a", "entry-b"]), 2, generation)
|
||||
.expect("terminal page build should stage entries");
|
||||
assert!(outcome.ready_to_commit);
|
||||
let generation = owner.generation().expect("supported owner should expose generation");
|
||||
owner.commit_building_page(generation).expect("terminal page should commit");
|
||||
|
||||
let encoded = rmp_serde::to_vec(&owner).expect("complete owner should encode");
|
||||
let mut restarted_owner: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("complete owner should decode");
|
||||
let generation = restarted_owner
|
||||
.generation()
|
||||
.expect("restarted owner should expose generation");
|
||||
|
||||
assert_eq!(
|
||||
restarted_owner.ingest_owner_entries(entries(&["entry-a", "entry-b", "entry-c"]), 1, generation),
|
||||
Err(RawEnumerationPageIndexError::IdentityMismatch)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn owner_page_rejects_invalid_boundaries_without_advancing_generation() {
|
||||
assert_eq!(RawEnumerationPageIndex::new("", 2), Err(RawEnumerationPageIndexError::EmptyParent));
|
||||
assert_eq!(RawEnumerationPageIndex::new("bucket", 0), Err(RawEnumerationPageIndexError::EmptyPage));
|
||||
|
||||
let mut owner = RawEnumerationPageIndex::new("bucket", 1).expect("page owner should initialize");
|
||||
assert_eq!(
|
||||
owner.ingest_owner_entries(entries(&["entry-a"]), 0, 0),
|
||||
Err(RawEnumerationPageIndexError::EmptyBudget)
|
||||
);
|
||||
assert_eq!(
|
||||
owner.ingest_owner_entries(entries(&["nested/name"]), 1, 0),
|
||||
Err(RawEnumerationPageIndexError::InvalidEntry)
|
||||
);
|
||||
assert_eq!(owner.generation(), Some(0));
|
||||
|
||||
let oversized = "x".repeat(RAW_PAGE_ENTRY_MAX_BYTES + 1);
|
||||
assert_eq!(
|
||||
owner.ingest_owner_entries(vec![oversized], 1, 0),
|
||||
Err(RawEnumerationPageIndexError::InvalidEntry)
|
||||
);
|
||||
assert_eq!(owner.generation(), Some(0));
|
||||
|
||||
let exact_boundary = "x".repeat(RAW_PAGE_ENTRY_MAX_BYTES);
|
||||
let outcome = owner
|
||||
.ingest_owner_entries(vec![exact_boundary.clone()], 1, 0)
|
||||
.expect("max-sized direct entry should be accepted");
|
||||
assert!(outcome.ready_to_commit);
|
||||
let page = owner.commit_building_page(1).expect("max-sized direct entry should commit");
|
||||
assert_eq!(page.entries(), &[exact_boundary]);
|
||||
}
|
||||
}
|
||||
@@ -25,6 +25,7 @@ use crate::data_usage_define::{
|
||||
PendingScannerHealKind, ScannerSizeSummaryExt, SizeReconciliationEntry, SizeSummary, hash_path,
|
||||
};
|
||||
use crate::error::ScannerError;
|
||||
use crate::raw_page_index::{RawEnumerationPageIndex, RawEnumerationPageIndexError};
|
||||
use crate::runtime_config::{
|
||||
scanner_alert_excess_folders, scanner_alert_excess_version_size, scanner_alert_excess_versions, scanner_yield_every_n_objects,
|
||||
};
|
||||
@@ -90,6 +91,8 @@ const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000;
|
||||
const SCANNER_LIST_PATH_RAW_STALL_TIMEOUT: Duration = Duration::from_secs(60);
|
||||
const SCANNER_ENTRY_PROGRESS_BATCH: u64 = 32;
|
||||
const SCANNER_ENTRY_PROGRESS_INTERVAL: Duration = Duration::from_secs(30);
|
||||
const SCANNER_RAW_ENUMERATION_PAGE_ENTRY_LIMIT: usize = 128;
|
||||
const SCANNER_RAW_ENUMERATION_PAGE_BUILD_BUDGET: usize = 1;
|
||||
// Erasure data directories contain direct part.N files; keep namespace probes bounded.
|
||||
const ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT: usize = 64;
|
||||
const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024;
|
||||
@@ -751,17 +754,34 @@ struct RawEnumerationProgress {
|
||||
last_entry: Option<String>,
|
||||
entries_seen: u64,
|
||||
digest: Sha256,
|
||||
observed_entries: Vec<String>,
|
||||
revalidate_after_entries: usize,
|
||||
page_index: Option<RawEnumerationPageIndex>,
|
||||
}
|
||||
|
||||
impl RawEnumerationProgress {
|
||||
fn new(parent: &str) -> Self {
|
||||
fn new(parent: &str, page_index: Option<RawEnumerationPageIndex>) -> Self {
|
||||
let mut digest = Sha256::new();
|
||||
update_raw_enumeration_digest(&mut digest, b"parent", parent.as_bytes());
|
||||
let mut revalidate_after_entries = 0;
|
||||
let page_index = match page_index {
|
||||
Some(index) => match index.indexed_entries() {
|
||||
Ok(entries) => {
|
||||
revalidate_after_entries = entries.len();
|
||||
Some(index)
|
||||
}
|
||||
Err(_) => None,
|
||||
},
|
||||
None => RawEnumerationPageIndex::new(parent, SCANNER_RAW_ENUMERATION_PAGE_ENTRY_LIMIT).ok(),
|
||||
};
|
||||
Self {
|
||||
parent: parent.to_string(),
|
||||
last_entry: None,
|
||||
entries_seen: 0,
|
||||
digest,
|
||||
observed_entries: Vec::new(),
|
||||
revalidate_after_entries,
|
||||
page_index,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -769,19 +789,55 @@ impl RawEnumerationProgress {
|
||||
update_raw_enumeration_digest(&mut self.digest, b"entry", entry.as_bytes());
|
||||
self.last_entry = Some(entry.to_string());
|
||||
self.entries_seen = self.entries_seen.saturating_add(1);
|
||||
self.observed_entries.push(entry.to_string());
|
||||
if let Some(index) = &mut self.page_index {
|
||||
if self.observed_entries.len() < self.revalidate_after_entries {
|
||||
return;
|
||||
}
|
||||
let result = index
|
||||
.generation()
|
||||
.ok_or(RawEnumerationPageIndexError::Unsupported)
|
||||
.and_then(|generation| {
|
||||
index.ingest_partial_owner_entries(
|
||||
self.observed_entries.clone(),
|
||||
SCANNER_RAW_ENUMERATION_PAGE_BUILD_BUDGET,
|
||||
generation,
|
||||
)
|
||||
});
|
||||
match result {
|
||||
Ok(outcome) if outcome.ready_to_commit => {
|
||||
if let Some(generation) = index.generation()
|
||||
&& index.commit_building_page(generation).is_err()
|
||||
{
|
||||
self.page_index = None;
|
||||
}
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(_) => {
|
||||
self.page_index = None;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn into_cursor(self) -> Option<DataUsageRawEnumerationCursor> {
|
||||
fn cursor(&self) -> Option<DataUsageRawEnumerationCursor> {
|
||||
if self.entries_seen == 0 {
|
||||
return None;
|
||||
}
|
||||
Some(DataUsageRawEnumerationCursor::new(
|
||||
self.parent,
|
||||
self.last_entry,
|
||||
self.parent.clone(),
|
||||
self.last_entry.clone(),
|
||||
self.entries_seen,
|
||||
self.digest.finalize().into(),
|
||||
self.digest.clone().finalize().into(),
|
||||
))
|
||||
}
|
||||
|
||||
fn page_index(&self) -> Option<RawEnumerationPageIndex> {
|
||||
self.page_index.clone().and_then(|index| match index.indexed_entries() {
|
||||
Ok(entries) if !entries.is_empty() => Some(index),
|
||||
_ => None,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn update_raw_enumeration_digest(digest: &mut Sha256, label: &[u8], value: &[u8]) {
|
||||
@@ -1049,6 +1105,19 @@ impl FolderScanner {
|
||||
if self.old_cache.info.scan_progress.is_none() {
|
||||
return;
|
||||
}
|
||||
let page_index = self
|
||||
.old_cache
|
||||
.validated_raw_enumeration_page_index()
|
||||
.filter(|index| match index.status() {
|
||||
crate::raw_page_index::RawEnumerationPageOwnerStatus::Building {
|
||||
parent: index_parent, ..
|
||||
}
|
||||
| crate::raw_page_index::RawEnumerationPageOwnerStatus::Ready {
|
||||
parent: index_parent, ..
|
||||
} => index_parent == parent,
|
||||
crate::raw_page_index::RawEnumerationPageOwnerStatus::Unsupported => false,
|
||||
})
|
||||
.cloned();
|
||||
if let Some(position) = self
|
||||
.raw_enumeration_progress
|
||||
.iter()
|
||||
@@ -1056,7 +1125,8 @@ impl FolderScanner {
|
||||
{
|
||||
self.raw_enumeration_progress.truncate(position + 1);
|
||||
} else {
|
||||
self.raw_enumeration_progress.push(RawEnumerationProgress::new(parent));
|
||||
self.raw_enumeration_progress
|
||||
.push(RawEnumerationProgress::new(parent, page_index));
|
||||
}
|
||||
if let Some(progress) = self.raw_enumeration_progress.last_mut() {
|
||||
progress.record_entry(entry);
|
||||
@@ -1073,11 +1143,11 @@ impl FolderScanner {
|
||||
});
|
||||
}
|
||||
|
||||
fn take_raw_enumeration_cursor(&mut self) -> Option<DataUsageRawEnumerationCursor> {
|
||||
self.raw_enumeration_progress
|
||||
.drain(..)
|
||||
.next()
|
||||
.and_then(RawEnumerationProgress::into_cursor)
|
||||
fn take_raw_enumeration_resume_state(&mut self) -> (Option<DataUsageRawEnumerationCursor>, Option<RawEnumerationPageIndex>) {
|
||||
match self.raw_enumeration_progress.drain(..).next() {
|
||||
Some(progress) => (progress.cursor(), progress.page_index()),
|
||||
None => (None, None),
|
||||
}
|
||||
}
|
||||
|
||||
fn carry_forward_old_children(&mut self, parent_hash: &DataUsageHash, entry: &mut DataUsageEntry) {
|
||||
@@ -2686,6 +2756,7 @@ pub(crate) async fn scan_data_folder_scoped(
|
||||
new_cache.info.scan_resume_after = None;
|
||||
new_cache.info.scan_checkpoint = None;
|
||||
new_cache.info.scan_raw_enumeration_cursor = None;
|
||||
new_cache.info.scan_raw_enumeration_page_index = None;
|
||||
new_cache.info.scan_coverage_receipt = None;
|
||||
if had_scan_checkpoint {
|
||||
global_metrics().record_scanner_checkpoint_cleared();
|
||||
@@ -2703,9 +2774,10 @@ pub(crate) async fn scan_data_folder_scoped(
|
||||
let root_hash = hash_path(&cache.info.name);
|
||||
let root_has_progress = data_usage_root_has_progress(&root);
|
||||
let pending_heals_changed = scanner.pending_heals_changed;
|
||||
let raw_enumeration_cursor = scanner.take_raw_enumeration_cursor();
|
||||
let carry_forward_cache =
|
||||
(raw_enumeration_cursor.is_some() && !root_has_progress).then(|| scanner.old_cache.cache.clone());
|
||||
let (raw_enumeration_cursor, raw_enumeration_page_index) = scanner.take_raw_enumeration_resume_state();
|
||||
let carry_forward_cache = ((raw_enumeration_cursor.is_some() || raw_enumeration_page_index.is_some())
|
||||
&& !root_has_progress)
|
||||
.then(|| scanner.old_cache.cache.clone());
|
||||
if root_has_progress {
|
||||
scanner.carry_forward_old_children(&root_hash, &mut root);
|
||||
}
|
||||
@@ -2722,8 +2794,15 @@ pub(crate) async fn scan_data_folder_scoped(
|
||||
new_cache.info.scan_resume_after = None;
|
||||
new_cache.info.scan_coverage_receipt = None;
|
||||
}
|
||||
if raw_enumeration_page_index.is_some() {
|
||||
new_cache.info.scan_raw_enumeration_page_index = raw_enumeration_page_index;
|
||||
new_cache.info.scan_checkpoint = None;
|
||||
new_cache.info.scan_resume_after = None;
|
||||
new_cache.info.scan_coverage_receipt = None;
|
||||
}
|
||||
if partial_cache_is_useful(&root, pending_heals_changed)
|
||||
|| new_cache.info.scan_raw_enumeration_cursor.is_some()
|
||||
|| new_cache.info.scan_raw_enumeration_page_index.is_some()
|
||||
|| !new_cache.info.size_reconciliation.is_empty()
|
||||
{
|
||||
if new_cache.root().is_some() {
|
||||
|
||||
@@ -2722,6 +2722,20 @@ async fn test_scan_data_folder_returns_raw_cursor_on_enumeration_cancel_without_
|
||||
assert!(raw_cursor.last_entry.is_some());
|
||||
assert_ne!(raw_cursor.page_digest, [0; 32]);
|
||||
assert_eq!(partial_cache.validated_raw_enumeration_cursor(), Some(raw_cursor));
|
||||
let page_index = partial_cache
|
||||
.validated_raw_enumeration_page_index()
|
||||
.expect("raw enumeration cancellation should persist a validated page index");
|
||||
assert_eq!(
|
||||
page_index
|
||||
.indexed_entries()
|
||||
.expect("persisted raw page index entries should validate")
|
||||
.len(),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
page_index.committed_entries().expect("uncommitted raw page should validate"),
|
||||
Vec::<String>::new()
|
||||
);
|
||||
assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Runtime));
|
||||
}
|
||||
|
||||
@@ -3370,3 +3384,48 @@ fn test_should_log_failed_object_samples_after_initial_limit() {
|
||||
assert!(!should_log_failed_object(SCANNER_FAILED_OBJECT_LOG_EVERY + 1));
|
||||
assert!(should_log_failed_object(SCANNER_FAILED_OBJECT_LOG_EVERY * 2));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn raw_enumeration_progress_waits_for_resume_index_floor_before_revalidation() {
|
||||
let mut index = RawEnumerationPageIndex::new("bucket", 2).expect("raw page index should initialize");
|
||||
let generation = index.generation().expect("raw page index should expose generation");
|
||||
index
|
||||
.ingest_partial_owner_entries(["entry-a".to_string(), "entry-b".to_string()], 2, generation)
|
||||
.expect("initial entries should build a page");
|
||||
let generation = index.generation().expect("raw page index should expose next generation");
|
||||
index.commit_building_page(generation).expect("initial page should commit");
|
||||
|
||||
let mut progress = RawEnumerationProgress::new("bucket", Some(index));
|
||||
progress.record_entry("entry-b");
|
||||
assert!(
|
||||
progress.page_index.is_some(),
|
||||
"resume index must not be dropped before the current run observes the old index floor"
|
||||
);
|
||||
|
||||
progress.record_entry("entry-a");
|
||||
assert!(
|
||||
progress.page_index.is_some(),
|
||||
"same entry identity after the observation floor should keep the resume index"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn raw_enumeration_progress_rejects_resume_index_after_floor_mismatch() {
|
||||
let mut index = RawEnumerationPageIndex::new("bucket", 2).expect("raw page index should initialize");
|
||||
let generation = index.generation().expect("raw page index should expose generation");
|
||||
index
|
||||
.ingest_partial_owner_entries(["entry-a".to_string(), "entry-b".to_string()], 2, generation)
|
||||
.expect("initial entries should build a page");
|
||||
let generation = index.generation().expect("raw page index should expose next generation");
|
||||
index.commit_building_page(generation).expect("initial page should commit");
|
||||
|
||||
let mut progress = RawEnumerationProgress::new("bucket", Some(index));
|
||||
progress.record_entry("entry-a");
|
||||
assert!(progress.page_index.is_some());
|
||||
|
||||
progress.record_entry("entry-c");
|
||||
assert!(
|
||||
progress.page_index.is_none(),
|
||||
"resume index must be discarded once enough current observations prove source drift"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
//! Fixture-only range diagnostics. No result is supplied to a scan selector.
|
||||
|
||||
use super::*;
|
||||
use crate::DATA_USAGE_CACHE_KEY_FORMAT;
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
const MAX_SEGMENTS: usize = 4;
|
||||
@@ -15,6 +16,89 @@ enum ProposalError {
|
||||
InvalidKey,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
enum ProducerKind {
|
||||
Put,
|
||||
Delete,
|
||||
DeleteMarker,
|
||||
Multipart,
|
||||
Replication,
|
||||
Tier,
|
||||
DirectoryObject,
|
||||
}
|
||||
|
||||
impl ProducerKind {
|
||||
const REQUIRED: [Self; 7] = [
|
||||
Self::Put,
|
||||
Self::Delete,
|
||||
Self::DeleteMarker,
|
||||
Self::Multipart,
|
||||
Self::Replication,
|
||||
Self::Tier,
|
||||
Self::DirectoryObject,
|
||||
];
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct SegmentObservationEnvelope<'a> {
|
||||
source: DataUsageCacheSource,
|
||||
bucket_incarnation: uuid::Uuid,
|
||||
key_format: u16,
|
||||
baseline_scan_plan_digest: DataUsageScanPlanDigest,
|
||||
process_epoch: &'a str,
|
||||
generation_start: u64,
|
||||
generation_end: u64,
|
||||
restart_gap: bool,
|
||||
overflow: bool,
|
||||
producers: BTreeSet<&'a str>,
|
||||
keys: &'a [&'a str],
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct SegmentObservationProof<'a> {
|
||||
source: DataUsageCacheSource,
|
||||
bucket_incarnation: uuid::Uuid,
|
||||
key_format: u16,
|
||||
baseline_scan_plan_digest: DataUsageScanPlanDigest,
|
||||
process_epoch: &'a str,
|
||||
}
|
||||
|
||||
fn producer_name(kind: ProducerKind) -> &'static str {
|
||||
match kind {
|
||||
ProducerKind::Put => "put",
|
||||
ProducerKind::Delete => "delete",
|
||||
ProducerKind::DeleteMarker => "delete_marker",
|
||||
ProducerKind::Multipart => "multipart",
|
||||
ProducerKind::Replication => "replication",
|
||||
ProducerKind::Tier => "tier",
|
||||
ProducerKind::DirectoryObject => "directory_object",
|
||||
}
|
||||
}
|
||||
|
||||
fn trusted_fixture_proposal(
|
||||
envelope: &SegmentObservationEnvelope<'_>,
|
||||
proof: &SegmentObservationProof<'_>,
|
||||
) -> Result<BTreeSet<String>, ProposalError> {
|
||||
if envelope.source != proof.source
|
||||
|| envelope.bucket_incarnation.is_nil()
|
||||
|| envelope.bucket_incarnation != proof.bucket_incarnation
|
||||
|| envelope.key_format != proof.key_format
|
||||
|| envelope.baseline_scan_plan_digest != proof.baseline_scan_plan_digest
|
||||
|| envelope.process_epoch != proof.process_epoch
|
||||
|| envelope.generation_start == 0
|
||||
|| envelope.generation_end < envelope.generation_start
|
||||
|| envelope.restart_gap
|
||||
|| envelope.overflow
|
||||
|| !ProducerKind::REQUIRED
|
||||
.iter()
|
||||
.all(|producer| envelope.producers.contains(producer_name(*producer)))
|
||||
{
|
||||
return Err(ProposalError::InvalidKey);
|
||||
}
|
||||
|
||||
fixture_proposal(envelope.keys)
|
||||
}
|
||||
|
||||
// Keys come from successful fixture writes, not a production mutation stream.
|
||||
fn fixture_proposal(keys: &[&str]) -> Result<BTreeSet<String>, ProposalError> {
|
||||
let mut segments = BTreeSet::new();
|
||||
@@ -54,6 +138,78 @@ fn segment_observation_fixture_proposal_bounds() {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn segment_observation_trusted_proposal_requires_identity_and_complete_producer_coverage() {
|
||||
let source = DataUsageCacheSource::new(2, 3);
|
||||
let incarnation = uuid::Uuid::from_u128(0x12345678123456781234567812345678);
|
||||
let baseline = DataUsageScanPlanDigest([9; 32]);
|
||||
let producers = ProducerKind::REQUIRED
|
||||
.iter()
|
||||
.map(|producer| producer_name(*producer))
|
||||
.collect::<BTreeSet<_>>();
|
||||
let envelope = SegmentObservationEnvelope {
|
||||
source,
|
||||
bucket_incarnation: incarnation,
|
||||
key_format: DATA_USAGE_CACHE_KEY_FORMAT,
|
||||
baseline_scan_plan_digest: baseline,
|
||||
process_epoch: "epoch-a",
|
||||
generation_start: 11,
|
||||
generation_end: 13,
|
||||
restart_gap: false,
|
||||
overflow: false,
|
||||
producers,
|
||||
keys: &["hot/one", "hot/two", "archive/delete-marker"],
|
||||
};
|
||||
let proof = SegmentObservationProof {
|
||||
source,
|
||||
bucket_incarnation: incarnation,
|
||||
key_format: DATA_USAGE_CACHE_KEY_FORMAT,
|
||||
baseline_scan_plan_digest: baseline,
|
||||
process_epoch: "epoch-a",
|
||||
};
|
||||
|
||||
assert_eq!(
|
||||
trusted_fixture_proposal(&envelope, &proof),
|
||||
Ok(BTreeSet::from(["archive".to_string(), "hot".to_string()]))
|
||||
);
|
||||
|
||||
let mut wrong_source = envelope.clone();
|
||||
wrong_source.source = DataUsageCacheSource::new(2, 4);
|
||||
assert_eq!(trusted_fixture_proposal(&wrong_source, &proof), Err(ProposalError::InvalidKey));
|
||||
|
||||
let mut missing_incarnation = envelope.clone();
|
||||
missing_incarnation.bucket_incarnation = uuid::Uuid::nil();
|
||||
assert_eq!(trusted_fixture_proposal(&missing_incarnation, &proof), Err(ProposalError::InvalidKey));
|
||||
|
||||
let mut wrong_key_format = envelope.clone();
|
||||
wrong_key_format.key_format = DATA_USAGE_CACHE_KEY_FORMAT.saturating_add(1);
|
||||
assert_eq!(trusted_fixture_proposal(&wrong_key_format, &proof), Err(ProposalError::InvalidKey));
|
||||
|
||||
let mut wrong_baseline = envelope.clone();
|
||||
wrong_baseline.baseline_scan_plan_digest = DataUsageScanPlanDigest([8; 32]);
|
||||
assert_eq!(trusted_fixture_proposal(&wrong_baseline, &proof), Err(ProposalError::InvalidKey));
|
||||
|
||||
let mut wrong_epoch = envelope.clone();
|
||||
wrong_epoch.process_epoch = "epoch-b";
|
||||
assert_eq!(trusted_fixture_proposal(&wrong_epoch, &proof), Err(ProposalError::InvalidKey));
|
||||
|
||||
let mut restart_gap = envelope.clone();
|
||||
restart_gap.restart_gap = true;
|
||||
assert_eq!(trusted_fixture_proposal(&restart_gap, &proof), Err(ProposalError::InvalidKey));
|
||||
|
||||
let mut overflow = envelope.clone();
|
||||
overflow.overflow = true;
|
||||
assert_eq!(trusted_fixture_proposal(&overflow, &proof), Err(ProposalError::InvalidKey));
|
||||
|
||||
let mut generation_gap = envelope.clone();
|
||||
generation_gap.generation_end = generation_gap.generation_start - 1;
|
||||
assert_eq!(trusted_fixture_proposal(&generation_gap, &proof), Err(ProposalError::InvalidKey));
|
||||
|
||||
let mut missing_producer = envelope.clone();
|
||||
missing_producer.producers.remove(producer_name(ProducerKind::Replication));
|
||||
assert_eq!(trusted_fixture_proposal(&missing_producer, &proof), Err(ProposalError::InvalidKey));
|
||||
}
|
||||
|
||||
fn cache_value(cache: &DataUsageCache) -> serde_json::Value {
|
||||
let mut value = serde_json::to_value(cache).expect("serialize the entire cache");
|
||||
// Children are a HashSet: canonicalize only that unordered field, without
|
||||
@@ -181,10 +337,6 @@ async fn walk_and_save(observe: bool) -> (Vec<String>, serde_json::Value) {
|
||||
2,
|
||||
"the two non-proposed segments must still be walked"
|
||||
);
|
||||
eprintln!(
|
||||
"segment fixture: proposed={proposed:?}, actual_segments={walked_segments:?}, actual_walk_callbacks={}, production_producer_coverage=unverified",
|
||||
paths.len()
|
||||
);
|
||||
} else {
|
||||
assert!(proposed_walked.lock().expect("read disabled observations").is_empty());
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user