src/heal.rs
// Apache-2.0 licensed.
//! Crash-recovery for a local (`file://`) warehouse.
//!
//! The catalog pointer in `catalog.redb` is advanced under a durable (fsync'd)
//! redb transaction, but the Iceberg metadata JSON it points at is written via
//! `FileIO` *without* an fsync. On a local filesystem an unclean shutdown
//! (power loss, host reboot) can therefore land the durable pointer on metadata
//! bytes that were still only in the page cache — the file survives as a
//! zero-byte stub and the table fails to open (`EOF while parsing a value`).
//!
//! Two defences live here:
//! - **Prevention** — [`fsync_local_commit`] flushes a just-written commit's
//! WHOLE file set — the manifests the new snapshot added, the manifest list,
//! and the metadata blob (plus their directories) — before the pointer
//! advances (called from the commit paths). Best-effort; object-store
//! backends are a no-op.
//! - **Recovery** — [`RedbCatalog::heal_table`] detects a pointer that resolves
//! to a missing/empty/unparseable metadata file and rolls it back to the
//! newest metadata in the table's directory that fully parses and whose
//! snapshot's manifest list is present and non-empty.
//!
//! Both are pointless unwired: [`RedbCatalog::heal_all_tables`] is the sweep an
//! embedder is expected to run once when it opens a warehouse.
use std::collections::HashSet;
use iceberg::io::FileIO;
use iceberg::spec::{ManifestList, Snapshot, TableMetadata};
use iceberg::{Catalog, NamespaceIdent, Result, TableIdent};
use crate::catalog::RedbCatalog;
use crate::error::map_redb;
use crate::keys::table_key;
use crate::store::{CommitOutcome, TABLES};
/// What [`RedbCatalog::heal_table`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum HealOutcome {
/// The current pointer resolves to loadable metadata — nothing changed.
Healthy,
/// The pointer was corrupt; it was rolled back from `from` to `to`.
Healed { from: String, to: String },
/// The table is not in the catalog — nothing to heal.
Unknown,
/// A non-`file://` backend — local crash-recovery does not apply.
Skipped,
/// The pointer is corrupt and NO intact metadata was found in the table
/// directory — manual intervention required.
Unrecoverable { location: String },
}
/// Flush one `file://` blob to disk. Best-effort: non-`file://` locations are a
/// no-op (object stores have their own durability) and I/O errors are swallowed
/// — [`RedbCatalog::heal_table`] is the backstop.
fn fsync_local_file(location: &str) {
let Some(path) = location.strip_prefix("file://") else {
return;
};
if let Ok(f) = std::fs::File::open(path) {
let _ = f.sync_all();
}
}
/// Flush a `file://` blob's parent DIRECTORY, so the entry naming it survives
/// too. Split from [`fsync_local_file`] because one commit's files share a
/// directory: the caller syncs each file, then each distinct directory once.
fn fsync_local_parent_dir(location: &str) {
let Some(path) = location.strip_prefix("file://") else {
return;
};
if let Some(dir) = std::path::Path::new(path).parent() {
if let Ok(d) = std::fs::File::open(dir) {
let _ = d.sync_all();
}
}
}
/// Every local blob a just-applied commit must have on **stable storage before
/// the durable catalog pointer names it**, in flush order (children first):
///
/// 1. every manifest the new snapshot ADDED (manifests carried forward from an
/// ancestor were flushed by the commit that added them — re-syncing them
/// would make a commit's cost grow with the table's whole manifest history),
/// 2. the snapshot's manifest list,
/// 3. the metadata JSON itself.
///
/// Why the avro files and not just the JSON — this is the whole point of the
/// function. iceberg-rust writes manifests, manifest lists AND metadata through
/// `FileIO`, and `FileIO` fsyncs none of them. Flushing only the JSON leaves the
/// durable pointer aimed at a metadata file that is present and *parses*, while
/// the avro bytes it transitively names are still page cache. That is exactly
/// the shape the 2026-08-16 20:11 oden crash left on the live nornir warehouse:
/// nine intact `metadata.json` versions, seven **0-byte `*-m0.avro` manifests**,
/// and every scan of `embeddings` / `embedding_manifest` /
/// `embedding_snapshots` dying on `Failed to load manifest … DataInvalid`. A
/// metadata-only fsync cannot prevent that shape; this can.
///
/// A metadata with no current snapshot (a fresh `create_table`) yields just the
/// metadata JSON — the same single flush the metadata-only version did.
/// A manifest list that fails to load yields no manifest paths: the commit is
/// already unsound, and heal is the backstop.
/// A snapshot's manifest list, read straight through `fileio` and parsed for the
/// table's format version.
///
/// iceberg-rust main (2026-10-02 rebase) removed `Snapshot::load_manifest_list`;
/// its replacement, `Table::manifest_list_reader`, needs a whole `Table`, and heal
/// holds only a `FileIO` and the metadata it is judging. This is that reader's
/// unencrypted path. skade never encrypts, so an encrypted snapshot here is
/// refused by name rather than misread.
async fn load_manifest_list(
fileio: &FileIO,
snap: &Snapshot,
metadata: &TableMetadata,
) -> Result<ManifestList> {
if snap.encryption_key_id().is_some() {
return Err(iceberg::Error::new(
iceberg::ErrorKind::FeatureUnsupported,
format!("heal cannot read encrypted manifest list {}", snap.manifest_list()),
));
}
let bytes = fileio.new_input(snap.manifest_list())?.read().await?;
ManifestList::parse_with_version(&bytes, metadata.format_version())
}
async fn durability_set(
fileio: &FileIO,
metadata: &TableMetadata,
metadata_location: &str,
) -> Vec<String> {
let mut out = Vec::new();
if let Some(snap) = metadata.current_snapshot() {
if let Ok(list) = load_manifest_list(fileio, snap, metadata).await {
for mf in list.entries() {
if mf.added_snapshot_id == snap.snapshot_id() {
out.push(mf.manifest_path.clone());
}
}
}
out.push(snap.manifest_list().to_string());
}
out.push(metadata_location.to_string());
out
}
/// Flush a just-written commit's whole file set (see [`durability_set`]) and the
/// directories holding it. Call after writing the metadata and BEFORE advancing
/// the catalog pointer, so a crash can never leave the durable pointer aimed at
/// un-flushed bytes — at any level of the tree, not just the top one.
///
/// The work is a handful of `fsync`s on the commit path, which already fsyncs
/// the redb transaction next.
pub(crate) async fn fsync_local_commit(
fileio: &FileIO,
metadata: &TableMetadata,
metadata_location: &str,
) {
let set = durability_set(fileio, metadata, metadata_location).await;
for location in &set {
fsync_local_file(location);
}
// One directory sync per distinct parent — a commit's blobs normally share
// the table's `metadata/` dir, so this is one fsync, not one per file.
let mut synced: HashSet<&str> = HashSet::new();
for location in &set {
let parent = location.rsplit_once('/').map(|(d, _)| d).unwrap_or("");
if synced.insert(parent) {
fsync_local_parent_dir(location);
}
}
}
/// Does `location` resolve to fully loadable metadata: present, non-empty,
/// parseable, and (if it has a current snapshot) with a manifest list that
/// parses AND whose every referenced manifest file is present and non-empty?
///
/// `require_snapshot` distinguishes the two uses. For the CURRENT-pointer health
/// check it is `false`: an unappended table (no snapshot) is legitimately
/// healthy. For a ROLLBACK candidate it is `true`: a snapshot-less create-time
/// metadata must NEVER be accepted as a recovery target, or heal would silently
/// roll a table that HAD data back to empty (and `cleanup_orphans` would then GC
/// the retired data dir) — total data loss masked as success. When the only
/// intact candidate is the empty create-metadata, the caller returns
/// [`HealOutcome::Unrecoverable`] (loud) instead.
///
/// Checking the individual manifests — not just the manifest *list* — is what
/// lets heal recover from a torn compaction. A SIGTERM landing mid-compaction can
/// leave a zero-byte `*-m0.avro` manifest that an intact, parseable manifest LIST
/// still references; validating only the list would pass such a snapshot as
/// healthy, yet every later scan dies loading the 0-byte manifest. Statting each
/// manifest and rejecting the snapshot on the first missing/empty one makes heal
/// roll the pointer back past the torn commit automatically.
async fn metadata_is_intact(fileio: &FileIO, location: &str, require_snapshot: bool) -> bool {
// `read_from` reads + parses; a missing/empty/truncated file errors here.
let Ok(md) = TableMetadata::read_from(fileio, location).await else {
return false;
};
// No current snapshot: healthy only when we are not demanding one (see the
// `require_snapshot` contract above — a create-time metadata must never be a
// rollback target).
let Some(snap) = md.current_snapshot() else {
return !require_snapshot;
};
// The manifest list must be present, non-empty AND parse — the commit that
// wrote this metadata wrote the list in the same burst, so a torn commit
// shows up as a zero-byte / truncated `snap-*.avro` and fails to load here.
let Ok(manifest_list) = load_manifest_list(fileio, snap, &md).await else {
return false;
};
// ...and every manifest the list references must be present and non-empty. A
// 0-byte `*-m0.avro` (torn compaction) is invisible to the list check above
// but fatal to any later scan. A `metadata()` stat is cheap and rejects both
// a missing file (Err) and a 0-byte one (size == 0); short-circuit on the
// first bad manifest so a rejected candidate costs one stat, not a full scan.
for mf in manifest_list.entries() {
let manifest_ok = match fileio.new_input(&mf.manifest_path) {
Ok(input) => matches!(input.metadata().await, Ok(m) if m.size > 0),
Err(_) => false,
};
if !manifest_ok {
return false;
}
}
true
}
/// Order `NNNNN-<uuid>.metadata.json` names newest-first by PARSED version.
/// Iceberg pads the version to a *minimum* of 5 digits (`{:0>5}`), not a fixed
/// width, so a plain lexicographic sort breaks the moment the version crosses
/// 100000 (`"99999"` sorts after `"100000"` as strings). Parse the leading
/// integer; non-conforming names sort last.
fn newest_first(mut names: Vec<String>) -> Vec<String> {
fn version_of(name: &str) -> u64 {
name.split('-')
.next()
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(0)
}
names.sort_by(|a, b| version_of(b).cmp(&version_of(a)).then_with(|| b.cmp(a)));
names
}
impl RedbCatalog {
/// If `ident`'s current metadata pointer resolves to a missing, empty, or
/// unparseable metadata file (e.g. after an unclean shutdown left the file
/// un-fsync'd while the durable pointer advanced), roll the pointer back to
/// the newest metadata file in the table's directory that fully loads.
///
/// A no-op ([`HealOutcome::Healthy`]) when the pointer is already sound, the
/// table is unknown, or the backend is not `file://`. Data files are never
/// touched — this only rewinds the pointer to an already-durable snapshot,
/// so it is safe to call unconditionally at startup.
pub async fn heal_table(&self, ident: &TableIdent) -> Result<HealOutcome> {
let key = table_key(&self.name, ident);
// The on-disk pointer (not the in-memory mirror — this runs at startup).
let current = {
let db = self.store.db.lock().await;
let read = db.begin_read().map_err(map_redb)?;
let tables = read.open_table(TABLES).map_err(map_redb)?;
match tables.get(key.as_str()).map_err(map_redb)? {
Some(v) => v.value().to_string(),
None => return Ok(HealOutcome::Unknown),
}
};
let Some(current_path) = current.strip_prefix("file://") else {
return Ok(HealOutcome::Skipped);
};
if metadata_is_intact(&self.fileio, ¤t, false).await {
return Ok(HealOutcome::Healthy);
}
// The pointer is corrupt. Scan the metadata directory for the newest
// intact candidate that actually carries data (see `require_snapshot`).
let Some(dir) = std::path::Path::new(current_path).parent() else {
return Ok(HealOutcome::Unrecoverable { location: current });
};
let names: Vec<String> = match std::fs::read_dir(dir) {
Ok(rd) => rd
.filter_map(|e| e.ok())
.filter_map(|e| e.file_name().into_string().ok())
.filter(|n| n.ends_with(".metadata.json"))
.collect(),
Err(_) => return Ok(HealOutcome::Unrecoverable { location: current }),
};
for name in newest_first(names) {
let cand = format!("file://{}/{}", dir.display(), name);
if cand == current {
continue;
}
if metadata_is_intact(&self.fileio, &cand, true).await {
// Repoint through the shared write-through path so the durable
// TABLES entry and the in-memory L1 mirror move together. No
// commit event: this rewinds to an existing snapshot, it does
// not create one.
let (k, loc) = (key.clone(), cand.clone());
self.store
.group_commit(Box::new(move |write| {
let mut tables = write.open_table(TABLES).map_err(map_redb)?;
tables.insert(k.as_str(), loc.as_str()).map_err(map_redb)?;
Ok(CommitOutcome::insert(k, loc))
}))
.await?;
return Ok(HealOutcome::Healed {
from: current,
to: cand,
});
}
}
Ok(HealOutcome::Unrecoverable { location: current })
}
/// Run [`heal_table`](Self::heal_table) over every table in `ns`, returning
/// one outcome per table (in `list_tables` order). An unknown namespace is
/// an empty sweep, not an error — a fresh warehouse has nothing to heal.
///
/// This is the entry point an embedder calls **once, when it opens the
/// warehouse for writing**. Without such a sweep `heal_table` is a repair
/// nobody ever performs: it shipped with zero callers, and the live nornir
/// warehouse on oden consequently sat torn from the 2026-08-16 crash until
/// 2026-08-22 — six days of `nornir-embed … embed FAILED … DataInvalid`,
/// 19 repos per cycle, with the fix sitting unwired in this file.
///
/// Cost is a stat-and-parse pass, not a scan: MEASURED on oden's live
/// 73-table nornir warehouse, **0.46 s for all 73 tables in an unoptimised
/// debug build** (2026-08-22) — cheap enough to run unconditionally at open.
/// Never call it on a copied-aside read-only catalog: it would rewind the
/// throwaway copy's pointer and hide the torn state from the real one.
pub async fn heal_all_tables(
&self,
ns: &NamespaceIdent,
) -> Result<Vec<(TableIdent, HealOutcome)>> {
if !self.namespace_exists(ns).await? {
return Ok(Vec::new());
}
let mut out = Vec::new();
for ident in self.list_tables(ns).await? {
let outcome = self.heal_table(&ident).await?;
out.push((ident, outcome));
}
Ok(out)
}
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::sync::Arc;
use iceberg::io::LocalFsStorageFactory;
use iceberg::spec::{
ManifestContentType, ManifestFile, ManifestListWriter, NestedField, Operation,
PrimitiveType, Schema, Snapshot, SnapshotReference, SnapshotRetention, Summary, Type,
};
use iceberg::table::Table;
use iceberg::{
Catalog, CatalogBuilder, NamespaceIdent, TableCreation, TableIdent, TableRequirement,
TableUpdate,
};
use tempfile::TempDir;
use super::{newest_first, HealOutcome, durability_set, metadata_is_intact};
use crate::{RedbCatalog, RedbCatalogBuilder};
fn schema() -> Schema {
Schema::builder()
.with_schema_id(0)
.with_fields(vec![
NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
])
.build()
.unwrap()
}
async fn catalog(tmp: &TempDir) -> RedbCatalog {
let warehouse = tmp.path().join("warehouse");
std::fs::create_dir_all(&warehouse).unwrap();
RedbCatalogBuilder::default()
.db_path(tmp.path().join("catalog.redb"))
.warehouse_location(format!("file://{}", warehouse.display()))
.with_storage_factory(Arc::new(LocalFsStorageFactory))
.load("t", HashMap::new())
.await
.unwrap()
}
fn local(path: &str) -> String {
path.strip_prefix("file://").unwrap_or(path).to_string()
}
fn manifest_entry(path: &str, added_snapshot_id: i64, seq: i64) -> ManifestFile {
ManifestFile {
manifest_path: path.to_string(),
manifest_length: 8,
partition_spec_id: 0,
content: ManifestContentType::Data,
sequence_number: seq,
min_sequence_number: seq,
added_snapshot_id,
added_files_count: Some(1),
existing_files_count: Some(0),
deleted_files_count: Some(0),
added_rows_count: Some(1),
existing_rows_count: Some(0),
deleted_rows_count: Some(0),
partitions: None,
key_metadata: None,
first_row_id: None,
}
}
/// Commit a snapshot whose manifest list names `own` (added by this
/// snapshot) plus every `carried` entry (added by an ancestor). The
/// manifests are real, non-empty files — heal only stats them, and
/// `durability_set` only needs their paths.
async fn commit_snapshot(
cat: &RedbCatalog,
table: &Table,
tag: &str,
carried: Vec<ManifestFile>,
) -> (i64, String, String) {
let metadata = table.metadata();
let file_io = table.file_io().clone();
let parent_id = metadata.current_snapshot().map(|s| s.snapshot_id());
let seq = metadata.last_sequence_number() + 1;
let snapshot_id = 1_000 + seq;
let location = metadata.location().to_string();
let manifest_path = format!("{location}/metadata/{tag}-m0.avro");
std::fs::write(local(&manifest_path), b"manifest").unwrap();
let own = manifest_entry(&manifest_path, snapshot_id, seq);
let list_path = format!("{location}/metadata/snap-{snapshot_id}-{tag}.avro");
let mut lw = ManifestListWriter::v2(
file_io.new_output(&list_path).unwrap().writer().await.unwrap(),
snapshot_id,
parent_id,
seq,
);
lw.add_manifests(carried.into_iter().chain(std::iter::once(own.clone())))
.unwrap();
lw.close().await.unwrap();
let snapshot = Snapshot::builder()
.with_snapshot_id(snapshot_id)
.with_parent_snapshot_id(parent_id)
.with_sequence_number(seq)
.with_timestamp_ms(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
+ seq,
)
.with_manifest_list(list_path.clone())
.with_schema_id(metadata.current_schema().schema_id())
.with_summary(Summary {
operation: Operation::Append,
additional_properties: HashMap::new(),
})
.build();
cat.commit_table(
table.identifier().clone(),
vec![TableRequirement::RefSnapshotIdMatch {
r#ref: "main".into(),
snapshot_id: parent_id,
}],
vec![
TableUpdate::AddSnapshot { snapshot },
TableUpdate::SetSnapshotRef {
ref_name: "main".into(),
reference: SnapshotReference {
snapshot_id,
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: None,
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
},
},
],
)
.await
.unwrap();
(snapshot_id, manifest_path, list_path)
}
async fn fresh_table(tmp: &TempDir, cat: &RedbCatalog, name: &str) -> TableIdent {
let ns = NamespaceIdent::new("d".to_string());
if !cat.namespace_exists(&ns).await.unwrap() {
cat.create_namespace(&ns, HashMap::new()).await.unwrap();
}
let _ = tmp;
cat.create_table(
&ns,
TableCreation::builder()
.name(name.to_string())
.schema(schema())
.build(),
)
.await
.unwrap();
TableIdent::new(ns, name.to_string())
}
/// The durability set is the CHILDREN of the pointer, not just the pointer.
///
/// RED before the fix: `fsync_local_metadata` flushed only the metadata
/// JSON, so a crash could (and on oden 2026-08-16 did) leave a parseable
/// metadata naming a 0-byte manifest. This asserts the manifest THIS
/// snapshot added and the manifest list are flushed too — and that a
/// manifest carried forward from an ancestor is NOT re-flushed, so a
/// commit's cost does not grow with the table's manifest history.
#[tokio::test]
async fn durability_set_covers_new_manifest_and_list_not_carried_ones() {
let tmp = TempDir::new().unwrap();
let cat = catalog(&tmp).await;
let ident = fresh_table(&tmp, &cat, "t").await;
let t1 = cat.load_table(&ident).await.unwrap();
let (_, m1, _) = commit_snapshot(&cat, &t1, "one", Vec::new()).await;
let t2 = cat.load_table(&ident).await.unwrap();
let carried = t2
.metadata()
.current_snapshot()
.map(|snap| t2.manifest_list_reader(snap))
.unwrap()
.load()
.await
.unwrap()
.entries()
.to_vec();
let (_, m2, list2) = commit_snapshot(&cat, &t2, "two", carried).await;
let t3 = cat.load_table(&ident).await.unwrap();
let loc = t3.metadata_location_result().unwrap().to_string();
let set = durability_set(t3.file_io(), t3.metadata(), &loc).await;
// Order is children-first: new manifest, its list, then the pointer.
assert_eq!(set, vec![m2.clone(), list2, loc]);
// The ancestor's manifest is already durable — not re-flushed.
assert!(!set.contains(&m1));
}
/// A create-time metadata has no snapshot: the set is exactly the one blob
/// the metadata-only version flushed. Guards against the fix making an
/// empty-table create pay for manifests that do not exist.
#[tokio::test]
async fn durability_set_of_a_snapshotless_create_is_just_the_metadata() {
let tmp = TempDir::new().unwrap();
let cat = catalog(&tmp).await;
let ident = fresh_table(&tmp, &cat, "t").await;
let t = cat.load_table(&ident).await.unwrap();
let loc = t.metadata_location_result().unwrap().to_string();
assert_eq!(
durability_set(t.file_io(), t.metadata(), &loc).await,
vec![loc]
);
}
/// A 0-byte `*-m0.avro` under an intact metadata + intact manifest LIST is
/// the exact oden 2026-08-16 shape. Heal must roll the pointer back past
/// the torn commit, and `heal_all_tables` must be the thing that finds it.
///
/// The RED probe is in the test: it asserts the torn table really is
/// un-intact BEFORE healing, so a heal that silently did nothing could not
/// pass by landing on an already-healthy table.
#[tokio::test]
async fn heal_all_rolls_a_zero_byte_manifest_back_to_the_last_whole_snapshot() {
let tmp = TempDir::new().unwrap();
let cat = catalog(&tmp).await;
let ns = NamespaceIdent::new("d".to_string());
let torn = fresh_table(&tmp, &cat, "torn").await;
let whole = fresh_table(&tmp, &cat, "whole").await;
// Two snapshots on `torn`; one on `whole` (the control).
let t1 = cat.load_table(&torn).await.unwrap();
let (snap1, _, _) = commit_snapshot(&cat, &t1, "one", Vec::new()).await;
let t2 = cat.load_table(&torn).await.unwrap();
let carried = t2
.metadata()
.current_snapshot()
.map(|snap| t2.manifest_list_reader(snap))
.unwrap()
.load()
.await
.unwrap()
.entries()
.to_vec();
let (snap2, m2, _) = commit_snapshot(&cat, &t2, "two", carried).await;
let w = cat.load_table(&whole).await.unwrap();
commit_snapshot(&cat, &w, "one", Vec::new()).await;
let before = cat.load_table(&torn).await.unwrap();
let torn_loc = before.metadata_location_result().unwrap().to_string();
assert_eq!(
before.metadata().current_snapshot_id(),
Some(snap2),
"fixture must start on the torn snapshot"
);
// Tear it exactly the way the crash did: truncate the manifest to 0
// bytes, leaving the metadata JSON and the manifest LIST intact.
std::fs::write(local(&m2), b"").unwrap();
assert_eq!(std::fs::metadata(local(&m2)).unwrap().len(), 0);
// RED probe: the pointer is genuinely un-intact now. Without this, a
// no-op heal on a healthy table would look like a pass.
assert!(
!metadata_is_intact(&cat.fileio, &torn_loc, false).await,
"0-byte manifest must make the current pointer un-intact"
);
let outcomes = cat.heal_all_tables(&ns).await.unwrap();
let torn_out = outcomes
.iter()
.find(|(i, _)| i == &torn)
.map(|(_, o)| o.clone())
.unwrap();
match torn_out {
HealOutcome::Healed { from, to } => {
assert_eq!(from, torn_loc);
assert_ne!(to, torn_loc);
}
other => panic!("expected Healed, got {other:?}"),
}
// The untouched table is left alone — heal is not a blanket rewind.
assert_eq!(
outcomes
.iter()
.find(|(i, _)| i == &whole)
.map(|(_, o)| o.clone()),
Some(HealOutcome::Healthy)
);
// GREEN: the table now opens on the last whole snapshot.
let after = cat.load_table(&torn).await.unwrap();
assert_eq!(after.metadata().current_snapshot_id(), Some(snap1));
assert!(
metadata_is_intact(
&cat.fileio,
after.metadata_location_result().unwrap(),
false
)
.await
);
}
/// An empty namespace is an empty sweep, not an error.
#[tokio::test]
async fn heal_all_on_an_unknown_namespace_is_empty() {
let tmp = TempDir::new().unwrap();
let cat = catalog(&tmp).await;
let ns = NamespaceIdent::new("nope".to_string());
assert!(cat.heal_all_tables(&ns).await.unwrap().is_empty());
}
#[test]
fn newest_first_orders_by_numeric_version_not_lexicographically() {
// The 100000 boundary: lexicographically "99999" > "100000", so a naive
// string sort would wrongly rank v99999 as newest.
let got = newest_first(vec![
"00001-a.metadata.json".into(),
"99999-b.metadata.json".into(),
"100000-c.metadata.json".into(),
"100001-d.metadata.json".into(),
"00000-e.metadata.json".into(),
]);
let versions: Vec<&str> = got.iter().map(|s| s.split('-').next().unwrap()).collect();
assert_eq!(
versions,
vec!["100001", "100000", "99999", "00001", "00000"]
);
}
}