use std::fmt;
use query_planning::error::{
AllocationSource, AttemptFailureCode, QueryError, QueryErrorCode, QueryErrorStatus, Result,
};
use serde::{
Deserialize, Serialize,
de::{self, SeqAccess, Visitor},
};
use crate::observability::{AttemptComposedScheduleProjection, AttemptId};
pub const QUERY_ATTEMPT_REPORT_SCHEMA: &str = "query_engine.query_attempt_report";
pub const SEALED_ATTEMPT_PROJECTION_SCHEMA: &str = "query_engine.sealed_attempt_projection";
pub const MAX_ATTEMPT_QUERY_ID_BYTES: usize = 256;
pub const MAX_ATTEMPT_TRACE_CONTEXT_BYTES: usize = 1024;
macro_rules! attempt_correlation {
($name:ident, $limit:ident, $field:literal, $doc:literal) => {
#[doc = $doc]
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)]
#[serde(transparent)]
pub struct $name(String);
impl $name {
pub fn new(value: impl Into<String>) -> Result<Self> {
let value = value.into();
if value.trim().is_empty() || value.len() > $limit {
return Err(QueryError::new(
QueryErrorCode::InvalidAttemptCorrelation,
"query_engine.attempt.correlation",
"attempt correlation must be non-blank and within its byte limit",
)
.with_fact("field", $field)
.with_fact("max_bytes", $limit)
.with_fact("reason", if value.trim().is_empty() { "blank" } else { "too_long" }));
}
Ok(Self(value))
}
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for $name {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.0)
}
}
impl<'de> Deserialize<'de> for $name {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where D: serde::Deserializer<'de> {
let value = String::deserialize(deserializer)?;
Self::new(value).map_err(de::Error::custom)
}
}
};
}
attempt_correlation!(
AttemptQueryId,
MAX_ATTEMPT_QUERY_ID_BYTES,
"query_id",
"Opaque caller query correlation, limited to 256 UTF-8 bytes and required to be non-blank."
);
attempt_correlation!(
AttemptTraceContext,
MAX_ATTEMPT_TRACE_CONTEXT_BYTES,
"trace_context",
"Opaque caller trace correlation, limited to 1024 UTF-8 bytes and required to be non-blank."
);
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptDisposition {
Unplanned,
CpuFallback,
PartialNative,
Native,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptCleanupDisposition {
NotRequired,
Completed,
CompletedWithFailures,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptCapacityDisposition {
NotAdmitted,
Released,
FrozenResident,
RetainedQuarantine,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptHealthTransition {
HealthyReuse,
DeviceQuarantined,
BackendUnhealthy,
}
pub(crate) const MAX_ATTEMPT_FAILURE_TEXT_BYTES: usize = 256;
#[cfg(test)]
const MAX_ATTEMPT_FAILURE_TEXT_INSPECTION_BYTES: usize = 1024;
const MAX_ATTEMPT_FAILURE_FACTS: usize = 4;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptFailureOrigin {
Native,
Datafusion,
AttemptControl,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptFailureStatus {
NotApplicable,
Permanent,
Temporary,
Persistent,
}
impl AttemptFailureStatus {
pub(crate) const fn from_query_error_status(status: QueryErrorStatus) -> Self {
match status {
QueryErrorStatus::Permanent => Self::Permanent,
QueryErrorStatus::Temporary => Self::Temporary,
QueryErrorStatus::Persistent => Self::Persistent,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptFailureOperation {
Planning,
Admission,
Execution,
Cleanup,
Datafusion,
AttemptControl,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptFailureFactKey {
Reason,
AllocationSource,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptFailureFactValue {
BackendClosing,
DeadlineExpired,
CallerCancelled,
AdmissionQueueFull,
QueueWaitElapsed,
MemoryGrantWaitTimeout,
RequiredFloorExceedsDeviceCapacity,
QueryCapExceedsResidentCapacity,
BackendUnhealthy,
CugraphRmmOutOfMemory,
CugraphRmmBadAlloc,
CugraphStdBadAlloc,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptFailureFact {
key: AttemptFailureFactKey,
value: AttemptFailureFactValue,
}
impl AttemptFailureFact {
pub const fn key(&self) -> AttemptFailureFactKey {
self.key
}
pub const fn value(&self) -> AttemptFailureFactValue {
self.value
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub(crate) struct AttemptFailureText {
value: String,
truncated: bool,
redacted: bool,
invalid_utf8: bool,
}
#[derive(Deserialize)]
struct AttemptFailureTextWire {
#[serde(deserialize_with = "deserialize_bounded_attempt_failure_text_value")]
value: String,
truncated: bool,
redacted: bool,
invalid_utf8: bool,
}
fn deserialize_bounded_attempt_failure_text_value<'de, D>(deserializer: D) -> std::result::Result<String, D::Error>
where D: serde::Deserializer<'de> {
struct BoundedTextVisitor;
impl Visitor<'_> for BoundedTextVisitor {
type Value = String;
fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(formatter, "a UTF-8 string no longer than {MAX_ATTEMPT_FAILURE_TEXT_BYTES} bytes")
}
fn visit_str<E>(self, value: &str) -> std::result::Result<Self::Value, E>
where E: de::Error {
if value.len() > MAX_ATTEMPT_FAILURE_TEXT_BYTES {
return Err(E::custom("attempt failure text exceeds its byte bound"));
}
Ok(value.to_owned())
}
fn visit_string<E>(self, value: String) -> std::result::Result<Self::Value, E>
where E: de::Error {
if value.len() > MAX_ATTEMPT_FAILURE_TEXT_BYTES {
return Err(E::custom("attempt failure text exceeds its byte bound"));
}
Ok(value)
}
}
deserializer.deserialize_string(BoundedTextVisitor)
}
impl<'de> Deserialize<'de> for AttemptFailureText {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where D: serde::Deserializer<'de> {
let wire = AttemptFailureTextWire::deserialize(deserializer)?;
if wire.value.len() > MAX_ATTEMPT_FAILURE_TEXT_BYTES {
return Err(de::Error::custom("attempt failure text exceeds its byte bound"));
}
if wire.redacted {
if wire.value != "<redacted>" {
return Err(de::Error::custom("redacted attempt failure text must not retain source text"));
}
} else if attempt_failure_text_requires_redaction(&wire.value.to_ascii_lowercase()) {
return Err(de::Error::custom("attempt failure text requires redaction"));
}
Ok(Self {
value: wire.value,
truncated: wire.truncated,
redacted: wire.redacted,
invalid_utf8: wire.invalid_utf8,
})
}
}
impl AttemptFailureText {
#[cfg(test)]
pub(crate) fn from_bytes(bytes: &[u8]) -> Self {
let inspected_len = bytes.len().min(MAX_ATTEMPT_FAILURE_TEXT_INSPECTION_BYTES);
let inspected = &bytes[..inspected_len];
let invalid_utf8 = std::str::from_utf8(inspected).is_err();
let mut value = String::from_utf8_lossy(inspected).split_whitespace().collect::<Vec<_>>().join(" ");
let mut redacted = false;
let lowercase = value.to_ascii_lowercase();
if attempt_failure_text_requires_redaction(&lowercase) {
value = "<redacted>".to_string();
redacted = true;
}
let (value, truncated_by_bound) = truncate_utf8(&value, MAX_ATTEMPT_FAILURE_TEXT_BYTES);
Self { value, truncated: bytes.len() > inspected_len || truncated_by_bound, redacted, invalid_utf8 }
}
pub fn value(&self) -> &str {
&self.value
}
#[cfg(test)]
pub(crate) const fn truncated(&self) -> bool {
self.truncated
}
#[cfg(test)]
pub(crate) const fn redacted(&self) -> bool {
self.redacted
}
#[cfg(test)]
pub(crate) const fn invalid_utf8(&self) -> bool {
self.invalid_utf8
}
}
fn attempt_failure_text_requires_redaction(lowercase: &str) -> bool {
lowercase.contains("://")
|| lowercase.contains("password")
|| lowercase.contains("passwd")
|| lowercase.contains("credential")
|| lowercase.contains("authorization")
|| lowercase.contains("token")
|| lowercase.contains("secret")
|| lowercase.contains("api_key")
|| lowercase.contains("access_key")
|| lowercase.contains("bearer ")
|| looks_like_sql_statement(lowercase)
}
fn looks_like_sql_statement(value: &str) -> bool {
let value = value.trim_start();
["select", "insert", "update", "delete", "merge", "with", "create", "alter", "drop"].into_iter().any(|keyword| {
value.strip_prefix(keyword).is_some_and(|suffix| {
suffix.is_empty()
|| suffix
.chars()
.next()
.is_some_and(|character| character.is_whitespace() || matches!(character, '*' | '('))
})
})
}
#[cfg(test)]
fn truncate_utf8(value: &str, max_bytes: usize) -> (String, bool) {
if value.len() <= max_bytes {
return (value.to_string(), false);
}
let mut end = max_bytes;
while !value.is_char_boundary(end) {
end -= 1;
}
(value[..end].to_string(), true)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AttemptFailureSummary {
origin: AttemptFailureOrigin,
code: AttemptFailureCode,
status: AttemptFailureStatus,
operation: AttemptFailureOperation,
text: Option<AttemptFailureText>,
facts: Vec<AttemptFailureFact>,
}
#[derive(Deserialize)]
struct AttemptFailureSummaryWire {
origin: AttemptFailureOrigin,
code: AttemptFailureCode,
status: AttemptFailureStatus,
operation: AttemptFailureOperation,
#[serde(default)]
text: Option<AttemptFailureText>,
#[serde(default, deserialize_with = "deserialize_bounded_attempt_failure_facts")]
facts: Vec<AttemptFailureFact>,
}
fn deserialize_bounded_attempt_failure_facts<'de, D>(
deserializer: D,
) -> std::result::Result<Vec<AttemptFailureFact>, D::Error>
where D: serde::Deserializer<'de> {
struct BoundedFactsVisitor;
impl<'de> Visitor<'de> for BoundedFactsVisitor {
type Value = Vec<AttemptFailureFact>;
fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(formatter, "at most {MAX_ATTEMPT_FAILURE_FACTS} attempt failure facts")
}
fn visit_seq<A>(self, mut sequence: A) -> std::result::Result<Self::Value, A::Error>
where A: SeqAccess<'de> {
let mut facts = Vec::with_capacity(MAX_ATTEMPT_FAILURE_FACTS);
while let Some(fact) = sequence.next_element()? {
if facts.len() == MAX_ATTEMPT_FAILURE_FACTS {
return Err(de::Error::custom("attempt failure facts exceed their fixed bound"));
}
facts.push(fact);
}
Ok(facts)
}
}
deserializer.deserialize_seq(BoundedFactsVisitor)
}
impl<'de> Deserialize<'de> for AttemptFailureSummary {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where D: serde::Deserializer<'de> {
let wire = AttemptFailureSummaryWire::deserialize(deserializer)?;
Self::from_parts(wire.origin, wire.code, wire.status, wire.operation, wire.text, wire.facts)
.map_err(de::Error::custom)
}
}
impl AttemptFailureSummary {
fn from_parts(
origin: AttemptFailureOrigin,
code: AttemptFailureCode,
status: AttemptFailureStatus,
operation: AttemptFailureOperation,
text: Option<AttemptFailureText>,
facts: Vec<AttemptFailureFact>,
) -> std::result::Result<Self, &'static str> {
if facts.len() > MAX_ATTEMPT_FAILURE_FACTS {
return Err("attempt failure facts exceed their fixed bound");
}
if facts.windows(2).any(|pair| pair[0].key >= pair[1].key) {
return Err("attempt failure facts must use unique deterministic key order");
}
Ok(Self { origin, code, status, operation, text, facts })
}
pub fn from_native_error(error: &QueryError, operation: AttemptFailureOperation) -> Self {
let mut facts = Vec::with_capacity(2);
let reason = match error.fact_value("reason") {
Some("backend_closing") => Some(AttemptFailureFactValue::BackendClosing),
Some("deadline_expired") => Some(AttemptFailureFactValue::DeadlineExpired),
Some("caller_cancelled") => Some(AttemptFailureFactValue::CallerCancelled),
Some("admission_queue_full") => Some(AttemptFailureFactValue::AdmissionQueueFull),
Some("queue_wait_elapsed") => Some(AttemptFailureFactValue::QueueWaitElapsed),
Some("memory_grant_wait_timeout") => Some(AttemptFailureFactValue::MemoryGrantWaitTimeout),
Some("required_floor_exceeds_device_capacity") | Some("impossible_requirement") => {
Some(AttemptFailureFactValue::RequiredFloorExceedsDeviceCapacity)
},
Some("query_cap_exceeds_resident_capacity") => {
Some(AttemptFailureFactValue::QueryCapExceedsResidentCapacity)
},
Some("backend_unhealthy") => Some(AttemptFailureFactValue::BackendUnhealthy),
_ => None,
};
if let Some(value) = reason {
facts.push(AttemptFailureFact { key: AttemptFailureFactKey::Reason, value });
}
let allocation_source = match error.context().allocation_source {
Some(AllocationSource::RmmOutOfMemory) => Some(AttemptFailureFactValue::CugraphRmmOutOfMemory),
Some(AllocationSource::RmmBadAlloc) => Some(AttemptFailureFactValue::CugraphRmmBadAlloc),
Some(AllocationSource::StdBadAlloc) => Some(AttemptFailureFactValue::CugraphStdBadAlloc),
Some(AllocationSource::Unknown) | None => None,
};
if let Some(value) = allocation_source {
facts.push(AttemptFailureFact { key: AttemptFailureFactKey::AllocationSource, value });
}
Self {
origin: AttemptFailureOrigin::Native,
code: AttemptFailureCode::from_query_error_code(error.code()),
status: AttemptFailureStatus::from_query_error_status(error.status()),
operation,
text: None,
facts,
}
}
pub const fn datafusion_fallback() -> Self {
Self {
origin: AttemptFailureOrigin::Datafusion,
code: AttemptFailureCode::DatafusionError,
status: AttemptFailureStatus::Permanent,
operation: AttemptFailureOperation::Datafusion,
text: None,
facts: Vec::new(),
}
}
pub const fn backend_closing() -> Self {
Self {
origin: AttemptFailureOrigin::AttemptControl,
code: AttemptFailureCode::BackendClosing,
status: AttemptFailureStatus::Permanent,
operation: AttemptFailureOperation::Admission,
text: None,
facts: Vec::new(),
}
}
const fn control(code: AttemptFailureCode) -> Self {
Self {
origin: AttemptFailureOrigin::AttemptControl,
code,
status: AttemptFailureStatus::NotApplicable,
operation: AttemptFailureOperation::AttemptControl,
text: None,
facts: Vec::new(),
}
}
pub const fn origin(&self) -> AttemptFailureOrigin {
self.origin
}
pub const fn code(&self) -> AttemptFailureCode {
self.code
}
pub const fn status(&self) -> AttemptFailureStatus {
self.status
}
pub const fn operation(&self) -> AttemptFailureOperation {
self.operation
}
pub(crate) fn text(&self) -> Option<&AttemptFailureText> {
self.text.as_ref()
}
pub fn facts(&self) -> &[AttemptFailureFact] {
&self.facts
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CleanupFailurePhase {
SchedulingCycle,
RunnableTaskSelection,
TaskPreparation,
TaskExecution,
TaskPanic,
TaskCompletion,
CompletedResourceRelease,
FinalValidation,
QueryServiceStreamSynchronization,
QueryServiceCallbackAbandoned,
QueryServiceCleanup,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CleanupFailureSummary {
phase: CleanupFailurePhase,
failure: AttemptFailureSummary,
complete: bool,
}
impl CleanupFailureSummary {
pub fn from_native_error(phase: CleanupFailurePhase, error: &QueryError) -> Self {
Self {
phase,
failure: AttemptFailureSummary::from_native_error(error, AttemptFailureOperation::Cleanup),
complete: true,
}
}
pub(crate) const fn stream_synchronization_failed() -> Self {
Self {
phase: CleanupFailurePhase::QueryServiceStreamSynchronization,
failure: AttemptFailureSummary::control(AttemptFailureCode::StreamSynchronizationFailed),
complete: true,
}
}
pub(crate) const fn callback_abandoned() -> Self {
Self {
phase: CleanupFailurePhase::QueryServiceCallbackAbandoned,
failure: AttemptFailureSummary::control(AttemptFailureCode::CleanupCallbackAbandoned),
complete: true,
}
}
pub const fn phase(&self) -> CleanupFailurePhase {
self.phase
}
pub const fn failure(&self) -> &AttemptFailureSummary {
&self.failure
}
pub const fn complete(&self) -> bool {
self.complete
}
pub(crate) fn mark_incomplete(&mut self) {
self.complete = false;
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptFinalDomainUsage {
pub domain_id: u64,
pub(crate) active: bool,
pub(crate) active_members: u64,
pub(crate) ceiling_bytes: u64,
pub(crate) allocation_limit_bytes: u64,
pub current_bytes: u64,
pub peak_bytes: u64,
pub(crate) total_bytes: u64,
pub(crate) current_aligned_bytes: u64,
pub peak_aligned_bytes: u64,
pub(crate) total_allocations: u64,
pub(crate) current_allocations: u64,
pub(crate) peak_allocations: u64,
pub(crate) failed_allocation_count: u64,
pub(crate) largest_failed_allocation_bytes: u64,
}
impl From<cudf::QueryMemoryDomainUsage> for AttemptFinalDomainUsage {
fn from(usage: cudf::QueryMemoryDomainUsage) -> Self {
Self {
domain_id: usage.domain_id,
active: usage.active,
active_members: usage.active_members,
ceiling_bytes: usage.ceiling_bytes,
allocation_limit_bytes: usage.allocation_limit_bytes,
current_bytes: usage.current_bytes,
peak_bytes: usage.peak_bytes,
total_bytes: usage.total_bytes,
current_aligned_bytes: usage.current_aligned_bytes,
peak_aligned_bytes: usage.peak_aligned_bytes,
total_allocations: usage.total_allocations,
current_allocations: usage.current_allocations,
peak_allocations: usage.peak_allocations,
failed_allocation_count: usage.failed_allocation_count,
largest_failed_allocation_bytes: usage.largest_failed_allocation_bytes,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum AttemptFaultIsolation {
DeviceLocal,
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum AttemptTeardownHealth {
Completed,
StreamSynchronizationFailed,
CallbackAbandoned,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum AttemptDeviceHealthStatus {
Healthy,
Unhealthy,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum AttemptNativeFaultKind {
CurrentDeviceQueryFailed,
OwnerDeviceEntryFailed,
AmbientDeviceRestoreFailed,
FatalDeviceOperation,
TeardownQuarantined,
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum AttemptNativeStatus {
Cudf(u32),
Cuda(i32),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct AttemptNativeFault {
pub(crate) kind: AttemptNativeFaultKind,
pub(crate) status: Option<AttemptNativeStatus>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct AttemptDeviceHealth {
pub(crate) status: AttemptDeviceHealthStatus,
pub(crate) fault_count: u64,
pub(crate) rejected_operation_count: u64,
pub(crate) quarantined_teardown_count: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptNativeFaultEvidence {
pub(crate) isolation: AttemptFaultIsolation,
pub(crate) teardown_health: AttemptTeardownHealth,
pub(crate) device_health: Option<AttemptDeviceHealth>,
pub(crate) first_fault: Option<AttemptNativeFault>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptEvidenceProvenance {
AttemptExact,
DeviceSnapshot,
SampledAtTerminalBoundary,
Unavailable,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptEvidenceField<T> {
pub provenance: AttemptEvidenceProvenance,
pub value: Option<T>,
}
impl<T> AttemptEvidenceField<T> {
pub(crate) const fn attempt_exact(value: T) -> Self {
Self { provenance: AttemptEvidenceProvenance::AttemptExact, value: Some(value) }
}
pub(crate) const fn device_snapshot(value: T) -> Self {
Self { provenance: AttemptEvidenceProvenance::DeviceSnapshot, value: Some(value) }
}
pub(crate) const fn sampled_at_terminal_boundary(value: T) -> Self {
Self { provenance: AttemptEvidenceProvenance::SampledAtTerminalBoundary, value: Some(value) }
}
pub(crate) const fn unavailable() -> Self {
Self { provenance: AttemptEvidenceProvenance::Unavailable, value: None }
}
}
impl<T> Default for AttemptEvidenceField<T> {
fn default() -> Self {
Self::unavailable()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptTimedDomainUsage {
pub(crate) sampled_ns: u64,
pub(crate) usage: AttemptFinalDomainUsage,
pub pool_usage: Option<AttemptAllocatorPoolUsage>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptAllocatorEventSource {
Physical,
ReservationLimit,
AllocationLimit,
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum AttemptAllocatorEventAction {
Decline,
Retry,
TrimAndRetry,
CallbackPanicked,
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptAllocatorPoolUsage {
pub(crate) reserved_current_bytes: u64,
pub reserved_high_bytes: u64,
pub(crate) used_current_bytes: u64,
pub used_high_bytes: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptAllocatorEvent {
pub sequence: u64,
pub(crate) monotonic_time_ns: AttemptEvidenceField<u64>,
pub requested_bytes: u64,
pub source: AttemptAllocatorEventSource,
pub(crate) action: AttemptEvidenceField<AttemptAllocatorEventAction>,
pub(crate) callback_attempt: u32,
pub(crate) trim_retry_count: u32,
pub(crate) pool_usage: AttemptEvidenceField<AttemptAllocatorPoolUsage>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptUnattributedAllocatorEvent {
pub(crate) device_ordinal: usize,
pub(crate) sampled_ns: u64,
pub(crate) event: AttemptAllocatorEvent,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptResourceEvidence {
pub ledger_sequence: AttemptEvidenceField<u64>,
pub(crate) ledger_sample_ns: AttemptEvidenceField<u64>,
pub(crate) configured_limit_bytes: AttemptEvidenceField<u64>,
pub(crate) headroom_bytes: AttemptEvidenceField<u64>,
pub(crate) backend_reserve_bytes: AttemptEvidenceField<u64>,
pub(crate) managed_capacity_bytes: AttemptEvidenceField<u64>,
pub(crate) live_grants_bytes: AttemptEvidenceField<u64>,
pub(crate) resident_owner_bytes: AttemptEvidenceField<u64>,
pub cache_evictable_resident_bytes: AttemptEvidenceField<u64>,
pub cache_pinned_in_use_bytes: AttemptEvidenceField<u64>,
pub(crate) active_attempt_count: AttemptEvidenceField<u64>,
pub grant_id: AttemptEvidenceField<u64>,
pub(crate) grant_cap_bytes: AttemptEvidenceField<u64>,
pub domain_id: AttemptEvidenceField<u64>,
pub(crate) domain_cap_bytes: AttemptEvidenceField<u64>,
pub outcome_terminal_usage: AttemptEvidenceField<AttemptTimedDomainUsage>,
pub(crate) post_cleanup_usage: AttemptEvidenceField<AttemptTimedDomainUsage>,
pub allocator_event: AttemptEvidenceField<AttemptAllocatorEvent>,
pub unattributed_device_allocator_event: AttemptEvidenceField<AttemptUnattributedAllocatorEvent>,
}
impl AttemptResourceEvidence {
pub(crate) const fn unavailable() -> Self {
Self {
ledger_sequence: AttemptEvidenceField::unavailable(),
ledger_sample_ns: AttemptEvidenceField::unavailable(),
configured_limit_bytes: AttemptEvidenceField::unavailable(),
headroom_bytes: AttemptEvidenceField::unavailable(),
backend_reserve_bytes: AttemptEvidenceField::unavailable(),
managed_capacity_bytes: AttemptEvidenceField::unavailable(),
live_grants_bytes: AttemptEvidenceField::unavailable(),
resident_owner_bytes: AttemptEvidenceField::unavailable(),
cache_evictable_resident_bytes: AttemptEvidenceField::unavailable(),
cache_pinned_in_use_bytes: AttemptEvidenceField::unavailable(),
active_attempt_count: AttemptEvidenceField::unavailable(),
grant_id: AttemptEvidenceField::unavailable(),
grant_cap_bytes: AttemptEvidenceField::unavailable(),
domain_id: AttemptEvidenceField::unavailable(),
domain_cap_bytes: AttemptEvidenceField::unavailable(),
outcome_terminal_usage: AttemptEvidenceField::unavailable(),
post_cleanup_usage: AttemptEvidenceField::unavailable(),
allocator_event: AttemptEvidenceField::unavailable(),
unattributed_device_allocator_event: AttemptEvidenceField::unavailable(),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct AttemptTerminalCommit {
capacity_disposition: AttemptCapacityDisposition,
final_domain_usage: Option<AttemptFinalDomainUsage>,
health_transition: AttemptHealthTransition,
native_fault_evidence: Option<AttemptNativeFaultEvidence>,
resource_evidence: Option<AttemptResourceEvidence>,
alternate_device_eligibility: AttemptAlternateDeviceEligibility,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum AttemptAlternateDeviceEligibility {
NotApplicable,
Eligible,
NoEligibleAlternate,
Unavailable,
}
impl AttemptTerminalCommit {
pub(crate) const fn new(
capacity_disposition: AttemptCapacityDisposition,
final_domain_usage: Option<AttemptFinalDomainUsage>,
health_transition: AttemptHealthTransition,
) -> Self {
Self {
capacity_disposition,
final_domain_usage,
health_transition,
native_fault_evidence: None,
resource_evidence: None,
alternate_device_eligibility: AttemptAlternateDeviceEligibility::NotApplicable,
}
}
pub(crate) const fn with_native_fault_evidence(mut self, evidence: AttemptNativeFaultEvidence) -> Self {
self.native_fault_evidence = Some(evidence);
self
}
pub(crate) const fn with_resource_evidence(mut self, evidence: AttemptResourceEvidence) -> Self {
self.resource_evidence = Some(evidence);
self
}
pub(crate) const fn with_alternate_device_eligibility(
mut self,
eligibility: AttemptAlternateDeviceEligibility,
) -> Self {
self.alternate_device_eligibility = eligibility;
self
}
pub(crate) const fn capacity_disposition(self) -> AttemptCapacityDisposition {
self.capacity_disposition
}
pub(crate) const fn final_domain_usage(self) -> Option<AttemptFinalDomainUsage> {
self.final_domain_usage
}
pub(crate) const fn health_transition(self) -> AttemptHealthTransition {
self.health_transition
}
pub(crate) const fn native_fault_evidence(self) -> Option<AttemptNativeFaultEvidence> {
self.native_fault_evidence
}
pub(crate) const fn resource_evidence(self) -> Option<AttemptResourceEvidence> {
self.resource_evidence
}
pub(crate) const fn alternate_device_eligibility(self) -> AttemptAlternateDeviceEligibility {
self.alternate_device_eligibility
}
pub(crate) const fn has_consistent_capacity_health_transition(self) -> bool {
matches!(
(self.capacity_disposition, self.health_transition),
(AttemptCapacityDisposition::Released, AttemptHealthTransition::HealthyReuse)
| (AttemptCapacityDisposition::FrozenResident, AttemptHealthTransition::HealthyReuse)
| (AttemptCapacityDisposition::RetainedQuarantine, AttemptHealthTransition::DeviceQuarantined)
| (AttemptCapacityDisposition::RetainedQuarantine, AttemptHealthTransition::BackendUnhealthy)
)
}
pub(crate) const fn is_admitted_cleanup_commit(self) -> bool {
!matches!(self.capacity_disposition, AttemptCapacityDisposition::NotAdmitted)
&& self.has_consistent_capacity_health_transition()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptRetryRecovery {
NoRetryNeeded,
DoNotRetry,
Backoff,
ChangeConfiguration,
RetryAnotherEligibleDevice,
RecoverBackend,
Indeterminate,
}
impl AttemptRetryRecovery {
pub const fn as_str(self) -> &'static str {
match self {
Self::NoRetryNeeded => "no_retry_needed",
Self::DoNotRetry => "do_not_retry",
Self::Backoff => "backoff",
Self::ChangeConfiguration => "change_configuration",
Self::RetryAnotherEligibleDevice => "retry_another_eligible_device",
Self::RecoverBackend => "recover_backend",
Self::Indeterminate => "indeterminate",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AttemptRetryPrerequisite {
None,
CapacityAvailable,
ConfigurationChanged,
QuarantinedDeviceAndEligibleAlternate,
BackendRecovered,
EvidenceUnavailable,
}
impl AttemptRetryPrerequisite {
pub const fn as_str(self) -> &'static str {
match self {
Self::None => "none",
Self::CapacityAvailable => "capacity_available",
Self::ConfigurationChanged => "configuration_changed",
Self::QuarantinedDeviceAndEligibleAlternate => "quarantined_device_and_eligible_alternate",
Self::BackendRecovered => "backend_recovered",
Self::EvidenceUnavailable => "evidence_unavailable",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct AttemptRetryAdvice {
pub recovery: AttemptRetryRecovery,
pub prerequisite: AttemptRetryPrerequisite,
}
#[derive(Deserialize)]
struct AttemptRetryAdviceWire {
recovery: AttemptRetryRecovery,
prerequisite: AttemptRetryPrerequisite,
}
impl<'de> Deserialize<'de> for AttemptRetryAdvice {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where D: serde::Deserializer<'de> {
let wire = AttemptRetryAdviceWire::deserialize(deserializer)?;
Self::from_parts(wire.recovery, wire.prerequisite).map_err(de::Error::custom)
}
}
impl AttemptRetryAdvice {
fn from_parts(
recovery: AttemptRetryRecovery,
prerequisite: AttemptRetryPrerequisite,
) -> std::result::Result<Self, &'static str> {
let valid = matches!(
(recovery, prerequisite),
(AttemptRetryRecovery::NoRetryNeeded, AttemptRetryPrerequisite::None)
| (AttemptRetryRecovery::DoNotRetry, AttemptRetryPrerequisite::None)
| (AttemptRetryRecovery::Backoff, AttemptRetryPrerequisite::CapacityAvailable)
| (AttemptRetryRecovery::ChangeConfiguration, AttemptRetryPrerequisite::ConfigurationChanged)
| (
AttemptRetryRecovery::RetryAnotherEligibleDevice,
AttemptRetryPrerequisite::QuarantinedDeviceAndEligibleAlternate
)
| (AttemptRetryRecovery::RecoverBackend, AttemptRetryPrerequisite::BackendRecovered)
| (AttemptRetryRecovery::Indeterminate, AttemptRetryPrerequisite::EvidenceUnavailable)
);
valid.then_some(Self { recovery, prerequisite }).ok_or("retry recovery and prerequisite are inconsistent")
}
pub const fn no_retry_needed() -> Self {
Self { recovery: AttemptRetryRecovery::NoRetryNeeded, prerequisite: AttemptRetryPrerequisite::None }
}
pub const fn do_not_retry() -> Self {
Self { recovery: AttemptRetryRecovery::DoNotRetry, prerequisite: AttemptRetryPrerequisite::None }
}
pub const fn indeterminate() -> Self {
Self {
recovery: AttemptRetryRecovery::Indeterminate,
prerequisite: AttemptRetryPrerequisite::EvidenceUnavailable,
}
}
fn backoff() -> Self {
Self { recovery: AttemptRetryRecovery::Backoff, prerequisite: AttemptRetryPrerequisite::CapacityAvailable }
}
fn change_configuration() -> Self {
Self {
recovery: AttemptRetryRecovery::ChangeConfiguration,
prerequisite: AttemptRetryPrerequisite::ConfigurationChanged,
}
}
fn retry_another_eligible_device() -> Self {
Self {
recovery: AttemptRetryRecovery::RetryAnotherEligibleDevice,
prerequisite: AttemptRetryPrerequisite::QuarantinedDeviceAndEligibleAlternate,
}
}
fn recover_backend() -> Self {
Self {
recovery: AttemptRetryRecovery::RecoverBackend,
prerequisite: AttemptRetryPrerequisite::BackendRecovered,
}
}
pub(crate) fn from_sealed_attempt(
outcome: &RuntimeTerminalOutcome,
primary_failure: Option<&AttemptFailureSummary>,
capacity_disposition: AttemptCapacityDisposition,
health_transition: Option<AttemptHealthTransition>,
native_fault_evidence: Option<AttemptNativeFaultEvidence>,
resource_evidence: Option<AttemptResourceEvidence>,
alternate_device_eligibility: AttemptAlternateDeviceEligibility,
) -> Self {
if requires_backend_recovery(primary_failure, health_transition, native_fault_evidence) {
return Self::recover_backend();
}
match outcome {
RuntimeTerminalOutcome::Succeeded => Self::no_retry_needed(),
RuntimeTerminalOutcome::Cancelled | RuntimeTerminalOutcome::Dropped | RuntimeTerminalOutcome::Panicked => {
Self::do_not_retry()
},
RuntimeTerminalOutcome::Failed { .. } => {
if has_proven_device_local_fault(capacity_disposition, health_transition, native_fault_evidence) {
return match alternate_device_eligibility {
AttemptAlternateDeviceEligibility::Eligible => Self::retry_another_eligible_device(),
AttemptAlternateDeviceEligibility::NoEligibleAlternate => Self::do_not_retry(),
AttemptAlternateDeviceEligibility::NotApplicable
| AttemptAlternateDeviceEligibility::Unavailable => Self::indeterminate(),
};
}
let Some(failure) = primary_failure else {
return Self::indeterminate();
};
if is_configuration_failure(failure) {
return Self::change_configuration();
}
if has_reason(failure, AttemptFailureFactValue::RequiredFloorExceedsDeviceCapacity)
|| has_reason(failure, AttemptFailureFactValue::QueryCapExceedsResidentCapacity)
{
return Self::change_configuration();
}
if is_queue_overload(failure) {
return Self::backoff();
}
if matches!(failure.code(), AttemptFailureCode::Query(QueryErrorCode::MemoryReservationFailed)) {
return match exact_allocator_event_source(resource_evidence) {
Some(
AttemptAllocatorEventSource::ReservationLimit
| AttemptAllocatorEventSource::AllocationLimit,
) => Self::change_configuration(),
Some(AttemptAllocatorEventSource::Physical) => Self::indeterminate(),
Some(AttemptAllocatorEventSource::Unknown) | None => Self::indeterminate(),
};
}
Self::indeterminate()
},
}
}
}
fn has_reason(failure: &AttemptFailureSummary, expected: AttemptFailureFactValue) -> bool {
failure.facts().iter().any(|fact| fact.key() == AttemptFailureFactKey::Reason && fact.value() == expected)
}
fn is_configuration_failure(failure: &AttemptFailureSummary) -> bool {
matches!(
failure.code(),
AttemptFailureCode::Query(QueryErrorCode::UnsupportedPipelineContract)
| AttemptFailureCode::Query(QueryErrorCode::UnsupportedOperator)
| AttemptFailureCode::Query(QueryErrorCode::UnsupportedExpression)
| AttemptFailureCode::Query(QueryErrorCode::UnsupportedType)
| AttemptFailureCode::Query(QueryErrorCode::CudfUnsupported)
| AttemptFailureCode::Query(QueryErrorCode::QueryMemoryLimitExceeded)
)
}
fn is_queue_overload(failure: &AttemptFailureSummary) -> bool {
failure.code() == AttemptFailureCode::Query(QueryErrorCode::ServiceOverloaded)
&& failure.facts().iter().any(|fact| {
fact.key() == AttemptFailureFactKey::Reason
&& matches!(
fact.value(),
AttemptFailureFactValue::AdmissionQueueFull
| AttemptFailureFactValue::QueueWaitElapsed
| AttemptFailureFactValue::MemoryGrantWaitTimeout
)
})
}
fn requires_backend_recovery(
primary_failure: Option<&AttemptFailureSummary>,
health_transition: Option<AttemptHealthTransition>,
native_fault_evidence: Option<AttemptNativeFaultEvidence>,
) -> bool {
matches!(health_transition, Some(AttemptHealthTransition::BackendUnhealthy))
|| native_fault_evidence.is_some_and(|evidence| evidence.isolation == AttemptFaultIsolation::Unknown)
|| primary_failure.is_some_and(|failure| has_reason(failure, AttemptFailureFactValue::BackendUnhealthy))
}
fn has_proven_device_local_fault(
capacity_disposition: AttemptCapacityDisposition,
health_transition: Option<AttemptHealthTransition>,
native_fault_evidence: Option<AttemptNativeFaultEvidence>,
) -> bool {
matches!(capacity_disposition, AttemptCapacityDisposition::RetainedQuarantine)
&& matches!(health_transition, Some(AttemptHealthTransition::DeviceQuarantined))
&& native_fault_evidence.as_ref().is_some_and(has_proven_device_local_fault_evidence)
}
pub(crate) fn has_proven_device_local_fault_evidence(evidence: &AttemptNativeFaultEvidence) -> bool {
if evidence.isolation != AttemptFaultIsolation::DeviceLocal {
return false;
}
matches!(evidence.teardown_health, AttemptTeardownHealth::StreamSynchronizationFailed)
|| (evidence.device_health.is_some_and(|health| health.status == AttemptDeviceHealthStatus::Unhealthy)
&& evidence.first_fault.is_some())
}
fn exact_allocator_event_source(
resource_evidence: Option<AttemptResourceEvidence>,
) -> Option<AttemptAllocatorEventSource> {
let allocator_event = resource_evidence?.allocator_event;
(allocator_event.provenance == AttemptEvidenceProvenance::AttemptExact)
.then_some(allocator_event.value?)
.map(|event| event.source)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptAdmissionFact {
pub admission_sequence: u64,
pub admission_wait_ns: u64,
pub device_ordinal: usize,
pub grant_bytes: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptAdmissionWaitFact {
pub submission_sequence: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AttemptTimestamps {
pub submitted_ns: u64,
pub planning_started_ns: Option<u64>,
pub planning_finished_ns: Option<u64>,
pub admission_requested_ns: Option<u64>,
pub admitted_ns: Option<u64>,
pub physical_execution_started_ns: Option<u64>,
pub gpu_execution_started_ns: Option<u64>,
pub outcome_terminal_ns: u64,
pub cleanup_complete_ns: Option<u64>,
pub report_ready_ns: Option<u64>,
}
fn deserialize_query_attempt_report_schema<'de, D>(deserializer: D) -> std::result::Result<String, D::Error>
where D: serde::Deserializer<'de> {
let schema = String::deserialize(deserializer)?;
if schema != QUERY_ATTEMPT_REPORT_SCHEMA {
return Err(de::Error::custom(format!(
"unsupported attempt report schema {schema:?}; expected {QUERY_ATTEMPT_REPORT_SCHEMA:?}"
)));
}
Ok(schema)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct QueryAttemptReport {
#[serde(deserialize_with = "deserialize_query_attempt_report_schema")]
pub schema: String,
pub attempt_id: AttemptId,
pub query_id: Option<AttemptQueryId>,
pub trace_context: Option<AttemptTraceContext>,
pub disposition: AttemptDisposition,
pub outcome: RuntimeTerminalOutcome,
pub primary_failure: Option<AttemptFailureSummary>,
pub retry_advice: AttemptRetryAdvice,
pub device_ordinal: Option<usize>,
pub admission: Option<AttemptAdmissionFact>,
pub admission_wait: Option<AttemptAdmissionWaitFact>,
#[serde(default)]
pub composed_schedule: Option<AttemptComposedScheduleProjection>,
pub timestamps: AttemptTimestamps,
pub cleanup_disposition: AttemptCleanupDisposition,
pub cleanup_failure_count: u64,
pub cleanup_failure_count_complete: bool,
pub first_cleanup_failure: Option<CleanupFailureSummary>,
pub capacity_disposition: AttemptCapacityDisposition,
pub final_domain_usage: Option<AttemptFinalDomainUsage>,
pub health_transition: Option<AttemptHealthTransition>,
pub native_fault_evidence: Option<AttemptNativeFaultEvidence>,
pub resource_evidence: Option<AttemptResourceEvidence>,
pub observation_lost_records: u64,
pub observation_retained_records: u64,
pub observation_retained_bytes: u64,
pub observation_truncated: bool,
pub observation_degraded: bool,
pub unix_epoch_ns: Option<u64>,
}
impl QueryAttemptReport {
pub fn sealed_projection(&self) -> SealedAttemptProjection {
SealedAttemptProjection::from(self)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct SealedAttemptProjection {
pub schema: &'static str,
pub attempt_id: AttemptId,
pub disposition: AttemptDisposition,
pub outcome: RuntimeTerminalOutcome,
pub cleanup_disposition: AttemptCleanupDisposition,
pub cleanup_failure_count: u64,
pub cleanup_failure_count_complete: bool,
pub capacity_disposition: AttemptCapacityDisposition,
pub health_transition: Option<AttemptHealthTransition>,
pub retry_advice: AttemptRetryAdvice,
pub observation_truncated: bool,
pub observation_degraded: bool,
pub native_fault_evidence_present: bool,
pub resource_evidence_present: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SealedAttemptLabels {
pub disposition: &'static str,
pub outcome: &'static str,
pub cleanup_disposition: &'static str,
pub cleanup_failure_count_complete: &'static str,
pub capacity_disposition: &'static str,
pub health_transition: &'static str,
pub retry_recovery: &'static str,
pub retry_prerequisite: &'static str,
pub observation_truncated: &'static str,
pub observation_degraded: &'static str,
pub native_fault_evidence: &'static str,
pub resource_evidence: &'static str,
}
impl SealedAttemptProjection {
pub const fn labels(&self) -> SealedAttemptLabels {
SealedAttemptLabels {
disposition: match self.disposition {
AttemptDisposition::Unplanned => "unplanned",
AttemptDisposition::CpuFallback => "cpu_fallback",
AttemptDisposition::PartialNative => "partial_native",
AttemptDisposition::Native => "native",
},
outcome: self.outcome.stable_label(),
cleanup_disposition: match self.cleanup_disposition {
AttemptCleanupDisposition::NotRequired => "not_required",
AttemptCleanupDisposition::Completed => "completed",
AttemptCleanupDisposition::CompletedWithFailures => "completed_with_failures",
},
cleanup_failure_count_complete: if self.cleanup_failure_count_complete { "true" } else { "false" },
capacity_disposition: match self.capacity_disposition {
AttemptCapacityDisposition::NotAdmitted => "not_admitted",
AttemptCapacityDisposition::Released => "released",
AttemptCapacityDisposition::FrozenResident => "frozen_resident",
AttemptCapacityDisposition::RetainedQuarantine => "retained_quarantine",
},
health_transition: match self.health_transition {
None => "none",
Some(AttemptHealthTransition::HealthyReuse) => "healthy_reuse",
Some(AttemptHealthTransition::DeviceQuarantined) => "device_quarantined",
Some(AttemptHealthTransition::BackendUnhealthy) => "backend_unhealthy",
},
retry_recovery: self.retry_advice.recovery.as_str(),
retry_prerequisite: self.retry_advice.prerequisite.as_str(),
observation_truncated: if self.observation_truncated { "true" } else { "false" },
observation_degraded: if self.observation_degraded { "true" } else { "false" },
native_fault_evidence: if self.native_fault_evidence_present { "present" } else { "absent" },
resource_evidence: if self.resource_evidence_present { "present" } else { "absent" },
}
}
}
impl From<&QueryAttemptReport> for SealedAttemptProjection {
fn from(report: &QueryAttemptReport) -> Self {
Self {
schema: SEALED_ATTEMPT_PROJECTION_SCHEMA,
attempt_id: report.attempt_id,
disposition: report.disposition,
outcome: report.outcome.clone(),
cleanup_disposition: report.cleanup_disposition,
cleanup_failure_count: report.cleanup_failure_count,
cleanup_failure_count_complete: report.cleanup_failure_count_complete,
capacity_disposition: report.capacity_disposition,
health_transition: report.health_transition,
retry_advice: report.retry_advice,
observation_truncated: report.observation_truncated,
observation_degraded: report.observation_degraded,
native_fault_evidence_present: report.native_fault_evidence.is_some(),
resource_evidence_present: report.resource_evidence.is_some(),
}
}
}
impl QueryAttemptReport {
pub fn total_ns(&self) -> Option<u64> {
self.timestamps.report_ready_ns.map(|ready| ready.saturating_sub(self.timestamps.submitted_ns))
}
pub fn planning_ns(&self) -> Option<u64> {
self.timestamps
.planning_started_ns
.zip(self.timestamps.planning_finished_ns)
.map(|(start, end)| end.saturating_sub(start))
}
pub fn queue_wait_ns(&self) -> Option<u64> {
self.timestamps
.admission_requested_ns
.zip(self.timestamps.admitted_ns)
.map(|(start, end)| end.saturating_sub(start))
}
pub fn post_planning_pre_admission_ns(&self) -> Option<u64> {
self.timestamps
.planning_finished_ns
.zip(self.timestamps.admission_requested_ns)
.map(|(start, end)| end.saturating_sub(start))
}
pub fn admitted_to_physical_execution_ns(&self) -> Option<u64> {
self.timestamps
.admitted_ns
.zip(self.timestamps.physical_execution_started_ns)
.map(|(start, end)| end.saturating_sub(start))
}
pub fn physical_to_gpu_dispatch_ns(&self) -> Option<u64> {
self.timestamps
.physical_execution_started_ns
.zip(self.timestamps.gpu_execution_started_ns)
.map(|(start, end)| end.saturating_sub(start))
}
pub fn admitted_to_gpu_dispatch_ns(&self) -> Option<u64> {
self.timestamps
.admitted_ns
.zip(self.timestamps.gpu_execution_started_ns)
.map(|(start, end)| end.saturating_sub(start))
}
pub fn submitted_to_gpu_dispatch_ns(&self) -> Option<u64> {
self.timestamps.gpu_execution_started_ns.map(|end| end.saturating_sub(self.timestamps.submitted_ns))
}
pub fn gpu_dispatch_to_terminal_ns(&self) -> Option<u64> {
self.timestamps.gpu_execution_started_ns.map(|start| self.timestamps.outcome_terminal_ns.saturating_sub(start))
}
pub fn physical_execution_ns(&self) -> Option<u64> {
self.timestamps
.physical_execution_started_ns
.map(|start| self.timestamps.outcome_terminal_ns.saturating_sub(start))
}
pub fn cleanup_ns(&self) -> Option<u64> {
self.timestamps.cleanup_complete_ns.map(|end| end.saturating_sub(self.timestamps.outcome_terminal_ns))
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum RuntimeTerminalOutcome {
Succeeded,
Failed {
error_code: AttemptFailureCode,
error_status: AttemptFailureStatus,
},
Cancelled,
Dropped,
Panicked,
}
impl RuntimeTerminalOutcome {
pub const fn stable_label(&self) -> &'static str {
match self {
Self::Succeeded => "succeeded",
Self::Failed { .. } => "failed",
Self::Cancelled => "cancelled",
Self::Dropped => "dropped",
Self::Panicked => "panicked",
}
}
pub const fn failed(summary: &AttemptFailureSummary) -> Self {
Self::Failed { error_code: summary.code(), error_status: summary.status() }
}
pub fn from_query_error(error: &QueryError) -> Self {
if error.code() == QueryErrorCode::Cancelled {
Self::Cancelled
} else {
Self::Failed {
error_code: AttemptFailureCode::from_query_error_code(error.code()),
error_status: AttemptFailureStatus::from_query_error_status(error.status()),
}
}
}
pub(crate) fn primary_failure_from_query_error(
error: &QueryError,
operation: AttemptFailureOperation,
) -> Option<AttemptFailureSummary> {
match error.code() {
QueryErrorCode::Cancelled if error.fact_value("reason") != Some("deadline_expired") => None,
_ => Some(AttemptFailureSummary::from_native_error(error, operation)),
}
}
pub(crate) const fn from_primary_failure(summary: &AttemptFailureSummary) -> Self {
match summary.code() {
AttemptFailureCode::Query(QueryErrorCode::Cancelled) => Self::Cancelled,
_ => Self::failed(summary),
}
}
pub(crate) fn fallback_failure_summary(&self) -> Option<AttemptFailureSummary> {
match self {
Self::Succeeded => None,
Self::Failed { error_code, error_status } => Some(AttemptFailureSummary {
origin: AttemptFailureOrigin::Native,
code: *error_code,
status: *error_status,
operation: AttemptFailureOperation::Execution,
text: None,
facts: Vec::new(),
}),
Self::Cancelled => {
Some(AttemptFailureSummary::control(AttemptFailureCode::Query(QueryErrorCode::Cancelled)))
},
Self::Dropped => Some(AttemptFailureSummary::control(AttemptFailureCode::Dropped)),
Self::Panicked => Some(AttemptFailureSummary::control(AttemptFailureCode::Panicked)),
}
}
}
#[cfg(test)]
mod tests;