diff --git a/src/infrastructure/services/azure_blob_backend.rs b/src/infrastructure/services/azure_blob_backend.rs index 2188a604..03feb291 100644 --- a/src/infrastructure/services/azure_blob_backend.rs +++ b/src/infrastructure/services/azure_blob_backend.rs @@ -239,11 +239,7 @@ impl BlobStorageBackend for AzureBlobBackend { let first = match pages.next().await { Some(Ok(response)) => response, Some(Err(e)) => { - return Err(DomainError::new( - ErrorKind::NotFound, - "Azure", - format!("Failed to get blob {hash}: {e}"), - )); + return Err(azure_read_error(format!("Failed to get blob {hash}"), &e)); } None => { let empty: BlobStream = @@ -324,10 +320,9 @@ impl BlobStorageBackend for AzureBlobBackend { let first = match pages.next().await { Some(Ok(response)) => response, Some(Err(e)) => { - return Err(DomainError::new( - ErrorKind::NotFound, - "Azure", - format!("Failed to get blob range {hash}: {e}"), + return Err(azure_read_error( + format!("Failed to get blob range {hash}"), + &e, )); } None => { @@ -415,13 +410,10 @@ impl BlobStorageBackend for AzureBlobBackend { let hash = hash.to_owned(); Box::pin(async move { let client = self.blob_client(&hash); - let props = client.get_properties().await.map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "Azure", - format!("Failed to stat blob {hash}: {e}"), - ) - })?; + let props = client + .get_properties() + .await + .map_err(|e| azure_read_error(format!("Failed to stat blob {hash}"), &e))?; Ok(props.blob.properties.content_length) }) } @@ -683,6 +675,33 @@ impl BlobStorageBackend for AzureBlobBackend { /// second wearing the clothes of the first. So the policy is to retry /// as if transient and let the bounded attempt cap turn the difference /// into a Paused run an operator can act on. +/// Read-path variant of [`azure_domain_error`]: only a real 404 is +/// `NotFound`. +/// +/// Every Azure read used to label EVERY failure `NotFound` — a refused +/// connection, a 503, an expired SAS token all reported as "blob +/// missing". That is the most dangerous wrong answer available on a read +/// path, because callers ACT on NotFound by concluding the bytes are +/// gone: a migration reading its source would treat an outage as "the +/// source does not have this blob" and move past it. +/// +/// Everything that is not a 404 goes through the normal classifier, so a +/// 403 stays permanent instead of being retried. +pub(crate) fn azure_read_error(context: String, err: &azure_core::Error) -> DomainError { + use azure_core::error::ErrorKind as AzKind; + + if let AzKind::HttpResponse { status, .. } = err.kind() + && u16::from(*status) == 404 + { + return DomainError::new( + ErrorKind::NotFound, + "Azure", + format!("{context}: not found"), + ); + } + azure_domain_error(context, err) +} + pub(crate) fn azure_domain_error(context: String, err: &azure_core::Error) -> DomainError { use azure_core::error::ErrorKind as AzKind; diff --git a/src/infrastructure/services/backend_migration_service.rs b/src/infrastructure/services/backend_migration_service.rs index 6edd1950..3c6fb506 100644 --- a/src/infrastructure/services/backend_migration_service.rs +++ b/src/infrastructure/services/backend_migration_service.rs @@ -690,21 +690,72 @@ impl RecoverableJobHandler for BackendMigrationService { .await; continue; } + Err(e) if e.is_transient() => { + // PAUSE. Skipping here was a data-loss path. + // + // The old comment called this "a network blip" + // and `continue`d, reasoning that a re-run would + // re-probe. It would not: the cursor advances to + // the batch's last hash regardless, so a skipped + // row is never revisited by THIS run — and unlike + // a copy failure it recorded no finding, so + // `failed` stayed 0, the run reached + // `finish_completed`, and the pointer flipped to + // a target missing every blob the outage + // covered. + // + // That is the worst shape available: a migration + // reporting success while having silently + // dropped whatever was unreachable at the time. + tracing::warn!( + target: "oxicloud::migration", + event = "backend_migration.source_unreachable", + run_id = %store.run_id(), + hash = %hash, + copied = copied_count, + error = %e, + "source unreachable while probing; pausing at the last checkpoint" + ); + return RunOutcome::from_domain_error( + cursor.as_ref().map(|s| s.as_bytes()), + &format!( + "source unreachable while probing ({copied_count} blob(s) \ + copied so far)" + ), + &e, + ); + } Err(e) => { - // Transient probe failure on source is NOT a - // finding — treat like a network blip. - // Skipping this row on this run; a re-run - // will re-probe. If the failure is - // persistent, `blobs_consistency` catches - // it. + // Permanent probe failure. Still skipped rather + // than fatal — one unprobeable blob must not + // abort the migration — but it now records a + // finding, so the run cannot report clean while + // having skipped rows, and `blobs_consistency` + // is not the only thing that would ever notice. tracing::warn!( target: "oxicloud::migration", event = "backend_migration.source_probe_error", run_id = %store.run_id(), hash = %hash, error = %e, - "source blob_exists probe failed; skipping this row" + "source blob_exists probe failed; recording finding, skipping row" ); + failed_count += 1; + record_or_log( + store, + BACKEND_MIGRATION_JOB_NAME, + "migration_failed", + "data_loss", + None, + serde_json::json!({ + "hash": hash, + "size": size, + "source": source_kind, + "target": target_kind, + "error": format!("source probe failed: {e}"), + }), + ) + .await; continue; } } diff --git a/src/infrastructure/services/local_blob_backend.rs b/src/infrastructure/services/local_blob_backend.rs index 4f0fac15..d089f1ad 100644 --- a/src/infrastructure/services/local_blob_backend.rs +++ b/src/infrastructure/services/local_blob_backend.rs @@ -615,13 +615,11 @@ impl BlobStorageBackend for LocalBlobBackend { let hash = hash.to_owned(); Box::pin(async move { let blob_path = self.blob_path(&hash); - let file = File::open(&blob_path).await.map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "Blob", - format!("Failed to open blob {}: {}", hash, e), - ) - })?; + // Was unconditional NotFound: a stale NFS handle or an + // unmounted iSCSI target reported the blob as missing. + let file = File::open(&blob_path) + .await + .map_err(|e| local_io_error("Blob", format!("Failed to open blob {hash}"), &e))?; Ok(Box::pin(ReaderStream::with_capacity(file, STREAM_CHUNK_SIZE)) as BlobStream) }) } @@ -636,13 +634,9 @@ impl BlobStorageBackend for LocalBlobBackend { let hash = hash.to_owned(); Box::pin(async move { let blob_path = self.blob_path(&hash); - let mut file = File::open(&blob_path).await.map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "Blob", - format!("Failed to open blob {}: {}", hash, e), - ) - })?; + let mut file = File::open(&blob_path) + .await + .map_err(|e| local_io_error("Blob", format!("Failed to open blob {hash}"), &e))?; file.seek(std::io::SeekFrom::Start(start)) .await @@ -697,13 +691,9 @@ impl BlobStorageBackend for LocalBlobBackend { let hash = hash.to_owned(); Box::pin(async move { let blob_path = self.blob_path(&hash); - let meta = fs::metadata(&blob_path).await.map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "Blob", - format!("Failed to stat blob {}: {}", hash, e), - ) - })?; + let meta = fs::metadata(&blob_path) + .await + .map_err(|e| local_io_error("Blob", format!("Failed to stat blob {hash}"), &e))?; Ok(meta.len()) }) } @@ -918,6 +908,60 @@ impl BlobStorageBackend for LocalBlobBackend { } } +/// Classify a filesystem error, because "local" does not mean +/// "reliable". +/// +/// A local backend is a PATH, and that path may be an iSCSI or NVMe-oF +/// LUN, an NFS mount, or a disk with a failing sector. Those produce +/// errors that clear on their own exactly like a remote 503 does, and +/// treating every one as permanent means a migration off a briefly +/// unreachable mount records data-loss findings for blobs that are +/// perfectly intact. +/// +/// It matters more here than for a remote backend, because +/// `RetryBlobBackend` is only applied when the active backend is NOT +/// Local (`di.rs`) — so nothing below this retries, and this +/// classification is the only thing standing between a flaky mount and +/// a run that concludes the data is gone. +/// +/// **`NotFound` stays `NotFound`, and nothing else becomes it.** Callers +/// act on that variant by concluding the bytes do not exist. +/// +/// Transient: the network-mount family (timeouts, unreachable, reset, +/// stale handle) plus `Interrupted` (EINTR) and `ResourceBusy` (EBUSY). +/// +/// Permanent, deliberately: `PermissionDenied` and +/// `ReadOnlyFilesystem` need an operator, retrying changes nothing. +/// `StorageFull` likewise. `InvalidData` is corruption, which is a +/// finding worth keeping. A bad sector surfaces as an uncategorised EIO +/// and therefore lands here too — right, because the useful outcome is +/// a `blob_corrupted`-style finding naming the blob, not a run that +/// pauses forever waiting for a disk to heal. +pub(crate) fn local_io_error( + entity: &'static str, + context: String, + err: &std::io::Error, +) -> DomainError { + use std::io::ErrorKind as Io; + + let message = format!("{context}: {err}"); + match err.kind() { + Io::NotFound => DomainError::new(ErrorKind::NotFound, entity, message), + Io::TimedOut + | Io::HostUnreachable + | Io::NetworkUnreachable + | Io::NetworkDown + | Io::ConnectionReset + | Io::ConnectionAborted + | Io::NotConnected + | Io::BrokenPipe + | Io::StaleNetworkFileHandle + | Io::Interrupted + | Io::ResourceBusy => DomainError::transient_backend(entity, message), + _ => DomainError::internal_error(entity, message), + } +} + #[cfg(test)] mod tests { use super::*; @@ -1085,4 +1129,56 @@ mod tests { "resume must start STRICTLY after the given hash" ); } + + /// "Local" does not mean reliable — the path can be an iSCSI LUN or + /// an NFS mount. The two directions this must never confuse: + /// + /// * a genuinely absent file must stay `NotFound`, because callers + /// act on that by concluding the bytes do not exist; + /// * an unreachable mount must NOT become `NotFound`, which is what + /// every one of these sites used to return unconditionally. + #[test] + fn local_io_errors_are_classified_not_all_notfound() { + use std::io::{Error, ErrorKind as Io}; + + let missing = local_io_error("Blob", "open".into(), &Error::from(Io::NotFound)); + assert_eq!(missing.kind, ErrorKind::NotFound); + assert!(!missing.is_transient()); + + // Network-backed mounts and interrupted syscalls: retry helps. + for kind in [ + Io::TimedOut, + Io::HostUnreachable, + Io::NetworkDown, + Io::ConnectionReset, + Io::StaleNetworkFileHandle, + Io::Interrupted, + Io::ResourceBusy, + ] { + let e = local_io_error("Blob", "open".into(), &Error::from(kind)); + assert!(e.is_transient(), "{kind:?} should be retryable"); + assert_ne!( + e.kind, + ErrorKind::NotFound, + "{kind:?} must never read as a missing blob" + ); + } + + // Operator-action or corruption: retrying changes nothing, and a + // finding naming the blob is the useful outcome. + for kind in [ + Io::PermissionDenied, + Io::ReadOnlyFilesystem, + Io::StorageFull, + Io::InvalidData, + ] { + let e = local_io_error("Blob", "open".into(), &Error::from(kind)); + assert!(!e.is_transient(), "{kind:?} should not be retryable"); + assert_ne!( + e.kind, + ErrorKind::NotFound, + "{kind:?} is not a missing blob" + ); + } + } } diff --git a/src/infrastructure/services/s3_blob_backend.rs b/src/infrastructure/services/s3_blob_backend.rs index 8b13178c..d61d90e8 100644 --- a/src/infrastructure/services/s3_blob_backend.rs +++ b/src/infrastructure/services/s3_blob_backend.rs @@ -286,11 +286,30 @@ impl BlobStorageBackend for S3BlobBackend { .send() .await .map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "S3", - format!("Failed to get blob {}: {}", hash, e), - ) + // Only a real NoSuchKey is NotFound. This used to + // label EVERY read failure that way — a refused + // connection, a 503, an expired credential all + // reported as "blob missing". + // + // That is the most dangerous wrong answer available + // here, because callers ACT on NotFound by concluding + // the bytes are gone. A migration reading its source + // through this would treat an outage as "the source + // does not have this blob" and move on. + // + // Everything else goes through the normal classifier, + // so a 403 stays permanent rather than being retried + // forever. + if let aws_sdk_s3::error::SdkError::ServiceError(svc) = &e + && svc.err().is_no_such_key() + { + return DomainError::new( + ErrorKind::NotFound, + "S3", + format!("Failed to get blob {hash}: no such key"), + ); + } + s3_domain_error("S3", format!("Failed to get blob {hash}"), &e) })?; // Convert S3 ByteStream into a Stream> @@ -325,11 +344,21 @@ impl BlobStorageBackend for S3BlobBackend { .send() .await .map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "S3", - format!("Failed to get blob range {}: {}", hash, e), - ) + // Same rule as the full read: only a real NoSuchKey + // is NotFound. Ranged reads feed CDC reassembly and + // deep verification, so mislabelling an outage here + // reads as "this chunk is gone" — a data-loss + // conclusion drawn from a network problem. + if let aws_sdk_s3::error::SdkError::ServiceError(svc) = &e + && svc.err().is_no_such_key() + { + return DomainError::new( + ErrorKind::NotFound, + "S3", + format!("Failed to get blob range {hash}: no such key"), + ); + } + s3_domain_error("S3", format!("Failed to get blob range {hash}"), &e) })?; let reader = output.body.into_async_read(); @@ -407,11 +436,19 @@ impl BlobStorageBackend for S3BlobBackend { .send() .await .map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "S3", - format!("Failed to stat blob {}: {}", hash, e), - ) + // `head_object` reports a missing key as NotFound + // rather than NoSuchKey, so match on the typed + // variant the SDK actually returns here. + if let aws_sdk_s3::error::SdkError::ServiceError(svc) = &e + && svc.err().is_not_found() + { + return DomainError::new( + ErrorKind::NotFound, + "S3", + format!("Failed to stat blob {hash}: not found"), + ); + } + s3_domain_error("S3", format!("Failed to stat blob {hash}"), &e) })?; Ok(output.content_length().unwrap_or(0) as u64)