mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-14 14:46:53 +00:00
fix(table): tell a missing table apart from a lost claim on 404
Both arrive as 404 on the LSM routes, and `classify_lsm_fault` read every one as `ReissueFromFlush`. A typo'd table name therefore re-issued from flush until the cap and then reported a crash-looping node for a table that never existed. These routes bypass `check_table_response`, so `handle_table_not_found` — which every other route gets — never ran. The server now marks a registry miss with `InvalidTableState`, so `send_lsm_route` can split the two: a body that does not say so becomes `Error::TableNotFound`, built exactly as `handle_table_not_found` builds it. `classify_lsm_fault` keeps the same check so a direct caller cannot reintroduce the confusion, and an unparseable body reads as not-a-miss — the ambiguous case surfaces as plain not-found rather than driving a re-claim loop. Also fixes `checkpoint_fault` flattening terminal errors. It classified from a borrow and so had to rebuild the error, which turned anything that was not an `LsmRoute` into a generic `Runtime` message — including the `TableNotFound` above. It now takes the error by value and moves it through untouched, and the clone helper is gone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
106f9ddb31
commit
0bd7d04079
@@ -27,7 +27,7 @@ use crate::table::MergeBranchResult;
|
||||
use crate::table::MergeResult;
|
||||
use crate::table::Tags;
|
||||
use crate::table::UpdateResult;
|
||||
use crate::table::checkpoint::{GetLsmStatsResponse, classify_lsm_fault};
|
||||
use crate::table::checkpoint::{CODE_INVALID_TABLE_STATE, GetLsmStatsResponse, classify_lsm_fault};
|
||||
use crate::table::merge::MergeFilter;
|
||||
use crate::table::query::create_multi_vector_plan;
|
||||
use crate::table::{AlterColumnsResult, FieldMetadataUpdate, UpdateFieldMetadataResult};
|
||||
@@ -909,6 +909,17 @@ impl<S: HttpSend> RemoteTable<S> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether a 404 body is a registry miss (`InvalidTableState`) rather
|
||||
/// than the table not existing (`TableNotFound`). An unparseable body
|
||||
/// reads as "not a miss", so the ambiguous case surfaces as the plain
|
||||
/// not-found error instead of driving a re-claim loop.
|
||||
fn is_registry_miss(body: &str) -> bool {
|
||||
serde_json::from_str::<serde_json::Value>(body)
|
||||
.ok()
|
||||
.and_then(|v| v.get("code").and_then(|c| c.as_u64()))
|
||||
== Some(CODE_INVALID_TABLE_STATE)
|
||||
}
|
||||
|
||||
/// Send an LSM operator request and decode its JSON body, classifying
|
||||
/// a failure into an [`Error::LsmRoute`] the checkpoint loop can act on.
|
||||
///
|
||||
@@ -934,6 +945,20 @@ impl<S: HttpSend> RemoteTable<S> {
|
||||
let status = response.status();
|
||||
let body = response.text().await.err_to_http(request_id.clone())?;
|
||||
if !status.is_success() {
|
||||
// A 404 that is not a registry miss means the table does not
|
||||
// exist, and must arrive as the same error every other route
|
||||
// reports. These routes bypass `check_table_response`, so
|
||||
// `handle_table_not_found` never runs for them.
|
||||
if status == StatusCode::NOT_FOUND && !Self::is_registry_miss(&body) {
|
||||
return Err(Error::TableNotFound {
|
||||
name: self.name.clone(),
|
||||
source: Box::new(Error::Http {
|
||||
source: body.into(),
|
||||
request_id,
|
||||
status_code: Some(status),
|
||||
}),
|
||||
});
|
||||
}
|
||||
return Err(Error::LsmRoute {
|
||||
fault: classify_lsm_fault(status.as_u16(), &body),
|
||||
status: status.as_u16(),
|
||||
@@ -6163,6 +6188,66 @@ mod tests {
|
||||
assert_eq!(memtables[0].indexes, vec!["vec_idx".to_string()]);
|
||||
}
|
||||
|
||||
/// A 404 for a table that does not exist must arrive as
|
||||
/// `TableNotFound`, like every other route — not as a lost claim the
|
||||
/// loop re-issues from flush until its cap, then blames on a crash
|
||||
/// loop. The two are distinguished only by the body's `code`.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_missing_table_is_not_read_as_a_lost_claim() {
|
||||
let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = calls.clone();
|
||||
let table = Table::new_with_handler("my_table", move |_request| {
|
||||
seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
http::Response::builder()
|
||||
.status(404)
|
||||
.body(r#"{"code":4,"error":"Not found: Table not found: my_table"}"#.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
|
||||
let err = table.checkpoint_lsm().await.unwrap_err();
|
||||
assert!(
|
||||
matches!(err, Error::TableNotFound { .. }),
|
||||
"a missing table must say so: {err:?}"
|
||||
);
|
||||
assert_eq!(
|
||||
calls.load(std::sync::atomic::Ordering::SeqCst),
|
||||
1,
|
||||
"no point re-claiming a table that does not exist"
|
||||
);
|
||||
}
|
||||
|
||||
/// The other 404: the node lost its claim mid-loop. That one *does*
|
||||
/// re-issue from flush, which is the call that re-claims and replays.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_registry_miss_reissues_from_flush() {
|
||||
let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let seen = calls.clone();
|
||||
let table = Table::new_with_handler("my_table", move |request| {
|
||||
let path = request.url().path().to_string();
|
||||
let n = seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
if path.contains("flush_lsm") {
|
||||
// First flush lands; the claim is then lost, and the
|
||||
// re-issued flush succeeds.
|
||||
return accepted();
|
||||
}
|
||||
if path.contains("compact_lsm") {
|
||||
if n < 4 {
|
||||
return http::Response::builder()
|
||||
.status(404)
|
||||
.body(r#"{"code":19,"error":"WAL table not claimed"}"#.to_string())
|
||||
.unwrap();
|
||||
}
|
||||
return accepted();
|
||||
}
|
||||
ok_json(stats_body(if n < 6 { &[1] } else { &[] }, false))
|
||||
});
|
||||
|
||||
table
|
||||
.checkpoint_lsm()
|
||||
.await
|
||||
.expect("a lost claim must be recovered by re-flushing, not surfaced");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_lsm_stats_absent_when_wal_off() {
|
||||
let table = Table::new_with_handler("my_table", |_request| {
|
||||
|
||||
@@ -1692,7 +1692,7 @@ impl Table {
|
||||
// no-op, so a re-issue does not churn empty generations.
|
||||
match self.flush_lsm().await {
|
||||
Ok(()) => {}
|
||||
Err(e) => match checkpoint_fault(&e, attempt)? {
|
||||
Err(e) => match checkpoint_fault(e, attempt)? {
|
||||
CheckpointControl::Retry | CheckpointControl::ReissueFromFlush => {
|
||||
backoff(attempt).await;
|
||||
continue;
|
||||
@@ -1780,7 +1780,7 @@ impl Table {
|
||||
if !all_compacting {
|
||||
match self.compact_lsm().await {
|
||||
Ok(()) => {}
|
||||
Err(e) => match checkpoint_fault(&e, attempt)? {
|
||||
Err(e) => match checkpoint_fault(e, attempt)? {
|
||||
// Every bucket busy (429) is the state the poll
|
||||
// above already handles; fall through and re-read.
|
||||
CheckpointControl::Retry => {}
|
||||
|
||||
@@ -169,21 +169,30 @@ pub(crate) const CODE_INVALID_TABLE_STATE: u64 = 19;
|
||||
pub(crate) fn classify_lsm_fault(status: u16, body: &str) -> LsmFault {
|
||||
match status {
|
||||
429 => LsmFault::Retry,
|
||||
404 => LsmFault::ReissueFromFlush,
|
||||
503 => {
|
||||
let code = serde_json::from_str::<HashMap<String, serde_json::Value>>(body)
|
||||
.ok()
|
||||
.and_then(|m| m.get("code").and_then(|c| c.as_u64()));
|
||||
if code == Some(CODE_INVALID_TABLE_STATE) {
|
||||
LsmFault::Draining
|
||||
} else {
|
||||
LsmFault::Retry
|
||||
}
|
||||
}
|
||||
// A 404 is a registry miss only when the body says so. Left
|
||||
// unchecked, a typo'd table name — also a 404 — is read as "the
|
||||
// node lost its claim", and the loop re-issues from flush until the
|
||||
// cap and then blames a crash loop for a name that never existed.
|
||||
// `send_lsm_route` turns the other 404 into `TableNotFound` before
|
||||
// this is reached; the check stays so a direct caller cannot
|
||||
// reintroduce the confusion.
|
||||
404 if body_code(body) == Some(CODE_INVALID_TABLE_STATE) => LsmFault::ReissueFromFlush,
|
||||
404 => LsmFault::Fatal,
|
||||
503 if body_code(body) == Some(CODE_INVALID_TABLE_STATE) => LsmFault::Draining,
|
||||
503 => LsmFault::Retry,
|
||||
_ => LsmFault::Fatal,
|
||||
}
|
||||
}
|
||||
|
||||
/// The lance-namespace error code from a JSON error body, if it parses.
|
||||
/// Absent, truncated, or non-numeric ⇒ `None`, which every caller reads as
|
||||
/// the *less* terminal branch.
|
||||
fn body_code(body: &str) -> Option<u64> {
|
||||
serde_json::from_str::<HashMap<String, serde_json::Value>>(body)
|
||||
.ok()
|
||||
.and_then(|m| m.get("code").and_then(|c| c.as_u64()))
|
||||
}
|
||||
|
||||
/// How long the loop waits between `get_lsm_stats` polls. A pass over a
|
||||
/// `prefix_max` prefix takes seconds, so a shorter interval only adds
|
||||
/// round trips.
|
||||
@@ -231,20 +240,25 @@ pub(crate) enum CheckpointOutcome {
|
||||
ReissueFromFlush,
|
||||
}
|
||||
|
||||
/// Decide what a failed LSM request means for the loop. Returns `Err` for
|
||||
/// genuinely terminal faults — a draining node, a dropping table, a table
|
||||
/// with no LSM write path, anything unrecognized — which are errors the
|
||||
/// caller should see rather than outcomes to keep looping on.
|
||||
/// Decide what a failed LSM request means for the loop. Returns `Err` with
|
||||
/// the original error for genuinely terminal faults — a draining node, a
|
||||
/// dropping table, a missing table, anything unrecognized — which the
|
||||
/// caller should see rather than keep looping on.
|
||||
///
|
||||
/// Takes the error by value so a terminal one propagates *as itself*.
|
||||
/// Classifying from a borrow meant rebuilding it, and anything that was
|
||||
/// not an `LsmRoute` (a `TableNotFound`, say) got flattened to a generic
|
||||
/// runtime message on the way out.
|
||||
///
|
||||
/// Draining is terminal by design: the drain gate is a one-way latch that
|
||||
/// never releases until the process restarts, so retrying against that
|
||||
/// node spins forever when the truthful answer was in the first response.
|
||||
pub(crate) fn checkpoint_fault(
|
||||
e: &crate::Error,
|
||||
e: crate::Error,
|
||||
attempt: usize,
|
||||
) -> crate::Result<CheckpointControl> {
|
||||
#[cfg(feature = "remote")]
|
||||
let fault = match e {
|
||||
let fault = match &e {
|
||||
crate::Error::LsmRoute { fault, .. } => Some(*fault),
|
||||
_ => None,
|
||||
};
|
||||
@@ -256,29 +270,7 @@ pub(crate) fn checkpoint_fault(
|
||||
Some(LsmFault::ReissueFromFlush) if attempt < MAX_REISSUES => {
|
||||
Ok(CheckpointControl::ReissueFromFlush)
|
||||
}
|
||||
_ => Err(clone_lsm_error(e)),
|
||||
}
|
||||
}
|
||||
|
||||
/// `crate::Error` is not `Clone`, and the loop borrows the error to
|
||||
/// classify it before deciding whether to surface it. Rebuild the variant
|
||||
/// that matters and fall back to a flattened message for the rest.
|
||||
fn clone_lsm_error(e: &crate::Error) -> crate::Error {
|
||||
#[cfg(feature = "remote")]
|
||||
if let crate::Error::LsmRoute {
|
||||
fault,
|
||||
status,
|
||||
message,
|
||||
} = e
|
||||
{
|
||||
return crate::Error::LsmRoute {
|
||||
fault: *fault,
|
||||
status: *status,
|
||||
message: message.clone(),
|
||||
};
|
||||
}
|
||||
crate::Error::Runtime {
|
||||
message: e.to_string(),
|
||||
_ => Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -302,9 +294,15 @@ mod tests {
|
||||
LsmFault::Retry,
|
||||
"fenced / no-slot / transport stay retryable"
|
||||
);
|
||||
assert_eq!(
|
||||
classify_lsm_fault(404, r#"{"code":19}"#),
|
||||
LsmFault::ReissueFromFlush,
|
||||
"a registry miss says InvalidTableState and must re-claim"
|
||||
);
|
||||
assert_eq!(
|
||||
classify_lsm_fault(404, r#"{"code":4}"#),
|
||||
LsmFault::ReissueFromFlush
|
||||
LsmFault::Fatal,
|
||||
"a table that does not exist must not be retried as a lost claim"
|
||||
);
|
||||
assert_eq!(classify_lsm_fault(409, r#"{"code":14}"#), LsmFault::Fatal);
|
||||
assert_eq!(classify_lsm_fault(400, r#"{"code":13}"#), LsmFault::Fatal);
|
||||
|
||||
Reference in New Issue
Block a user