Edouard Vanbelle
eba22f4c2c
The previous commit fixed get / get-range / stat but left `blob_exists`
returning `internal_error` on S3 and Azure, which undoes the point of
the exercise: `blob_exists` is the FIRST call `backend_migration` makes
against the source for every blob.
match self.source.blob_exists(hash).await { // migration, per blob
An unclassified error there is permanent, so a refused connection during
a migration takes the permanent branch — record a finding and move on —
which is the skip-and-advance behaviour the pause was added to prevent.
The classification has to hold at the probe, not only at the read that
follows it.
Both now classify before deciding: only a genuine 404 / `is_not_found`
answers "absent", everything else keeps its transient class. On S3 that
means classifying the `SdkError` by reference first, since
`into_service_error()` consumes it.
Local was already routed through `local_io_error` at its stat site.
Audited the rest of the S3 surface: initialize, put ×3, get, get-range,
delete, stat, list and exists all classify. The two remaining
`internal_error`s in `put_blob` read a *local* source file, so there is
no network class to preserve.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
801 lines
32 KiB
Rust
801 lines
32 KiB
Rust
//! Azure Blob Storage Backend — stores blobs in an Azure Storage container.
|
|
//!
|
|
//! Authenticates via Account Name + Account Key (or SAS token).
|
|
//! Blob key scheme mirrors local/S3: `{2-char-prefix}/{hash}.blob`.
|
|
|
|
use std::path::{Path, PathBuf};
|
|
use std::pin::Pin;
|
|
|
|
use azure_storage::StorageCredentials;
|
|
use azure_storage_blobs::prelude::*;
|
|
use bytes::Bytes;
|
|
use futures::{StreamExt, TryStreamExt};
|
|
use tokio::fs;
|
|
|
|
use crate::application::ports::blob_storage_ports::{
|
|
BackendBlobEntry, BackendUnknownEntry, BlobListPage, BlobStorageBackend, BlobStream,
|
|
StorageHealthStatus,
|
|
};
|
|
use crate::common::config::AzureStorageConfig;
|
|
use crate::domain::errors::{DomainError, ErrorKind};
|
|
|
|
/// Azure Blob Storage backend.
|
|
pub struct AzureBlobBackend {
|
|
container_client: ContainerClient,
|
|
container_name: String,
|
|
}
|
|
|
|
impl AzureBlobBackend {
|
|
/// Build a new Azure backend from configuration.
|
|
pub fn new(config: &AzureStorageConfig) -> Self {
|
|
let credentials = if let Some(ref sas) = config.sas_token {
|
|
StorageCredentials::sas_token(sas).expect("Invalid SAS token")
|
|
} else {
|
|
StorageCredentials::access_key(&config.account_name, config.account_key.clone())
|
|
};
|
|
|
|
// Custom endpoint (Azurite emulator / private deployment /
|
|
// benches) mirrors S3's `endpoint_url`; default is the public
|
|
// cloud URL derived from the account name.
|
|
let container_client = match &config.endpoint_url {
|
|
Some(uri) => ClientBuilder::with_location(
|
|
azure_storage::CloudLocation::Custom {
|
|
account: config.account_name.clone(),
|
|
uri: uri.trim_end_matches('/').to_string(),
|
|
},
|
|
credentials,
|
|
)
|
|
.container_client(&config.container),
|
|
None => ClientBuilder::new(&config.account_name, credentials)
|
|
.container_client(&config.container),
|
|
};
|
|
|
|
Self {
|
|
container_client,
|
|
container_name: config.container.clone(),
|
|
}
|
|
}
|
|
|
|
/// Inverse of [`Self::blob_name`] — the hash a listing entry names,
|
|
/// or `None` when the entry is not one of ours.
|
|
///
|
|
/// Mirrors `S3BlobBackend::hash_from_object_key`, including the check
|
|
/// that the shard equals the hash's own first two characters: without
|
|
/// it, `blob_name(hash)` would not reproduce the name we just parsed,
|
|
/// and a mis-sharded object would be reported as a live blob that no
|
|
/// read path can find.
|
|
fn hash_from_blob_name(name: &str) -> Option<String> {
|
|
let (prefix, rest) = name.split_once('/')?;
|
|
if prefix.len() != 2 || !prefix.chars().all(|c| c.is_ascii_hexdigit()) {
|
|
return None;
|
|
}
|
|
let stem = rest.strip_suffix(".blob")?;
|
|
if stem.len() != 64 || !stem.chars().all(|c| c.is_ascii_hexdigit()) {
|
|
return None;
|
|
}
|
|
if !stem.starts_with(prefix) {
|
|
return None;
|
|
}
|
|
Some(stem.to_string())
|
|
}
|
|
|
|
/// Compute the blob name for a given hash.
|
|
fn blob_name(hash: &str) -> String {
|
|
let prefix = &hash[0..2];
|
|
format!("{prefix}/{hash}.blob")
|
|
}
|
|
|
|
/// Get a `BlobClient` for a given hash.
|
|
fn blob_client(&self, hash: &str) -> BlobClient {
|
|
self.container_client.blob_client(Self::blob_name(hash))
|
|
}
|
|
}
|
|
|
|
impl BlobStorageBackend for AzureBlobBackend {
|
|
fn initialize(
|
|
&self,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
|
|
Box::pin(async move {
|
|
// Verify container exists by getting its properties
|
|
self.container_client.get_properties().await.map_err(|e| {
|
|
DomainError::internal_error(
|
|
"Azure",
|
|
format!("Cannot access container '{}': {}", self.container_name, e),
|
|
)
|
|
})?;
|
|
|
|
tracing::info!(
|
|
"Azure blob backend initialized: container={}",
|
|
self.container_name
|
|
);
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
fn put_blob(
|
|
&self,
|
|
hash: &str,
|
|
source_path: &Path,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
|
|
let hash = hash.to_owned();
|
|
let source_path = source_path.to_owned();
|
|
Box::pin(async move {
|
|
let client = self.blob_client(&hash);
|
|
|
|
// Check if blob already exists (idempotent)
|
|
if client.get_properties().await.is_ok() {
|
|
let file_size = fs::metadata(&source_path)
|
|
.await
|
|
.map_err(|e| {
|
|
DomainError::internal_error(
|
|
"Azure",
|
|
format!("Failed to stat source file: {e}"),
|
|
)
|
|
})?
|
|
.len();
|
|
let _ = fs::remove_file(&source_path).await;
|
|
return Ok(file_size);
|
|
}
|
|
|
|
// Read file and upload as block blob
|
|
let data = fs::read(&source_path).await.map_err(|e| {
|
|
DomainError::internal_error("Azure", format!("Failed to read source: {e}"))
|
|
})?;
|
|
let file_size = data.len() as u64;
|
|
|
|
client
|
|
.put_block_blob(data)
|
|
.await
|
|
.map_err(|e| azure_domain_error(format!("Failed to upload blob {hash}"), &e))?;
|
|
|
|
let _ = fs::remove_file(&source_path).await;
|
|
Ok(file_size)
|
|
})
|
|
}
|
|
|
|
fn put_blob_from_bytes(
|
|
&self,
|
|
hash: &str,
|
|
data: Bytes,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
|
|
let hash = hash.to_owned();
|
|
Box::pin(async move {
|
|
let client = self.blob_client(&hash);
|
|
let size = data.len() as u64;
|
|
|
|
// Idempotent: skip if exists
|
|
if client.get_properties().await.is_ok() {
|
|
return Ok(size);
|
|
}
|
|
|
|
// `Bytes` converts into `azure_core::Body` by reference count —
|
|
// the old `data.to_vec()` copied every chunk once more.
|
|
client
|
|
.put_block_blob(data)
|
|
.await
|
|
.map_err(|e| azure_domain_error(format!("Failed to upload blob {hash}"), &e))?;
|
|
|
|
Ok(size)
|
|
})
|
|
}
|
|
|
|
/// Dedup settle path: PUT unconditionally. Content-addressed keys make
|
|
/// re-PUTs idempotent, so the `get_properties` probe
|
|
/// `put_blob_from_bytes` pays is a pure extra round-trip on every NEW
|
|
/// chunk (2 RTTs -> 1, benches/S3-PUT.md — same shape as S3).
|
|
fn put_blob_from_bytes_unsynced(
|
|
&self,
|
|
hash: &str,
|
|
data: Bytes,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
|
|
let hash = hash.to_owned();
|
|
Box::pin(async move {
|
|
let client = self.blob_client(&hash);
|
|
let size = data.len() as u64;
|
|
client
|
|
.put_block_blob(data)
|
|
.await
|
|
.map_err(|e| azure_domain_error(format!("Failed to upload blob {hash}"), &e))?;
|
|
Ok(size)
|
|
})
|
|
}
|
|
|
|
/// Atomic overwrite path used by `backend_rotate` and
|
|
/// `backend_migration` when re-writing an already-present blob
|
|
/// under a new head key/format. Trait default delegates to
|
|
/// `put_blob_from_bytes` which `get_properties`-probes and
|
|
/// silently skips — exactly wrong for the rotate/migrate use
|
|
/// case (the whole point is to replace the existing bytes).
|
|
/// Override delegates to the same unconditional PUT as
|
|
/// `put_blob_from_bytes_unsynced` — Azure's PUT is durable on
|
|
/// return, no separate sync barrier needed.
|
|
fn put_blob_from_bytes_replace(
|
|
&self,
|
|
hash: &str,
|
|
data: Bytes,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
|
|
self.put_blob_from_bytes_unsynced(hash, data)
|
|
}
|
|
|
|
fn get_blob_stream(
|
|
&self,
|
|
hash: &str,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<BlobStream, DomainError>> + Send + '_>>
|
|
{
|
|
let hash = hash.to_owned();
|
|
Box::pin(async move {
|
|
let client = self.blob_client(&hash);
|
|
|
|
// The old implementation drained the ENTIRE blob into one
|
|
// `Vec<u8>` before yielding a single mega-chunk — whole-blob
|
|
// RAM residency per reader, and with `read_prefetch() = 8`
|
|
// up to 8 entire chunk-blobs resident at once during CDC
|
|
// reassembly. Now the SDK's page/body streams forward
|
|
// directly. The FIRST page is still awaited eagerly so a
|
|
// missing blob surfaces as the same up-front NotFound the
|
|
// old code produced; later pages/chunks map to io::Error
|
|
// items like every other backend's stream.
|
|
let mut pages = client.get().into_stream();
|
|
let first = match pages.next().await {
|
|
Some(Ok(response)) => response,
|
|
Some(Err(e)) => {
|
|
return Err(azure_read_error(format!("Failed to get blob {hash}"), &e));
|
|
}
|
|
None => {
|
|
let empty: BlobStream =
|
|
Box::pin(futures::stream::once(async move { Ok(Bytes::new()) }));
|
|
return Ok(empty);
|
|
}
|
|
};
|
|
|
|
let first_body = first.data.map(|chunk| {
|
|
chunk.map_err(|e| std::io::Error::other(format!("Stream read error: {e}")))
|
|
});
|
|
let tail = pages
|
|
.map(|page| match page {
|
|
Ok(response) => Ok(response.data.map(|chunk| {
|
|
chunk.map_err(|e| std::io::Error::other(format!("Stream read error: {e}")))
|
|
})),
|
|
Err(e) => Err(std::io::Error::other(format!(
|
|
"Failed to get blob page: {e}"
|
|
))),
|
|
})
|
|
.try_flatten();
|
|
let stream: BlobStream = Box::pin(first_body.chain(tail));
|
|
Ok(stream)
|
|
})
|
|
}
|
|
|
|
/// # Known incompatibility: sub-4 MiB ranges break on Azurite
|
|
///
|
|
/// `azure_core` 0.21's `Range::as_headers`
|
|
/// (`src/request_options/range.rs`) attaches
|
|
/// `x-ms-range-get-content-crc64: true` to **any range shorter than
|
|
/// 4 MiB**, unconditionally and with no opt-out. Real Azure honours
|
|
/// it; Azurite answers 500. `azure_core` then classifies 500 as
|
|
/// retryable and loops on a deterministic error, forever.
|
|
///
|
|
/// The reachable path is `backend_migration` →
|
|
/// `EncryptedBlobBackend::head_check` →
|
|
/// `get_blob_range_stream(hash, 0, HEADER_SIZE)`. `HEADER_SIZE` is a
|
|
/// few dozen bytes, and it runs against the TARGET before each write,
|
|
/// so a local→Azurite migration hangs on its first blob while holding
|
|
/// `migration_readonly` — writes refused application-wide.
|
|
///
|
|
/// **Deliberately not worked around here.** The available workaround
|
|
/// is to issue an unranged `get()` for small requests (its 16 MiB
|
|
/// `initial_range` clears the threshold, so the header is never sent)
|
|
/// and truncate client-side. That is correct against real Azure but
|
|
/// pays for an emulator with production cost: a ~40-byte format probe
|
|
/// becomes a whole-blob transfer, and it puts new offset arithmetic
|
|
/// on the read path, where a mistake serves wrong bytes silently
|
|
/// rather than failing.
|
|
///
|
|
/// The real fix is the official `azure_storage_blob` 1.x, where
|
|
/// `range_get_content_crc64` is an explicit field on
|
|
/// `BlobClientDownloadOptions` — leave it unset and the request is
|
|
/// never made. Until then, the Azurite suite exercises enumeration
|
|
/// and round-trips but not migration; see
|
|
/// `tests/api/backend_consistency_azure.hurl`.
|
|
fn get_blob_range_stream(
|
|
&self,
|
|
hash: &str,
|
|
start: u64,
|
|
end: Option<u64>,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<BlobStream, DomainError>> + Send + '_>>
|
|
{
|
|
let hash = hash.to_owned();
|
|
Box::pin(async move {
|
|
let client = self.blob_client(&hash);
|
|
|
|
let range = match end {
|
|
Some(e) => azure_core::request_options::Range::new(start, e),
|
|
None => azure_core::request_options::Range::new(start, u64::MAX),
|
|
};
|
|
|
|
// Same forwarding shape as `get_blob_stream` — a ranged read
|
|
// doubly so: the caller explicitly asked NOT to pay for the
|
|
// whole blob, yet the old code buffered the full range.
|
|
let mut pages = client.get().range(range).into_stream();
|
|
let first = match pages.next().await {
|
|
Some(Ok(response)) => response,
|
|
Some(Err(e)) => {
|
|
return Err(azure_read_error(
|
|
format!("Failed to get blob range {hash}"),
|
|
&e,
|
|
));
|
|
}
|
|
None => {
|
|
let empty: BlobStream =
|
|
Box::pin(futures::stream::once(async move { Ok(Bytes::new()) }));
|
|
return Ok(empty);
|
|
}
|
|
};
|
|
|
|
let first_body = first.data.map(|chunk| {
|
|
chunk.map_err(|e| std::io::Error::other(format!("Stream range read error: {e}")))
|
|
});
|
|
let tail = pages
|
|
.map(|page| match page {
|
|
Ok(response) => Ok(response.data.map(|chunk| {
|
|
chunk.map_err(|e| {
|
|
std::io::Error::other(format!("Stream range read error: {e}"))
|
|
})
|
|
})),
|
|
Err(e) => Err(std::io::Error::other(format!(
|
|
"Failed to get blob range page: {e}"
|
|
))),
|
|
})
|
|
.try_flatten();
|
|
let stream: BlobStream = Box::pin(first_body.chain(tail));
|
|
Ok(stream)
|
|
})
|
|
}
|
|
|
|
fn delete_blob(
|
|
&self,
|
|
hash: &str,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
|
|
let hash = hash.to_owned();
|
|
Box::pin(async move {
|
|
let client = self.blob_client(&hash);
|
|
|
|
// Azure delete is not fully idempotent — 404 is expected for missing blobs
|
|
match client.delete().await {
|
|
Ok(_) => Ok(()),
|
|
Err(e) => {
|
|
// If 404, treat as success (idempotent)
|
|
let status = e.as_http_error().map(|h| h.status());
|
|
if status == Some(azure_core::StatusCode::NotFound) {
|
|
Ok(())
|
|
} else {
|
|
Err(azure_domain_error(
|
|
format!("Failed to delete blob {hash}"),
|
|
&e,
|
|
))
|
|
}
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
fn blob_exists(
|
|
&self,
|
|
hash: &str,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<bool, DomainError>> + Send + '_>> {
|
|
let hash = hash.to_owned();
|
|
Box::pin(async move {
|
|
let client = self.blob_client(&hash);
|
|
match client.get_properties().await {
|
|
Ok(_) => Ok(true),
|
|
Err(e) => {
|
|
let status = e.as_http_error().map(|h| h.status());
|
|
if status == Some(azure_core::StatusCode::NotFound) {
|
|
Ok(false)
|
|
} else {
|
|
// Only the 404 means "absent"; everything else keeps
|
|
// its transient/permanent class so the migration's
|
|
// source probe can pause on an outage instead of
|
|
// recording a permanent finding.
|
|
Err(azure_domain_error(
|
|
format!("Failed to check blob {hash}"),
|
|
&e,
|
|
))
|
|
}
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
fn blob_size(
|
|
&self,
|
|
hash: &str,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
|
|
let hash = hash.to_owned();
|
|
Box::pin(async move {
|
|
let client = self.blob_client(&hash);
|
|
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)
|
|
})
|
|
}
|
|
|
|
fn health_check(
|
|
&self,
|
|
) -> Pin<
|
|
Box<dyn std::future::Future<Output = Result<StorageHealthStatus, DomainError>> + Send + '_>,
|
|
> {
|
|
Box::pin(async move {
|
|
match self.container_client.get_properties().await {
|
|
Ok(_) => Ok(StorageHealthStatus {
|
|
connected: true,
|
|
backend_type: "azure".to_string(),
|
|
message: format!("Azure container '{}' is accessible", self.container_name),
|
|
available_bytes: None,
|
|
}),
|
|
Err(e) => Ok(StorageHealthStatus {
|
|
connected: false,
|
|
backend_type: "azure".to_string(),
|
|
message: format!(
|
|
"Azure container '{}' is not accessible: {}",
|
|
self.container_name, e
|
|
),
|
|
available_bytes: None,
|
|
}),
|
|
}
|
|
})
|
|
}
|
|
|
|
/// Enumerate blob hashes in lexicographic order, so
|
|
/// `backend_consistency` can merge-join against `storage.blobs`
|
|
/// instead of degrading to a per-row probe that structurally cannot
|
|
/// see orphans.
|
|
///
|
|
/// ## Why this is a shard walk and not one flat listing
|
|
///
|
|
/// **The cursor IS a blob hash**, not a provider token. The caller
|
|
/// forces that: it advances ONE cursor across both sides of the join,
|
|
/// feeding the same value here and to `WHERE hash > $1` in SQL. S3
|
|
/// satisfies it with `start_after(object_key(cursor))`.
|
|
///
|
|
/// Azure has no `StartAfter`. REST API 2023-05-03 added `startFrom`,
|
|
/// which would be the direct equivalent — but this SDK
|
|
/// (`azure_storage_blobs` 0.21, archived) never sends it: `ListBlobs`
|
|
/// exposes only `prefix`, `delimiter`, `max_results` and `marker`,
|
|
/// and `marker` is an opaque continuation token that cannot be
|
|
/// derived from a hash.
|
|
///
|
|
/// So resume rides on `prefix` instead. Names are
|
|
/// `{hash[0..2]}/{hash}.blob`, which partitions the container into
|
|
/// 256 shards that are themselves in hash order. Walking
|
|
/// `00/` … `ff/` therefore yields exactly the global hash order, and
|
|
/// a cursor names the shard to restart in. Re-listing on resume is
|
|
/// bounded by shard width — 1/256th of the container — rather than
|
|
/// by the whole container, which is what a client-side skip over a
|
|
/// flat listing would cost on every single page.
|
|
///
|
|
/// `marker` is used only INSIDE one call, to page within a shard, and
|
|
/// never escapes as the cursor — the same treatment the S3 impl gives
|
|
/// its continuation token.
|
|
///
|
|
/// ## What this does NOT see, unlike S3
|
|
///
|
|
/// S3 lists the bucket with no prefix, so any foreign object lands in
|
|
/// `unknowns`. Constraining to `{2-hex}/` means foreign names outside
|
|
/// that shape are invisible here.
|
|
///
|
|
/// That asymmetry is deliberate and safe in the direction that
|
|
/// matters: an orphan is a blob **we** wrote and later stopped
|
|
/// referencing, so it always has the canonical name and is always
|
|
/// enumerated. Only genuinely foreign files — another workload
|
|
/// sharing the container — can be missed, and they are informational
|
|
/// notices, never findings. Trading them for O(N) enumeration instead
|
|
/// of O(N²/limit) is worth it.
|
|
fn list_blob_hashes(
|
|
&self,
|
|
cursor: Option<String>,
|
|
limit: usize,
|
|
) -> Pin<Box<dyn std::future::Future<Output = Result<BlobListPage, DomainError>> + Send + '_>>
|
|
{
|
|
Box::pin(async move {
|
|
// A run of foreign entries can't produce a resume cursor, and
|
|
// buffering the container to find one blob is worse than
|
|
// failing. Mirrors the S3 impl's bound, and like it is on
|
|
// entries accumulated rather than requests made: request
|
|
// count scales with the caller's `limit`, so a request cap
|
|
// would fire on a healthy container merely because the caller
|
|
// paged finely.
|
|
const MAX_UNKNOWNS: usize = 10_000;
|
|
|
|
// A shard is `{2-hex}/`, so the space is 0x00..=0xff.
|
|
const LAST_SHARD: u16 = 0xff;
|
|
|
|
let mut blobs: Vec<BackendBlobEntry> = Vec::new();
|
|
let mut unknowns: Vec<BackendUnknownEntry> = Vec::new();
|
|
|
|
// Resume in the cursor's own shard — its remaining entries
|
|
// still sort after it, and the client-side skip below drops
|
|
// the ones that don't. A malformed cursor is a bug in the
|
|
// caller's checkpoint, and silently restarting from `00`
|
|
// would re-report every blob as new, so refuse it.
|
|
let mut shard: u16 = match cursor.as_deref() {
|
|
Some(c) => u16::from(u8::from_str_radix(c.get(0..2).unwrap_or(""), 16).map_err(
|
|
|_| {
|
|
DomainError::internal_error(
|
|
"Blob",
|
|
format!(
|
|
"Azure enumeration cursor '{c}' is not a blob hash — it must \
|
|
start with the two hex characters naming its shard"
|
|
),
|
|
)
|
|
},
|
|
)?),
|
|
None => 0,
|
|
};
|
|
|
|
// Azure caps a page at 5000; asking for the caller's `limit`
|
|
// keeps a small page cheap. `MaxResults` rejects zero, and a
|
|
// caller asking for nothing still needs a well-formed
|
|
// request — and, more importantly, must not be answered with
|
|
// an empty page and a `None` cursor, which would read as
|
|
// "container fully enumerated, nothing here".
|
|
let want = limit.max(1);
|
|
let page_size = want.min(5000) as u32;
|
|
|
|
'shards: while shard <= LAST_SHARD {
|
|
let prefix = format!("{shard:02x}/");
|
|
// `Pageable` follows `next_marker` itself, so one stream
|
|
// covers the whole shard however many round-trips it takes.
|
|
let mut pages = self
|
|
.container_client
|
|
.list_blobs()
|
|
.prefix(prefix)
|
|
.max_results(std::num::NonZeroU32::new(page_size).expect("clamped above 0"))
|
|
.into_stream();
|
|
|
|
while let Some(page) = pages.next().await {
|
|
let page = page.map_err(|e| {
|
|
// Classified: `backend_consistency` fails the whole
|
|
// run on an enumeration error, so a throttle
|
|
// partway through the 256-shard walk should be
|
|
// retryable rather than discarding the sweep.
|
|
azure_domain_error(
|
|
format!(
|
|
"Azure ListBlobs failed on shard {shard:02x} of container '{}'",
|
|
self.container_name
|
|
),
|
|
&e,
|
|
)
|
|
})?;
|
|
|
|
for blob in page.blobs.blobs() {
|
|
let name = blob.name.clone();
|
|
// `OffsetDateTime` → chrono, for the caller's
|
|
// grace window. A value outside chrono's range
|
|
// degrades to `None`, which the port documents as
|
|
// "treat as old enough" — the conservative side,
|
|
// since it only ever suppresses a finding on a
|
|
// freshly-written blob.
|
|
let mtime = chrono::DateTime::<chrono::Utc>::from_timestamp(
|
|
blob.properties.last_modified.unix_timestamp(),
|
|
blob.properties.last_modified.nanosecond(),
|
|
);
|
|
|
|
match Self::hash_from_blob_name(&name) {
|
|
Some(hash) => {
|
|
// `prefix` is inclusive of the cursor's own
|
|
// entry and of everything before it in the
|
|
// shard. Without this skip the caller sees
|
|
// a hash it already consumed and the
|
|
// merge-join never advances past it.
|
|
if cursor.as_deref().is_some_and(|c| hash.as_str() <= c) {
|
|
continue;
|
|
}
|
|
blobs.push(BackendBlobEntry { hash, mtime });
|
|
}
|
|
// Not ours — a foreign workload sharing the
|
|
// container. Surfaced rather than dropped so
|
|
// an operator can see it.
|
|
None => unknowns.push(BackendUnknownEntry { path: name, mtime }),
|
|
}
|
|
}
|
|
|
|
if blobs.len() >= want {
|
|
break 'shards;
|
|
}
|
|
|
|
if unknowns.len() >= MAX_UNKNOWNS {
|
|
return Err(DomainError::internal_error(
|
|
"Blob",
|
|
format!(
|
|
"Azure enumeration accumulated {} non-blob entrie(s) without \
|
|
filling a page, so no resume cursor can be produced. Container \
|
|
'{}' likely holds a large foreign namespace — give OxiCloud a \
|
|
dedicated container.",
|
|
unknowns.len(),
|
|
self.container_name,
|
|
),
|
|
));
|
|
}
|
|
}
|
|
|
|
shard += 1;
|
|
}
|
|
|
|
// Exhausting every shard is the ONLY end of enumeration.
|
|
// Stopping early because one shard was empty would truncate
|
|
// the sweep and report the rest of the container as absent,
|
|
// so `shard > LAST_SHARD` — not "this page was empty" — is
|
|
// what produces `None`.
|
|
let next_cursor = if shard > LAST_SHARD {
|
|
None
|
|
} else {
|
|
blobs.last().map(|entry| entry.hash.clone())
|
|
};
|
|
|
|
Ok(BlobListPage {
|
|
blobs,
|
|
unknowns,
|
|
next_cursor,
|
|
})
|
|
})
|
|
}
|
|
|
|
fn backend_type(&self) -> &'static str {
|
|
"azure"
|
|
}
|
|
|
|
/// Remote object store: overlap chunk GETs to hide per-request latency.
|
|
fn read_prefetch(&self) -> usize {
|
|
8
|
|
}
|
|
|
|
fn local_blob_path(&self, _hash: &str) -> Option<PathBuf> {
|
|
None
|
|
}
|
|
}
|
|
|
|
/// Wrap an `azure_core` error as a `DomainError` that says whether
|
|
/// retrying it could help. Azure counterpart of `s3_domain_error`.
|
|
///
|
|
/// `azure_core::error::ErrorKind::HttpResponse` carries the status, so
|
|
/// this works on the archived 0.21 SDK — no need to wait for the
|
|
/// official-crate migration. That matters because Azure is the backend
|
|
/// the retry-then-pause plan was written for: a ranged GET carrying
|
|
/// `x-ms-range-get-content-crc64` that Azurite answers 500 to, retried
|
|
/// forever by `azure_core` while `migration_readonly` refused writes
|
|
/// application-wide.
|
|
///
|
|
/// Transient: 5xx, 429, 408. Also `Io` — connection resets, DNS, TLS.
|
|
/// Permanent: other 4xx (credentials, missing container, malformed
|
|
/// request), `DataConversion`, `Credential`.
|
|
///
|
|
/// **A deterministic 500 still classifies as transient**, and that is
|
|
/// deliberate rather than an oversight. Nothing at this layer can tell
|
|
/// "this provider is briefly unwell" from "this provider will answer
|
|
/// 500 to this exact request forever" — the Azurite CRC64 case is the
|
|
/// 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;
|
|
|
|
let transient = match err.kind() {
|
|
AzKind::HttpResponse { status, .. } => {
|
|
let code = u16::from(*status);
|
|
code >= 500 || code == 429 || code == 408
|
|
}
|
|
AzKind::Io => true,
|
|
AzKind::DataConversion | AzKind::Credential | AzKind::MockFramework | AzKind::Other => {
|
|
false
|
|
}
|
|
};
|
|
|
|
let message = format!("{context}: {err}");
|
|
if transient {
|
|
DomainError::transient_backend("Azure", message)
|
|
} else {
|
|
DomainError::internal_error("Azure", message)
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
const H: &str = "0a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f9";
|
|
|
|
/// The enumeration cursor is fed straight back in as a shard prefix,
|
|
/// so a name that does not round-trip would resume in the wrong shard
|
|
/// and silently skip everything between.
|
|
#[test]
|
|
fn blob_name_round_trips_through_hash_from_blob_name() {
|
|
let name = AzureBlobBackend::blob_name(H);
|
|
assert_eq!(name, format!("0a/{H}.blob"));
|
|
assert_eq!(
|
|
AzureBlobBackend::hash_from_blob_name(&name).as_deref(),
|
|
Some(H)
|
|
);
|
|
}
|
|
|
|
/// Each of these would otherwise be treated as a hash — and the
|
|
/// resume path slices `[0..2]` off it to pick the next shard.
|
|
#[test]
|
|
fn non_canonical_names_are_rejected() {
|
|
let cases = [
|
|
"0a/junk.tmp".to_string(), // spool file
|
|
"junk.tmp".to_string(), // no shard
|
|
"0a/junk".to_string(), // no suffix
|
|
"thumbnails/abc.jpg".to_string(), // foreign namespace
|
|
format!("0a/{H}.blob.corrupt"), // sidecar
|
|
format!("0a/{H}"), // suffix missing
|
|
format!("zz/{H}.blob"), // non-hex shard
|
|
format!("ff/{H}.blob"), // shard != hash prefix
|
|
format!("0a/{}.blob", &H[..63]), // wrong length
|
|
];
|
|
for name in &cases {
|
|
assert_eq!(
|
|
AzureBlobBackend::hash_from_blob_name(name),
|
|
None,
|
|
"must not be read as a blob: {name}"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// The shard walk relies on `{hash[0..2]}/…` ordering lexicographic
|
|
/// names into exactly the order `ORDER BY hash` produces. If the
|
|
/// shard were not the hash's own prefix the two sequences would
|
|
/// interleave differently and the merge-join would emit phantom
|
|
/// findings in BOTH directions.
|
|
#[test]
|
|
fn shard_order_matches_hash_order() {
|
|
let hashes = ["00aa", "0a1b", "0aff", "b0cd", "ffff"]
|
|
.map(|p| format!("{p}{}", "0".repeat(60)))
|
|
.to_vec();
|
|
|
|
let mut names: Vec<String> = hashes
|
|
.iter()
|
|
.map(|h| AzureBlobBackend::blob_name(h))
|
|
.collect();
|
|
names.sort();
|
|
|
|
let recovered: Vec<String> = names
|
|
.iter()
|
|
.filter_map(|n| AzureBlobBackend::hash_from_blob_name(n))
|
|
.collect();
|
|
|
|
let mut sorted_hashes = hashes.clone();
|
|
sorted_hashes.sort();
|
|
assert_eq!(recovered, sorted_hashes);
|
|
}
|
|
}
|