diff --git a/docs/src/js/interfaces/FunctionColumnAutoRefreshStatus.md b/docs/src/js/interfaces/FunctionColumnAutoRefreshStatus.md index bc8f7daab..3a203026c 100644 --- a/docs/src/js/interfaces/FunctionColumnAutoRefreshStatus.md +++ b/docs/src/js/interfaces/FunctionColumnAutoRefreshStatus.md @@ -27,7 +27,9 @@ optional blockedReason: | "dependency_in_flight" | "table_capacity" | "cancelled" - | "preparation_backoff"; + | "preparation_backoff" + | "failure_threshold_reached" + | "binding_revision_missing"; ``` *** @@ -85,6 +87,7 @@ optional nextRunAtReason: string; ```ts state: | "running" + | "queued" | "disabled" | "ready" | "up_to_date" diff --git a/docs/src/js/interfaces/FunctionColumnLastRefreshJob.md b/docs/src/js/interfaces/FunctionColumnLastRefreshJob.md index 1d7d3b300..a7c1df2aa 100644 --- a/docs/src/js/interfaces/FunctionColumnLastRefreshJob.md +++ b/docs/src/js/interfaces/FunctionColumnLastRefreshJob.md @@ -26,6 +26,16 @@ optional completedAtMillis: bigint; *** +### failure? + +```ts +optional failure: JobFailureInfo; +``` + +Structured failure details for a failed Job. + +*** + ### jobId ```ts diff --git a/docs/src/js/interfaces/FunctionColumnStatus.md b/docs/src/js/interfaces/FunctionColumnStatus.md index 755a8af81..0c299ee2b 100644 --- a/docs/src/js/interfaces/FunctionColumnStatus.md +++ b/docs/src/js/interfaces/FunctionColumnStatus.md @@ -58,6 +58,17 @@ function: FunctionColumnVersionStatus; *** +### lastFailureJob? + +```ts +optional lastFailureJob: FunctionColumnLastRefreshJob; +``` + +Newest failed refresh Job. This remains present when `lastRefreshJob` +is a later cancellation. + +*** + ### lastRefreshJob? ```ts diff --git a/docs/src/js/interfaces/JobFailureInfo.md b/docs/src/js/interfaces/JobFailureInfo.md index fe9ce04e3..cdea222c5 100644 --- a/docs/src/js/interfaces/JobFailureInfo.md +++ b/docs/src/js/interfaces/JobFailureInfo.md @@ -34,6 +34,16 @@ optional retryable: boolean; *** +### sqlJobId? + +```ts +optional sqlJobId: string; +``` + +Isolated SQL job that raised this failure, when supplied by the server. + +*** + ### traceback? ```ts diff --git a/docs/web/feature-engineering/refresh-column.mdx b/docs/web/feature-engineering/refresh-column.mdx index e1a9c43f3..ed4838497 100644 --- a/docs/web/feature-engineering/refresh-column.mdx +++ b/docs/web/feature-engineering/refresh-column.mdx @@ -197,6 +197,17 @@ println!("{}", final_status.failed); ``` +`auto_refresh.state` reports `queued` while the scheduler has accepted the +refresh but a worker has not started it; `running` means execution has started. +`last_refresh_job` describes the most recent attempt, including a cancellation +or a failure. If a later attempt was cancelled, `last_failure_job` retains the +most recent job that actually failed, so its nested `failure` continues to +explain the failed rows. + +These fields require a matching Agent, Query Node, and SDK deployment. An older +client cannot decode the `queued` state, and an older server can omit the new +optional failure fields. + The wait helpers do not submit or cancel a refresh. They retry only the typed finalization-pending response and propagate other errors. */} diff --git a/nodejs/__test__/remote.test.ts b/nodejs/__test__/remote.test.ts index 6c160ed03..f87c70091 100644 --- a/nodejs/__test__/remote.test.ts +++ b/nodejs/__test__/remote.test.ts @@ -681,14 +681,33 @@ describe("remote connection", () => { "stale":2, "never_computed":0, "failed":0, + "last_refresh_job":{ + "job_id":"job-cancelled", + "state":"cancelled", + "automatic":true, + "completed_at_millis":1790000002000 + }, + "last_failure_job":{ + "job_id":"job-failed", + "state":"failed", + "automatic":true, + "completed_at_millis":1790000001000, + "failure":{ + "phase":"finalize", + "message":"worker timed out", + "traceback":"Traceback: timeout", + "retryable":true, + "sql_job_id":"sql-job-7" + } + }, "auto_refresh":{ "enabled":true, "interval_seconds":60, - "state":"blocked", - "failure_streak":3, - "effective_cooldown_millis":8000, - "blocked_reason":"preparation_backoff", - "next_run_at_millis":null + "active_job_id":"job-queued", + "state":"queued", + "failure_streak":1, + "next_run_at_millis":null, + "next_run_at_reason":"job_queued_not_started" } }]`, ); @@ -701,10 +720,22 @@ describe("remote connection", () => { await table.checkout(42); const status = await table.functionColumnStatus("embedding"); expect(status.tableVersion).toBe(42n); - expect(status.autoRefresh.state).toBe("blocked"); - expect(status.autoRefresh.failureStreak).toBe(3); - expect(status.autoRefresh.effectiveCooldownMillis).toBe(8000n); - expect(status.autoRefresh.blockedReason).toBe("preparation_backoff"); + expect(status.autoRefresh.state).toBe("queued"); + expect(status.autoRefresh.activeJobId).toBe("job-queued"); + expect(status.autoRefresh.failureStreak).toBe(1); + expect(status.autoRefresh.nextRunAtReason).toBe( + "job_queued_not_started", + ); + expect(status.lastRefreshJob?.state).toBe("cancelled"); + expect(status.lastRefreshJob?.failure).toBeUndefined(); + expect(status.lastFailureJob?.jobId).toBe("job-failed"); + expect(status.lastFailureJob?.failure).toEqual({ + phase: "finalize", + message: "worker timed out", + traceback: "Traceback: timeout", + retryable: true, + sqlJobId: "sql-job-7", + }); }, ); expect(queries).toEqual([{ column: "embedding", version: "42" }]); diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index 99eb0d1f9..703fc7926 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -35,6 +35,7 @@ import { FunctionErrorsOptions, IndexConfig, IndexStatistics, + JobFailureInfo, Branches as NativeBranches, OptimizeStats, RTEStats, @@ -274,6 +275,8 @@ export interface FunctionColumnLastRefreshJob { state: string; automatic?: boolean; completedAtMillis?: bigint; + /** Structured failure details for a failed Job. */ + failure?: JobFailureInfo; } /** Automatic-refresh projection for a Function output. */ @@ -284,6 +287,7 @@ export interface FunctionColumnAutoRefreshStatus { state: | "disabled" | "running" + | "queued" | "ready" | "up_to_date" | "waiting_interval" @@ -298,7 +302,9 @@ export interface FunctionColumnAutoRefreshStatus { | "dependency_in_flight" | "table_capacity" | "cancelled" - | "preparation_backoff"; + | "preparation_backoff" + | "failure_threshold_reached" + | "binding_revision_missing"; nextRunAtMillis: bigint | null; nextRunAtReason?: string; } @@ -315,6 +321,11 @@ export interface FunctionColumnStatus { neverComputed: bigint; failed: bigint; lastRefreshJob?: FunctionColumnLastRefreshJob; + /** + * Newest failed refresh Job. This remains present when `lastRefreshJob` + * is a later cancellation. + */ + lastFailureJob?: FunctionColumnLastRefreshJob; autoRefresh: FunctionColumnAutoRefreshStatus; } diff --git a/nodejs/src/job.rs b/nodejs/src/job.rs index 485035d33..09b245f97 100644 --- a/nodejs/src/job.rs +++ b/nodejs/src/job.rs @@ -138,12 +138,7 @@ impl Job { /// Why the job failed, when it failed and the server reports a reason. #[napi(getter)] pub fn failure(&self) -> Option { - self.inner.failure().map(|failure| JobFailureInfo { - phase: failure.phase, - message: failure.message, - traceback: failure.traceback, - retryable: failure.retryable, - }) + self.inner.failure().map(Into::into) } /// This job's recorded lifecycle events, as an Arrow IPC stream buffer. @@ -216,10 +211,25 @@ impl From for JobInfo { /// The server's account of why a job failed. #[napi(object)] +#[derive(Clone, Debug)] pub struct JobFailureInfo { pub phase: Option, pub message: Option, /// User-code traceback, possibly truncated by the server. pub traceback: Option, pub retryable: Option, + /// Isolated SQL job that raised this failure, when supplied by the server. + pub sql_job_id: Option, +} + +impl From for JobFailureInfo { + fn from(failure: lancedb::error::JobFailure) -> Self { + Self { + phase: failure.phase, + message: failure.message, + traceback: failure.traceback, + retryable: failure.retryable, + sql_job_id: failure.sql_job_id, + } + } } diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index f6d3e538e..62fe811ad 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -20,6 +20,7 @@ use napi_derive::napi; use crate::blob::{BlobFile, copy_blob_buffers, parse_row_ids, spawn_abortable}; use crate::error::NapiErrorExt; use crate::index::Index; +use crate::job::JobFailureInfo; use crate::merge::NativeMergeInsertBuilder; use crate::query::{Query, TakeQuery, VectorQuery}; use crate::util::schema_to_buffer; @@ -1868,6 +1869,19 @@ pub struct FunctionColumnLastRefreshJob { pub state: String, pub automatic: Option, pub completed_at_millis: Option, + pub failure: Option, +} + +impl From for FunctionColumnLastRefreshJob { + fn from(job: lancedb::table::FunctionColumnLastRefreshJob) -> Self { + Self { + job_id: job.job_id, + state: job.state, + automatic: job.automatic, + completed_at_millis: job.completed_at_millis.map(BigInt::from), + failure: job.failure.map(Into::into), + } + } } /// Automatic-refresh projection for a Function output. @@ -1899,6 +1913,7 @@ pub struct FunctionColumnStatus { pub never_computed: BigInt, pub failed: BigInt, pub last_refresh_job: Option, + pub last_failure_job: Option, pub auto_refresh: FunctionColumnAutoRefreshStatus, } @@ -1921,14 +1936,8 @@ impl From for FunctionColumnStatus { stale: BigInt::from(status.stale), never_computed: BigInt::from(status.never_computed), failed: BigInt::from(status.failed), - last_refresh_job: status - .last_refresh_job - .map(|job| FunctionColumnLastRefreshJob { - job_id: job.job_id, - state: job.state, - automatic: job.automatic, - completed_at_millis: job.completed_at_millis.map(BigInt::from), - }), + last_refresh_job: status.last_refresh_job.map(Into::into), + last_failure_job: status.last_failure_job.map(Into::into), auto_refresh: FunctionColumnAutoRefreshStatus { enabled: status.auto_refresh.enabled, interval_seconds: status.auto_refresh.interval_seconds.map(BigInt::from), diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 07306c1d9..4c5f5758a 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -1108,6 +1108,7 @@ class FunctionColumnLastRefreshJob: state: str automatic: Optional[bool] completed_at_millis: Optional[int] + failure: Optional[JobFailureInfo] class FunctionColumnAutoRefreshStatus: enabled: bool @@ -1131,6 +1132,7 @@ class FunctionColumnStatus: never_computed: int failed: int last_refresh_job: Optional[FunctionColumnLastRefreshJob] + last_failure_job: Optional[FunctionColumnLastRefreshJob] auto_refresh: FunctionColumnAutoRefreshStatus class RefreshMaterializedViewResult: diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index fd933b43c..407596d26 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -2306,15 +2306,33 @@ def test_remote_function_column_status_passes_pinned_version(): "stale": 2, "never_computed": 0, "failed": 0, + "last_refresh_job": { + "job_id": "job-cancelled", + "state": "cancelled", + "automatic": True, + "completed_at_millis": 1790000002000, + }, + "last_failure_job": { + "job_id": "job-failed", + "state": "failed", + "automatic": True, + "completed_at_millis": 1790000001000, + "failure": { + "phase": "finalize", + "message": "worker timed out", + "traceback": "Traceback: timeout", + "retryable": True, + "sql_job_id": "sql-job-7", + }, + }, "auto_refresh": { "enabled": True, "interval_seconds": 60, - "active_job_id": None, - "state": "blocked", - "failure_streak": 3, - "effective_cooldown_millis": 8000, - "blocked_reason": "preparation_backoff", + "active_job_id": "job-queued", + "state": "queued", + "failure_streak": 1, "next_run_at_millis": None, + "next_run_at_reason": "job_queued_not_started", }, } @@ -2350,10 +2368,19 @@ def test_remote_function_column_status_passes_pinned_version(): 0, 0, ) - assert result.auto_refresh.state == "blocked" - assert result.auto_refresh.failure_streak == 3 - assert result.auto_refresh.effective_cooldown_millis == 8000 - assert result.auto_refresh.blocked_reason == "preparation_backoff" + assert result.auto_refresh.state == "queued" + assert result.auto_refresh.active_job_id == "job-queued" + assert result.auto_refresh.failure_streak == 1 + assert result.auto_refresh.next_run_at_reason == "job_queued_not_started" + assert result.last_refresh_job.state == "cancelled" + assert result.last_refresh_job.failure is None + assert result.last_failure_job.job_id == "job-failed" + failure = result.last_failure_job.failure + assert failure.phase == "finalize" + assert failure.message == "worker timed out" + assert failure.traceback == "Traceback: timeout" + assert failure.retryable is True + assert failure.sql_job_id == "sql-job-7" def test_remote_wait_fresh_retries_only_typed_finalization_pending(): diff --git a/python/src/job.rs b/python/src/job.rs index 79a9cd524..f1fd85844 100644 --- a/python/src/job.rs +++ b/python/src/job.rs @@ -228,7 +228,7 @@ impl From for JobInfo { /// The server's account of why a job failed. #[pyclass(module = "lancedb._lancedb", get_all, skip_from_py_object)] -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct JobFailureInfo { phase: Option, message: Option, @@ -238,6 +238,18 @@ pub struct JobFailureInfo { sql_job_id: Option, } +impl From for JobFailureInfo { + fn from(failure: lancedb::error::JobFailure) -> Self { + Self { + phase: failure.phase, + message: failure.message, + traceback: failure.traceback, + retryable: failure.retryable, + sql_job_id: failure.sql_job_id, + } + } +} + #[pymethods] impl JobFailureInfo { fn __repr__(&self) -> String { @@ -341,13 +353,7 @@ impl From for JobDescription { .filter(|result| !result.is_null()) .map(|result| result.to_string()), _progress_json: Some(progress_json), - failure: description.failure.map(|failure| JobFailureInfo { - phase: failure.phase, - message: failure.message, - traceback: failure.traceback, - retryable: failure.retryable, - sql_job_id: failure.sql_job_id, - }), + failure: description.failure.map(Into::into), } } } diff --git a/python/src/table.rs b/python/src/table.rs index cf627961c..c4211e44d 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -15,6 +15,7 @@ use crate::{ error::PythonErrorExt, expr::PyExpr, index::{CodeAnalyzerOptions, IndexConfig, extract_index_params}, + job::JobFailureInfo, query::{Query, TakeQuery}, table::scannable::PyScannable, }; @@ -925,6 +926,7 @@ pub struct FunctionColumnLastRefreshJob { pub state: String, pub automatic: Option, pub completed_at_millis: Option, + pub failure: Option, } impl From for FunctionColumnLastRefreshJob { @@ -934,6 +936,7 @@ impl From for FunctionColumnLastRe state: job.state, automatic: job.automatic, completed_at_millis: job.completed_at_millis, + failure: job.failure.map(Into::into), } } } @@ -985,6 +988,7 @@ pub struct FunctionColumnStatus { pub never_computed: u64, pub failed: u64, pub last_refresh_job: Option, + pub last_failure_job: Option, pub auto_refresh: FunctionColumnAutoRefreshStatus, } @@ -1017,6 +1021,7 @@ impl From for FunctionColumnStatus { never_computed: status.never_computed, failed: status.failed, last_refresh_job: status.last_refresh_job.map(Into::into), + last_failure_job: status.last_failure_job.map(Into::into), auto_refresh: status.auto_refresh.into(), } } diff --git a/rust/lancedb/src/error.rs b/rust/lancedb/src/error.rs index 853ca831f..89ad7a229 100644 --- a/rust/lancedb/src/error.rs +++ b/rust/lancedb/src/error.rs @@ -253,24 +253,48 @@ impl snafu::AsErrorSource for InterruptionSource { /// A job run remotely carries whatever the server reported, which older servers /// do not report at all. Every field is absent rather than invented when the /// backend does not supply it. -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone, Default, serde::Deserialize, serde::Serialize)] pub struct JobFailure { /// The stage the job was in, when known. + #[serde(default, skip_serializing_if = "Option::is_none")] pub phase: Option, /// A human-readable reason, when known. + #[serde(default, skip_serializing_if = "Option::is_none")] pub message: Option, /// User-code traceback, possibly truncated by the server. + #[serde(default, skip_serializing_if = "Option::is_none")] pub traceback: Option, /// Whether a retry could clear the failure, when known. + #[serde(default, skip_serializing_if = "Option::is_none")] pub retryable: Option, /// Isolated SQL job that raised this failure, when the server supplies /// one across the Flight SQL boundary. + #[serde(default, skip_serializing_if = "Option::is_none")] pub sql_job_id: Option, /// The underlying error from a job run in this process or a structured /// remote Flight SQL failure. + #[serde(skip)] pub source: Option>, } +impl PartialEq for JobFailure { + fn eq(&self, other: &Self) -> bool { + let source_matches = match (&self.source, &other.source) { + (None, None) => true, + (Some(left), Some(right)) => Arc::ptr_eq(left, right), + _ => false, + }; + self.phase == other.phase + && self.message == other.message + && self.traceback == other.traceback + && self.retryable == other.retryable + && self.sql_job_id == other.sql_job_id + && source_matches + } +} + +impl Eq for JobFailure {} + impl JobFailure { /// A failure whose only known detail is the error that caused it. pub(crate) fn from_source(source: Arc) -> Self { diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 222ff527b..37890fd17 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -11655,12 +11655,13 @@ mod tests { } #[tokio::test] - async fn test_function_column_status_decodes_blocked_scheduler_fields() { + async fn test_function_column_status_decodes_scheduler_lifecycle_and_failures() { let table = Table::new_with_handler("my_table", |_request| { http::Response::builder() .status(200) .body( - r#"[{ + r#"[ + { "column":"embedding", "binding_id":"fb_01", "function":{ @@ -11676,6 +11677,101 @@ mod tests { "stale":2, "never_computed":0, "failed":0, + "last_refresh_job":{ + "job_id":"job-cancelled", + "state":"cancelled", + "automatic":true, + "completed_at_millis":1790000002000 + }, + "last_failure_job":{ + "job_id":"job-failed", + "state":"failed", + "automatic":true, + "completed_at_millis":1790000001000, + "failure":{ + "phase":"finalize", + "message":"worker timed out", + "traceback":"Traceback: timeout", + "retryable":true, + "sql_job_id":"sql-job-7" + } + }, + "auto_refresh":{ + "enabled":true, + "interval_seconds":60, + "active_job_id":"job-queued", + "state":"queued", + "failure_streak":1, + "next_run_at_millis":null, + "next_run_at_reason":"job_queued_not_started" + } + }, + { + "column":"rerank", + "binding_id":"fb_02", + "function":{ + "name":"rerank", + "object_id":"rerank", + "location":"objects/rerank", + "version":"3", + "manifest_digest":"sha256:def" + }, + "table_version":42, + "rows":0, + "fresh":0, + "stale":0, + "never_computed":0, + "failed":0, + "auto_refresh":{ + "enabled":true, + "interval_seconds":60, + "state":"blocked", + "failure_streak":5, + "blocked_reason":"failure_threshold_reached", + "next_run_at_millis":null + } + }, + { + "column":"classify", + "binding_id":"fb_03", + "function":{ + "name":"classify", + "object_id":"classify", + "location":"objects/classify", + "version":"2", + "manifest_digest":"sha256:ghi" + }, + "table_version":42, + "rows":0, + "fresh":0, + "stale":0, + "never_computed":0, + "failed":0, + "auto_refresh":{ + "enabled":true, + "interval_seconds":60, + "state":"blocked", + "failure_streak":0, + "blocked_reason":"binding_revision_missing", + "next_run_at_millis":null + } + }, + { + "column":"summarize", + "binding_id":"fb_04", + "function":{ + "name":"summarize", + "object_id":"summarize", + "location":"objects/summarize", + "version":"4", + "manifest_digest":"sha256:jkl" + }, + "table_version":42, + "rows":0, + "fresh":0, + "stale":0, + "never_computed":0, + "failed":0, "auto_refresh":{ "enabled":true, "interval_seconds":60, @@ -11690,15 +11786,52 @@ mod tests { .unwrap() }); - let status = table.function_column_status("embedding").await.unwrap(); + let statuses = table.function_column_statuses(None).await.unwrap(); + assert_eq!(statuses.len(), 4); + let status = &statuses[0]; assert_eq!( status.auto_refresh.state, - crate::table::FunctionColumnAutoRefreshState::Blocked + crate::table::FunctionColumnAutoRefreshState::Queued ); - assert_eq!(status.auto_refresh.failure_streak, 3); - assert_eq!(status.auto_refresh.effective_cooldown_millis, Some(8000)); assert_eq!( - status.auto_refresh.blocked_reason, + status.auto_refresh.active_job_id.as_deref(), + Some("job-queued") + ); + assert_eq!( + status.auto_refresh.next_run_at_reason.as_deref(), + Some("job_queued_not_started") + ); + assert_eq!( + status + .last_refresh_job + .as_ref() + .map(|job| job.state.as_str()), + Some("cancelled") + ); + let failure_job = status.last_failure_job.as_ref().unwrap(); + assert_eq!(failure_job.job_id, "job-failed"); + let failure = failure_job.failure.as_ref().unwrap(); + assert_eq!(failure.phase.as_deref(), Some("finalize")); + assert_eq!(failure.message.as_deref(), Some("worker timed out")); + assert_eq!(failure.retryable, Some(true)); + assert_eq!(failure.sql_job_id.as_deref(), Some("sql-job-7")); + + assert_eq!(statuses[1].auto_refresh.failure_streak, 5); + assert_eq!( + statuses[1].auto_refresh.blocked_reason, + Some(crate::table::FunctionColumnAutoRefreshBlockedReason::FailureThresholdReached) + ); + assert_eq!( + statuses[2].auto_refresh.blocked_reason, + Some(crate::table::FunctionColumnAutoRefreshBlockedReason::BindingRevisionMissing) + ); + assert_eq!(statuses[3].auto_refresh.failure_streak, 3); + assert_eq!( + statuses[3].auto_refresh.effective_cooldown_millis, + Some(8000) + ); + assert_eq!( + statuses[3].auto_refresh.blocked_reason, Some(crate::table::FunctionColumnAutoRefreshBlockedReason::PreparationBackoff) ); } diff --git a/rust/lancedb/src/table/function_column_status.rs b/rust/lancedb/src/table/function_column_status.rs index 729ea33ba..15703f4e9 100644 --- a/rust/lancedb/src/table/function_column_status.rs +++ b/rust/lancedb/src/table/function_column_status.rs @@ -5,6 +5,7 @@ use serde::{Deserialize, Serialize}; +use crate::error::JobFailure; use crate::function::FunctionVersionRef; use crate::{Error, Result}; @@ -23,6 +24,9 @@ pub struct FunctionColumnLastRefreshJob { pub automatic: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub completed_at_millis: Option, + /// Why this Job failed, when the server retained a structured failure. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub failure: Option, } /// Automatic-refresh scheduling state that the server can prove. @@ -31,6 +35,7 @@ pub struct FunctionColumnLastRefreshJob { pub enum FunctionColumnAutoRefreshState { Disabled, Running, + Queued, UpToDate, Ready, WaitingInterval, @@ -44,6 +49,7 @@ impl FunctionColumnAutoRefreshState { match self { Self::Disabled => "disabled", Self::Running => "running", + Self::Queued => "queued", Self::UpToDate => "up_to_date", Self::Ready => "ready", Self::WaitingInterval => "waiting_interval", @@ -64,6 +70,8 @@ pub enum FunctionColumnAutoRefreshBlockedReason { TableCapacity, Cancelled, PreparationBackoff, + FailureThresholdReached, + BindingRevisionMissing, } impl FunctionColumnAutoRefreshBlockedReason { @@ -75,6 +83,8 @@ impl FunctionColumnAutoRefreshBlockedReason { Self::TableCapacity => "table_capacity", Self::Cancelled => "cancelled", Self::PreparationBackoff => "preparation_backoff", + Self::FailureThresholdReached => "failure_threshold_reached", + Self::BindingRevisionMissing => "binding_revision_missing", } } } @@ -113,6 +123,10 @@ pub struct FunctionColumnStatus { pub failed: u64, #[serde(default, skip_serializing_if = "Option::is_none")] pub last_refresh_job: Option, + /// Newest failed refresh Job. Unlike `last_refresh_job`, this remains + /// visible when the latest attempt was cancelled. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_failure_job: Option, pub auto_refresh: FunctionColumnAutoRefreshStatus, } @@ -216,6 +230,7 @@ mod tests { never_computed: 1, failed: 1, last_refresh_job: None, + last_failure_job: None, auto_refresh: FunctionColumnAutoRefreshStatus { enabled: true, interval_seconds: Some(60), @@ -253,6 +268,7 @@ mod tests { state: "finished".to_string(), automatic: Some(true), completed_at_millis: Some(1_790_000_000_000), + failure: None, }); let encoded = serde_json::to_value(value).unwrap(); assert_eq!(encoded["last_refresh_job"]["job_id"], "job-7"); @@ -264,11 +280,31 @@ mod tests { } #[test] - fn up_to_date_wire_is_stable() { - assert_eq!( - serde_json::to_string(&FunctionColumnAutoRefreshState::UpToDate).unwrap(), - "\"up_to_date\"" - ); + fn scheduler_wire_names_are_stable() { + for (state, expected) in [ + (FunctionColumnAutoRefreshState::Queued, "queued"), + (FunctionColumnAutoRefreshState::UpToDate, "up_to_date"), + ] { + assert_eq!( + serde_json::to_string(&state).unwrap(), + format!("\"{expected}\"") + ); + } + for (reason, expected) in [ + ( + FunctionColumnAutoRefreshBlockedReason::FailureThresholdReached, + "failure_threshold_reached", + ), + ( + FunctionColumnAutoRefreshBlockedReason::BindingRevisionMissing, + "binding_revision_missing", + ), + ] { + assert_eq!( + serde_json::to_string(&reason).unwrap(), + format!("\"{expected}\"") + ); + } } #[test]