From ee420ce10600dcdb18b019e2f84fdd74cd3dd0ca Mon Sep 17 00:00:00 2001 From: Luxian Date: Wed, 7 Oct 2026 13:57:12 +0800 Subject: [PATCH] fix: preserve metadata readiness errors in v3 readers --- src/snapshot/error_wire.rs | 3 +- src/snapshot/fuse.rs | 1 + src/snapshot/fuse_owned_tests.rs | 74 ++++++++++++++++- src/snapshot/reader.rs | 4 + src/snapshot/types.rs | 3 + tests/mst2_canonical_reader_http.rs | 120 ++++++++++++++++++++++++++-- tests/mst2_error_wire_http.rs | 15 ++-- tests/mst2_reader_lifecycle_http.rs | 4 +- 8 files changed, 204 insertions(+), 20 deletions(-) diff --git a/src/snapshot/error_wire.rs b/src/snapshot/error_wire.rs index 8cd41a68..6bf2b3f0 100644 --- a/src/snapshot/error_wire.rs +++ b/src/snapshot/error_wire.rs @@ -5,8 +5,7 @@ use serde_json::Value; use super::{SnapshotError, SnapshotErrorCode}; -/// SPEC error types for explicit canonical-envelope consumers. The original -/// `SnapshotErrorCode` remains source-compatible for exhaustive matches. +/// SPEC error types for explicit canonical-envelope consumers. #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[non_exhaustive] pub enum CanonicalSnapshotErrorCode { diff --git a/src/snapshot/fuse.rs b/src/snapshot/fuse.rs index 8a705b2c..28ce919a 100644 --- a/src/snapshot/fuse.rs +++ b/src/snapshot/fuse.rs @@ -2491,6 +2491,7 @@ fn io_err(e: crate::snapshot::SnapshotError) -> Errno { Unauthenticated | ScopeForbidden | LeaseUnknown => libc::EACCES, LeaseExpired | SnapshotGone => libc::ESTALE, NotDirectory => libc::ENOTDIR, + MetadataNotReady => libc::EAGAIN, DigestMismatch => libc::EIO, _ => libc::EIO, }; diff --git a/src/snapshot/fuse_owned_tests.rs b/src/snapshot/fuse_owned_tests.rs index 500239c6..2cd54887 100644 --- a/src/snapshot/fuse_owned_tests.rs +++ b/src/snapshot/fuse_owned_tests.rs @@ -12,7 +12,7 @@ use axum::{ body::Body, extract::{Query, State as HttpState}, http::StatusCode, - response::Response, + response::{IntoResponse, Response}, routing::{get, post}, Json, Router, }; @@ -111,6 +111,7 @@ struct Fixture { metadata: Mutex>, requests: AtomicUsize, map_requests: AtomicUsize, + map_not_ready: AtomicUsize, leaf_requests: AtomicUsize, chunk_requests: AtomicUsize, emitted: Arc, @@ -168,6 +169,7 @@ impl Fixture { metadata: Mutex::new(Vec::new()), requests: AtomicUsize::new(0), map_requests: AtomicUsize::new(0), + map_not_ready: AtomicUsize::new(0), leaf_requests: AtomicUsize::new(0), chunk_requests: AtomicUsize::new(0), emitted: Arc::new(AtomicUsize::new(0)), @@ -411,7 +413,7 @@ async fn raw( async fn map_response( HttpState(f): HttpState>, Query(query): Query>, -) -> Json { +) -> Response { f.map_requests.fetch_add(1, Ordering::SeqCst); f.range_paths .lock() @@ -420,11 +422,23 @@ async fn map_response( let map = &f.large.as_ref().unwrap().map; assert!(query["path"].trim_start_matches('/').starts_with("range")); assert_eq!(query["expected_digest"], id(&map.file_content_id)); + if f.map_not_ready + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| { + remaining.checked_sub(1) + }) + .is_ok() + { + return ( + StatusCode::SERVICE_UNAVAILABLE, + Json(json!({"error":{"code":"METADATA_NOT_READY","message":"fixed map is being prepared","request_id":"fuse-map","retryable":false}})), + ) + .into_response(); + } let mut value = json!({"snapshot_id":id(&f.descriptor.snapshot_id().unwrap()),"path":query["path"],"schema_version":2,"file_content_id":id(&map.file_content_id),"file_size":map.file_size.to_string(),"chunk_size":CHUNK_SIZE,"chunk_count":map.chunk_count.to_string(),"page_count":map.page_count.to_string(),"pages_root":id(&map.pages_root),"map_id":id(&map.map_id())}); if f.mode.load(Ordering::SeqCst) == 5 { value["file_size"] = json!((map.file_size + 1).to_string()); } - Json(value) + Json(value).into_response() } async fn leaf_response( HttpState(f): HttpState>, @@ -1797,6 +1811,60 @@ async fn actual_range_reply_retention_blocks_admission_until_the_last_bytes_drop assert_eq!(server.reader.content_usage().output_bytes, 0); } +#[tokio::test] +async fn actual_metadata_not_ready_returns_eagain_without_publishing_or_latching_a_range() { + let _serial = TEST_LOCK.lock().await; + let fixture = Fixture::new(false, true).with_large(); + fixture.map_not_ready.store(1, Ordering::SeqCst); + let server = Server::start(fixture, 8 * 1024 * 1024).await; + let fs = server.view(false).await; + let file = inode(&fs, "range000").await; + let baseline = server.reader.content_usage(); + assert_eq!( + i32::from( + fs.read(Request::default(), file, file, CHUNK_SIZE as u64 - 2, 5) + .await + .unwrap_err() + ), + -libc::EAGAIN + ); + idle(&server.reader).await; + assert_eq!(server.fixture.map_requests.load(Ordering::SeqCst), 1); + assert_eq!(server.fixture.leaf_requests.load(Ordering::SeqCst), 0); + assert_eq!(server.fixture.chunk_requests.load(Ordering::SeqCst), 0); + assert_eq!(server.fixture.requests.load(Ordering::SeqCst), 0); + assert_eq!(server.reader.content_usage(), baseline); + assert_eq!( + fs.state + .lock() + .unwrap() + .owned + .as_ref() + .unwrap() + .ranges + .len(), + 0 + ); + + server.fixture.map_not_ready.store(0, Ordering::SeqCst); + let reply = fs + .read(Request::default(), file, file, CHUNK_SIZE as u64 - 2, 5) + .await + .unwrap(); + assert_eq!(reply.data.as_ref(), [0, 0, 1, 1, 1]); + assert_eq!(server.fixture.map_requests.load(Ordering::SeqCst), 2); + assert_eq!(server.fixture.leaf_requests.load(Ordering::SeqCst), 1); + assert_eq!(server.fixture.chunk_requests.load(Ordering::SeqCst), 2); + assert_eq!( + server.reader.snapshot_id(), + id(&server.fixture.descriptor.snapshot_id().unwrap()) + ); + drop(reply); + drop(fs); + idle(&server.reader).await; + assert_eq!(server.reader.content_usage().output_bytes, 0); +} + #[tokio::test] async fn malformed_actual_range_map_chunk_and_end_publish_no_reply_or_reader_cache() { let _serial = TEST_LOCK.lock().await; diff --git a/src/snapshot/reader.rs b/src/snapshot/reader.rs index 151d601f..e0fcabbf 100644 --- a/src/snapshot/reader.rs +++ b/src/snapshot/reader.rs @@ -332,6 +332,7 @@ fn lease_window(expiry: &str) -> Result { fn recoverable_renewal_error(error: &SnapshotError) -> bool { match error.code { + SnapshotErrorCode::MetadataNotReady => error.http_status == 503, SnapshotErrorCode::TemporaryUnavailable if error.http_status == 0 => true, SnapshotErrorCode::TemporaryUnavailable | SnapshotErrorCode::Internal @@ -1753,6 +1754,9 @@ mod tests { (SnapshotErrorCode::Internal, 503, true), (SnapshotErrorCode::Internal, 429, true), (SnapshotErrorCode::SnapshotNotReady, 503, true), + (SnapshotErrorCode::MetadataNotReady, 503, true), + (SnapshotErrorCode::MetadataNotReady, 0, false), + (SnapshotErrorCode::MetadataNotReady, 403, false), (SnapshotErrorCode::Internal, 0, false), (SnapshotErrorCode::TemporaryUnavailable, 403, false), (SnapshotErrorCode::IntegrityError, 503, false), diff --git a/src/snapshot/types.rs b/src/snapshot/types.rs index a9fb6ac0..b5c0891f 100644 --- a/src/snapshot/types.rs +++ b/src/snapshot/types.rs @@ -58,6 +58,8 @@ pub enum SnapshotErrorCode { ScopeForbidden, ViewNotFound, SnapshotNotReady, + /// Fixed-view metadata is temporarily unavailable (METADATA_NOT_READY, 503). + MetadataNotReady, /// The fixed view no longer exists (spec 14 ยง5 SNAPSHOT_GONE, 410). SnapshotGone, PathNotFound, @@ -111,6 +113,7 @@ impl SnapshotErrorCode { "SCOPE_FORBIDDEN" => Self::ScopeForbidden, "VIEW_NOT_FOUND" => Self::ViewNotFound, "SNAPSHOT_NOT_READY" => Self::SnapshotNotReady, + "METADATA_NOT_READY" => Self::MetadataNotReady, "SNAPSHOT_GONE" => Self::SnapshotGone, "PATH_NOT_FOUND" => Self::PathNotFound, "NOT_DIRECTORY" => Self::NotDirectory, diff --git a/tests/mst2_canonical_reader_http.rs b/tests/mst2_canonical_reader_http.rs index 3b9047cd..c4a58300 100644 --- a/tests/mst2_canonical_reader_http.rs +++ b/tests/mst2_canonical_reader_http.rs @@ -1,13 +1,17 @@ //! Canonical discovery must drive the real fixed reader and both transports. use std::{ collections::BTreeMap, - sync::{Arc, Mutex}, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, + }, + time::Duration, }; use axum::{ body::{Body, Bytes}, extract::{Path, Query, State}, - http::HeaderMap, + http::{HeaderMap, StatusCode}, response::{IntoResponse, Response}, routing::{get, post}, Json, Router, @@ -22,8 +26,8 @@ use scorpiofs::snapshot::{ capabilities::CapabilityAdvertisement, durable::digest_of, frames::{parse_digest, MetadataPageItem}, - ChunkedFile, CompletionKind, DurableStore, LocalPinState, Mst2Client, ResolveDelivery, - ResolveRequest, ResolveTarget, ScopeCache, SnapshotErrorCode, SnapshotReader, + ChunkedFile, CompletionKind, DurableStore, LocalPinState, Mst2Client, OwnedChunkedFile, + ResolveDelivery, ResolveRequest, ResolveTarget, ScopeCache, SnapshotErrorCode, SnapshotReader, }; use serde_json::{json, Value}; @@ -62,6 +66,7 @@ struct Fixture { caps: Mutex, resolve_override: Mutex>, legacy_flat_map: Mutex, + map_not_ready: AtomicUsize, calls: Mutex>, descriptor: ServingDescriptor, pages: BTreeMap>, @@ -137,6 +142,7 @@ impl Fixture { caps: Mutex::new(caps), resolve_override: Mutex::new(None), legacy_flat_map: Mutex::new(false), + map_not_ready: AtomicUsize::new(0), calls: Mutex::new(vec![]), descriptor, pages, @@ -314,9 +320,22 @@ async fn frames( async fn map( State(f): State>, Query(query): Query>, -) -> Json { +) -> Response { assert_eq!(query["path"].trim_start_matches('/'), "large"); + assert_eq!(query["expected_digest"], id(&f.map.file_content_id)); f.record("chunk-map", json!(query)); + if f.map_not_ready + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| { + remaining.checked_sub(1) + }) + .is_ok() + { + return ( + StatusCode::SERVICE_UNAVAILABLE, + Json(json!({"error":{"code":"METADATA_NOT_READY","message":"fixed map is being prepared","request_id":"canonical-map","retryable":true}})), + ) + .into_response(); + } let descriptor = json!({"schema_version":2,"file_content_id":id(&f.map.file_content_id),"file_size":f.map.file_size.to_string(),"chunk_size":CHUNK_SIZE,"chunk_count":f.map.chunk_count.to_string(),"page_count":"1","pages_root":id(&f.map.pages_root),"map_id":id(&f.map.map_id())}); let mut body = json!({"snapshot_id":f.sid(),"path":query["path"],"map":descriptor}); if *f.legacy_flat_map.lock().unwrap() { @@ -325,7 +344,7 @@ async fn map( .unwrap() .extend(descriptor.as_object().unwrap().clone()); } - Json(body) + Json(body).into_response() } async fn leaf( State(f): State>, @@ -536,6 +555,95 @@ async fn canonical_ranges_use_page_index_and_split_actual_chunks_by_advertised_b assert_eq!(s.calls("resolve").len(), 1); } +#[tokio::test] +async fn canonical_owned_range_recovers_metadata_not_ready_without_resolving_another_view() { + let fixture = Fixture::new(); + fixture.map_not_ready.store(1, Ordering::SeqCst); + let s = Server::start(fixture).await; + let reader = s.reader().await; + let closure = reader.snapshot_closure().await.unwrap(); + reader.seed_content_membership(&closure).unwrap(); + let file = closure + .files() + .iter() + .find(|file| file.rel_path == "large") + .unwrap(); + let baseline = reader.content_usage(); + let range = OwnedChunkedFile::open(&reader, "/large", &file.content_digest, file.size) + .await + .unwrap(); + let bytes = range + .read_range_owned(CHUNK_SIZE as u64 - 2, 6) + .await + .unwrap(); + assert_eq!(bytes.as_bytes(), [0x51; 6]); + assert_eq!(s.calls("chunk-map").len(), 2); + assert_eq!(s.client.retry_count(), 1); + assert_eq!(s.calls("chunk-map/pages").len(), 1); + assert_eq!(s.calls("chunks").len(), 2); + assert_eq!(s.calls("resolve").len(), 1); + assert_eq!(reader.snapshot_id(), s.f.sid()); + assert_eq!(reader.lease_id(), "canonical-lease"); + drop(bytes); + drop(range); + assert_eq!(reader.content_usage(), baseline); +} + +#[tokio::test] +async fn canonical_owned_range_metadata_not_ready_stops_at_the_attempt_budget_and_can_recover() { + let fixture = Fixture::new(); + fixture.map_not_ready.store(usize::MAX, Ordering::SeqCst); + let mut s = Server::start(fixture).await; + s.client = s + .client + .clone() + .with_request_timeout(Duration::from_secs(3)); + let reader = s.reader().await; + let closure = reader.snapshot_closure().await.unwrap(); + reader.seed_content_membership(&closure).unwrap(); + let file = closure + .files() + .iter() + .find(|file| file.rel_path == "large") + .unwrap(); + let baseline = reader.content_usage(); + let error = tokio::time::timeout( + Duration::from_secs(5), + OwnedChunkedFile::open(&reader, "/large", &file.content_digest, file.size), + ) + .await + .unwrap() + .err() + .unwrap(); + assert_eq!(error.code, SnapshotErrorCode::MetadataNotReady); + assert_eq!(error.http_status, 503); + assert_eq!(s.calls("chunk-map").len(), 4); + assert_eq!(s.client.retry_count(), 3); + assert!(s.calls("chunk-map/pages").is_empty()); + assert!(s.calls("chunks").is_empty()); + assert_eq!(reader.content_usage(), baseline); + + s.f.map_not_ready.store(0, Ordering::SeqCst); + let range = OwnedChunkedFile::open(&reader, "/large", &file.content_digest, file.size) + .await + .unwrap(); + let bytes = range + .read_range_owned(2 * CHUNK_SIZE as u64, 7) + .await + .unwrap(); + assert_eq!(bytes.as_bytes(), [0x51; 7]); + assert_eq!(s.calls("chunk-map").len(), 5); + assert_eq!(s.client.retry_count(), 3); + assert_eq!(s.calls("chunk-map/pages").len(), 1); + assert_eq!(s.calls("chunks").len(), 1); + assert_eq!(s.calls("resolve").len(), 1); + assert_eq!(reader.snapshot_id(), s.f.sid()); + assert_eq!(reader.lease_id(), "canonical-lease"); + drop(bytes); + drop(range); + assert_eq!(reader.content_usage(), baseline); +} + #[tokio::test] async fn typed_view_and_lazy_delivery_are_sent_and_bound_to_the_fixed_reader() { let s = Server::start(Fixture::new()).await; diff --git a/tests/mst2_error_wire_http.rs b/tests/mst2_error_wire_http.rs index 5f6e463e..db300084 100644 --- a/tests/mst2_error_wire_http.rs +++ b/tests/mst2_error_wire_http.rs @@ -57,7 +57,7 @@ async fn every_spec_error_preserves_legacy_fallback_and_actual_http_status() { ("LEASE_EXPIRED", 410, LeaseExpired), ("SNAPSHOT_GONE", 410, SnapshotGone), ("SNAPSHOT_NOT_READY", 503, SnapshotNotReady), - ("METADATA_NOT_READY", 503, Internal), + ("METADATA_NOT_READY", 503, MetadataNotReady), ("OBJECT_UNAVAILABLE", 503, ObjectUnavailable), ("INTEGRITY_ERROR", 502, IntegrityError), ("CURSOR_STALE", 409, CursorStale), @@ -308,17 +308,18 @@ fn explicit_canonical_parser_retains_all_spec_error_types() { } #[test] -fn original_public_error_enum_still_supports_exhaustive_external_matches() { +fn public_error_enum_supports_exhaustive_external_matches() { fn classify(code: SnapshotErrorCode) -> bool { use SnapshotErrorCode::*; match code { ScopeInvalid | InvalidRequest | LimitExceeded | Unauthenticated | ScopeForbidden - | ViewNotFound | SnapshotNotReady | SnapshotGone | PathNotFound | NotDirectory - | UnsupportedEntry | LeaseUnknown | LeaseExpired | CursorInvalid | CursorStale - | ProofBudgetExceeded | DigestMismatch | IntegrityError | ObjectUnavailable - | RangeNotSupported | SymlinkTraversal | DurableViewConflict | TemporaryUnavailable - | Internal => true, + | ViewNotFound | SnapshotNotReady | MetadataNotReady | SnapshotGone | PathNotFound + | NotDirectory | UnsupportedEntry | LeaseUnknown | LeaseExpired | CursorInvalid + | CursorStale | ProofBudgetExceeded | DigestMismatch | IntegrityError + | ObjectUnavailable | RangeNotSupported | SymlinkTraversal | DurableViewConflict + | TemporaryUnavailable | Internal => true, } } assert!(classify(SnapshotErrorCode::Internal)); + assert!(classify(SnapshotErrorCode::MetadataNotReady)); } diff --git a/tests/mst2_reader_lifecycle_http.rs b/tests/mst2_reader_lifecycle_http.rs index 0b30d7a2..eb03d5ba 100644 --- a/tests/mst2_reader_lifecycle_http.rs +++ b/tests/mst2_reader_lifecycle_http.rs @@ -866,7 +866,7 @@ async fn typed_rate_limit_and_metadata_outages_recover_without_changing_the_fixe canonical_renewal_failure( "METADATA_NOT_READY", StatusCode::SERVICE_UNAVAILABLE, - SnapshotErrorCode::Internal, + SnapshotErrorCode::MetadataNotReady, true ) ); @@ -884,7 +884,7 @@ async fn typed_rate_limit_and_metadata_outages_cannot_extend_or_revive_the_origi canonical_renewal_failure( "METADATA_NOT_READY", StatusCode::SERVICE_UNAVAILABLE, - SnapshotErrorCode::Internal, + SnapshotErrorCode::MetadataNotReady, false ) );