Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions src/snapshot/error_wire.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
1 change: 1 addition & 0 deletions src/snapshot/fuse.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
};
Expand Down
74 changes: 71 additions & 3 deletions src/snapshot/fuse_owned_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
};
Expand Down Expand Up @@ -111,6 +111,7 @@ struct Fixture {
metadata: Mutex<Vec<String>>,
requests: AtomicUsize,
map_requests: AtomicUsize,
map_not_ready: AtomicUsize,
leaf_requests: AtomicUsize,
chunk_requests: AtomicUsize,
emitted: Arc<AtomicUsize>,
Expand Down Expand Up @@ -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)),
Expand Down Expand Up @@ -411,7 +413,7 @@ async fn raw(
async fn map_response(
HttpState(f): HttpState<Arc<Fixture>>,
Query(query): Query<BTreeMap<String, String>>,
) -> Json<Value> {
) -> Response {
f.map_requests.fetch_add(1, Ordering::SeqCst);
f.range_paths
.lock()
Expand All @@ -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<Arc<Fixture>>,
Expand Down Expand Up @@ -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;
Expand Down
4 changes: 4 additions & 0 deletions src/snapshot/reader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -332,6 +332,7 @@ fn lease_window(expiry: &str) -> Result<LeaseWindow, SnapshotError> {

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
Expand Down Expand Up @@ -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),
Expand Down
3 changes: 3 additions & 0 deletions src/snapshot/types.rs
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,8 @@ pub enum SnapshotErrorCode {
ScopeForbidden,
ViewNotFound,
SnapshotNotReady,
/// Fixed-view metadata is temporarily unavailable (METADATA_NOT_READY, 503).
MetadataNotReady,
Comment on lines +61 to +62

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve compatibility of the public error enum

When a downstream crate exhaustively matches SnapshotErrorCode—a usage explicitly supported by the existing API—adding MetadataNotReady makes that crate fail to compile after upgrading. Updating this repository's exhaustive-match test only masks the break for existing consumers. Keep the legacy projection as Internal and expose the typed value through the already #[non_exhaustive] CanonicalSnapshotErrorCode, or treat this as an explicitly breaking release.

Useful? React with 👍 / 👎.

/// The fixed view no longer exists (spec 14 §5 SNAPSHOT_GONE, 410).
SnapshotGone,
PathNotFound,
Expand Down Expand Up @@ -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,
Expand Down
120 changes: 114 additions & 6 deletions tests/mst2_canonical_reader_http.rs
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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};

Expand Down Expand Up @@ -62,6 +66,7 @@ struct Fixture {
caps: Mutex<Value>,
resolve_override: Mutex<Option<Value>>,
legacy_flat_map: Mutex<bool>,
map_not_ready: AtomicUsize,
calls: Mutex<Vec<(String, Value)>>,
descriptor: ServingDescriptor,
pages: BTreeMap<String, Vec<u8>>,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -314,9 +320,22 @@ async fn frames(
async fn map(
State(f): State<Arc<Fixture>>,
Query(query): Query<BTreeMap<String, String>>,
) -> Json<Value> {
) -> 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() {
Expand All @@ -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<Arc<Fixture>>,
Expand Down Expand Up @@ -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;
Expand Down
15 changes: 8 additions & 7 deletions tests/mst2_error_wire_http.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down Expand Up @@ -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));
}
Loading
Loading