1use std::fmt;
4use std::time::Duration;
5
6use thiserror::Error;
7use uuid::Uuid;
8
9pub type RuntimeResult<T> = Result<T, RuntimeError>;
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
14#[serde(rename_all = "snake_case")]
15pub enum DomainDisposition {
16 Committed,
18 NotCommitted,
20 Unknown,
22}
23
24impl DomainDisposition {
25 pub const fn as_str(self) -> &'static str {
27 match self {
28 Self::Committed => "committed",
29 Self::NotCommitted => "not_committed",
30 Self::Unknown => "unknown",
31 }
32 }
33}
34
35#[derive(Debug, Error)]
37#[error("{source}")]
38pub struct DispatchError {
39 #[source]
40 source: RuntimeError,
41 disposition: DomainDisposition,
42}
43
44impl DispatchError {
45 pub(crate) fn new(source: RuntimeError, disposition: DomainDisposition) -> Self {
46 Self {
47 source,
48 disposition,
49 }
50 }
51
52 pub(crate) fn before_dispatch(source: RuntimeError) -> Self {
53 Self::new(source, DomainDisposition::NotCommitted)
54 }
55
56 pub(crate) fn after_handler(source: RuntimeError, domain_succeeded: bool) -> Self {
57 Self::new(
58 source,
59 if domain_succeeded {
60 DomainDisposition::Committed
61 } else {
62 DomainDisposition::Unknown
63 },
64 )
65 }
66
67 pub fn source(&self) -> &RuntimeError {
69 &self.source
70 }
71
72 pub fn disposition(&self) -> DomainDisposition {
74 self.disposition
75 }
76
77 pub fn into_parts(self) -> (RuntimeError, DomainDisposition) {
79 (self.source, self.disposition)
80 }
81
82 pub fn into_source(self) -> RuntimeError {
84 self.source
85 }
86}
87
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
90pub enum AuditObligationReason {
91 Terminal(crate::audit_batch::AuditTerminalReason),
93 GitDigestReceiptFailure,
95}
96
97#[derive(Debug, Error)]
99#[error("{message}")]
100pub struct AuditObligationFailure {
101 pub reason: AuditObligationReason,
103 pub verb: String,
105 pub message: String,
107 #[source]
108 source: Option<khive_storage::StorageError>,
109}
110
111impl AuditObligationFailure {
112 pub fn new(verb: impl Into<String>, reason: crate::audit_batch::AuditTerminalReason) -> Self {
114 let verb = verb.into();
115 Self {
116 message: format!("audit obligation commit failed for verb {verb:?}: {reason:?}"),
117 verb,
118 reason: AuditObligationReason::Terminal(reason),
119 source: None,
120 }
121 }
122
123 pub(crate) fn from_store(verb: &str, source: khive_storage::StorageError) -> Self {
124 let mut failure = Self::new(verb, crate::audit_batch::AuditTerminalReason::StoreFailure);
125 failure.message = format!("audit obligation commit failed for verb {verb:?}: {source}");
126 failure.source = Some(source);
127 failure
128 }
129
130 pub(crate) fn git_digest_receipt(branch: &'static str) -> Self {
131 Self {
132 reason: AuditObligationReason::GitDigestReceiptFailure,
133 verb: "git.digest".into(),
134 message: format!(
135 "git_digest_receipt_persist_failed: {branch}; git.digest writes may have committed, but no durable success receipt was confirmed; inspect ingest state before retrying"
136 ),
137 source: None,
138 }
139 }
140
141 pub const fn wire_code(&self) -> &'static str {
143 use crate::audit_batch::AuditTerminalReason;
144 match self.reason {
145 AuditObligationReason::Terminal(reason) => match reason {
146 AuditTerminalReason::PreflightRejected => "preflight_rejected",
147 AuditTerminalReason::AdmissionClosed => "admission_closed",
148 AuditTerminalReason::QueueAdmissionExhausted => "queue_admission_exhausted",
149 AuditTerminalReason::AdmissionDeadlineExpired => "admission_deadline_expired",
150 AuditTerminalReason::ResolutionDeadlineExpired => "resolution_deadline_expired",
151 AuditTerminalReason::IdentityConflict => "identity_conflict",
152 AuditTerminalReason::StoreFailure => "store_failure",
153 AuditTerminalReason::RetryExhausted => "retry_exhausted",
154 AuditTerminalReason::IdempotencyUnsupported => "idempotency_unsupported",
155 AuditTerminalReason::DriverPanicked => "driver_panicked",
156 AuditTerminalReason::DriverCancelled => "driver_cancelled",
157 AuditTerminalReason::DriverJoinLost => "driver_join_lost",
158 AuditTerminalReason::DriverExitedInconsistent => "driver_exited_inconsistent",
159 AuditTerminalReason::DriverAppendAbandoned => "driver_append_abandoned",
160 AuditTerminalReason::StoreWedged => "store_wedged",
161 },
162 AuditObligationReason::GitDigestReceiptFailure => "git_digest_receipt_failure",
163 }
164 }
165}
166
167pub const WRITER_POOL_CHECKOUT_TIMEOUT_STAGE: &str = "writer_pool_checkout_timeout";
170
171pub use khive_db::{SQLITE_WAL_CAPACITY_REFUSED_STAGE, SQLITE_WAL_CAPACITY_UNAVAILABLE_STAGE};
173
174pub const WRITER_QUEUE_SATURATED_STAGE: &str = "writer_queue_saturated";
177
178pub const WRITER_TASK_BEGIN_BUSY_STAGE: &str = "writer_task_begin_busy";
181
182pub const WRITER_TASK_REQUEST_FAILED_STAGE: &str = "writer_task_request_failed";
185
186pub const WRITER_TASK_TERMINATED_STAGE: &str = "writer_task_terminated";
189
190pub const STORAGE_ADMISSION_TIMEOUT_STAGE: &str = "storage_admission_timeout";
195
196pub const READ_TX_AGE_EVICTED_STAGE: &str = "read_tx_age_evicted";
207
208pub const WRITER_ADMISSION_SCOPE: &str = "writer_admission";
215
216#[derive(Debug, Clone, PartialEq, Eq)]
223pub struct AdmissionFailureContext {
224 pub stage: &'static str,
228 pub timeout: Duration,
230 pub capability: Option<khive_storage::StorageCapability>,
234 pub operation: Option<String>,
236 pub pool_identity: Option<String>,
238 pub scope: Option<&'static str>,
243 pub retry_after_ms: Option<u64>,
249}
250
251#[derive(Debug, Clone, PartialEq, Eq)]
258pub struct RetryableFailureContext {
259 pub stage: &'static str,
261 pub timeout: Duration,
263 pub capability: Option<khive_storage::StorageCapability>,
265 pub operation: Option<String>,
267 pub pool_identity: Option<String>,
269 pub scope: Option<&'static str>,
271 pub retry_after_ms: Option<u64>,
273}
274
275#[derive(Debug, Clone, Copy, PartialEq, Eq)]
282pub struct WriterTaskFailureContext {
283 pub stage: &'static str,
285 pub request_state: khive_storage::WriterTaskRequestState,
287 pub task_terminated: bool,
289 pub retryable: bool,
291}
292
293impl From<AdmissionFailureContext> for RetryableFailureContext {
294 fn from(context: AdmissionFailureContext) -> Self {
295 Self {
296 stage: context.stage,
297 timeout: context.timeout,
298 capability: context.capability,
299 operation: context.operation,
300 pool_identity: context.pool_identity,
301 scope: context.scope,
302 retry_after_ms: context.retry_after_ms,
303 }
304 }
305}
306
307#[derive(Debug, Clone, Copy, PartialEq, Eq)]
313pub enum ChannelIngestFailureClass {
314 Retryable { reason: &'static str },
316 Permanent { reason: &'static str },
318 Unknown { reason: &'static str },
320}
321
322impl ChannelIngestFailureClass {
323 pub const fn name(self) -> &'static str {
325 match self {
326 Self::Retryable { .. } => "retryable",
327 Self::Permanent { .. } => "permanent",
328 Self::Unknown { .. } => "unknown",
329 }
330 }
331
332 pub const fn reason(self) -> &'static str {
334 match self {
335 Self::Retryable { reason } | Self::Permanent { reason } | Self::Unknown { reason } => {
336 reason
337 }
338 }
339 }
340}
341
342#[derive(Debug, Clone, PartialEq, Eq)]
345pub struct WriterPoolCheckoutTimeoutContext {
346 pub timeout: Duration,
349 pub capability: Option<khive_storage::StorageCapability>,
351 pub operation: Option<String>,
353}
354
355#[derive(Debug, Clone, PartialEq, Eq)]
360pub struct GuardedWriteFailure {
361 pub entry_index: Option<usize>,
364 pub missing_source: Option<Uuid>,
367 pub missing_target: Option<Uuid>,
370}
371
372impl fmt::Display for GuardedWriteFailure {
373 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
374 let mut missing = Vec::new();
375 if let Some(source) = self.missing_source {
376 missing.push(format!("source {source}"));
377 }
378 if let Some(target) = self.missing_target {
379 missing.push(format!("target {target}"));
380 }
381 let missing = if missing.is_empty() {
382 "endpoint(s)".to_string()
383 } else {
384 missing.join(" and ")
385 };
386 match self.entry_index {
387 Some(index) => write!(
388 f,
389 "batch entry {index}: {missing} no longer exist at write time"
390 ),
391 None => write!(f, "{missing} no longer exist at write time"),
392 }
393 }
394}
395
396impl std::error::Error for GuardedWriteFailure {}
397
398#[derive(Debug, Clone, PartialEq, Eq)]
400pub struct MissingPackDependency {
401 pub from: String,
402 pub requires: String,
403}
404
405impl fmt::Display for MissingPackDependency {
406 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
407 write!(
408 f,
409 "pack '{}' requires '{}', but '{}' is not in the loaded pack set",
410 self.from, self.requires, self.requires
411 )
412 }
413}
414
415impl std::error::Error for MissingPackDependency {}
416
417#[derive(Debug, Clone, PartialEq, Eq)]
419pub struct MissingPackDependencies {
420 pub missing: Vec<MissingPackDependency>,
421}
422
423impl fmt::Display for MissingPackDependencies {
424 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
425 let parts: Vec<String> = self.missing.iter().map(ToString::to_string).collect();
426 write!(f, "{}", parts.join("; "))
427 }
428}
429
430impl std::error::Error for MissingPackDependencies {}
431
432#[derive(Debug, Clone, PartialEq, Eq)]
434pub struct CircularPackDependency {
435 pub cycle: Vec<String>,
436}
437
438impl fmt::Display for CircularPackDependency {
439 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
440 write!(
441 f,
442 "circular dependency detected among packs: {}",
443 self.cycle.join(" -> ")
444 )
445 }
446}
447
448impl std::error::Error for CircularPackDependency {}
449
450#[derive(Debug, Clone, Copy, PartialEq, Eq)]
457pub enum DenialAuditOutcome {
458 Committed,
461 NotCommitted(&'static str),
464 NoStore,
466 NotAudited,
469}
470
471impl DenialAuditOutcome {
472 pub fn wire_code(&self) -> String {
475 match self {
476 Self::Committed => "committed".to_string(),
477 Self::NotCommitted(code) => format!("not_committed:{code}"),
478 Self::NoStore => "no_store".to_string(),
479 Self::NotAudited => "not_audited".to_string(),
480 }
481 }
482}
483
484#[derive(Debug, Clone, PartialEq, Eq)]
488pub struct DenialReceipt {
489 pub audit_event_id: Option<uuid::Uuid>,
491 pub audit_outcome: DenialAuditOutcome,
493}
494
495impl DenialReceipt {
496 pub fn not_audited() -> Self {
498 Self {
499 audit_event_id: None,
500 audit_outcome: DenialAuditOutcome::NotAudited,
501 }
502 }
503
504 pub fn no_store() -> Self {
506 Self {
507 audit_event_id: None,
508 audit_outcome: DenialAuditOutcome::NoStore,
509 }
510 }
511}
512
513#[derive(Debug, Clone)]
515pub struct ReceiptRefusal {
516 pub code: &'static str,
518 pub message: String,
520 pub receipt_id: String,
522 pub reason: String,
524 pub detail: serde_json::Value,
526}
527
528#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
530#[serde(rename_all = "snake_case")]
531pub enum RefusalRecordingErrorClass {
532 EventStoreUnavailable,
533 EventAppendFailed,
534}
535
536impl RefusalRecordingErrorClass {
537 pub const fn as_str(self) -> &'static str {
538 match self {
539 Self::EventStoreUnavailable => "event_store_unavailable",
540 Self::EventAppendFailed => "event_append_failed",
541 }
542 }
543}
544
545#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
548#[serde(untagged)]
549pub enum RefusalEventRecording {
550 Recorded {
551 item_index: usize,
552 subject: Uuid,
553 event_id: Uuid,
554 },
555 Failed {
556 item_index: usize,
557 subject: Uuid,
558 error_class: RefusalRecordingErrorClass,
559 },
560}
561
562#[derive(Debug)]
568pub struct RefusalEventContext {
569 pub source: Box<RuntimeError>,
570 pub recordings: Vec<RefusalEventRecording>,
571}
572
573impl std::fmt::Display for RefusalEventContext {
574 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
575 std::fmt::Display::fmt(self.source.as_ref(), formatter)
576 }
577}
578
579impl std::error::Error for RefusalEventContext {
580 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
581 Some(self.source.as_ref())
582 }
583}
584
585#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
587pub struct ResolutionFacts {
588 pub project_id: Uuid,
589 pub duplicate_anchor_ids: Vec<Uuid>,
590 pub slug_backfilled: bool,
591 pub project_created: bool,
592 pub orphaned_project_id: Option<Uuid>,
593 pub orphaned_note_count: u64,
594}
595
596#[derive(Debug)]
598pub struct ResolutionFailureContext {
599 pub source: Box<RuntimeError>,
600 pub resolution: ResolutionFacts,
601}
602
603impl std::fmt::Display for ResolutionFailureContext {
604 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
605 std::fmt::Display::fmt(self.source.as_ref(), formatter)
606 }
607}
608
609impl std::error::Error for ResolutionFailureContext {
610 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
611 Some(self.source.as_ref())
612 }
613}
614
615#[derive(Debug, Error)]
619pub enum RuntimeError {
620 #[error("{source}")]
622 LintFix {
623 #[source]
624 source: Box<RuntimeError>,
625 detail: Box<crate::graph_lint::LintFixFailureDetail>,
626 },
627
628 #[error("{failure}")]
630 AuditObligation {
631 #[source]
633 failure: Box<AuditObligationFailure>,
634 domain_result: serde_json::Value,
636 },
637
638 #[error("storage: {0}")]
639 Storage(#[from] khive_storage::StorageError),
640
641 #[error("query: {0}")]
642 Query(#[from] khive_query::QueryError),
643
644 #[error("not found: {0}")]
645 NotFound(String),
646
647 #[error("invalid input: {0}")]
648 InvalidInput(String),
649
650 #[error("invalid input: {0}")]
651 UnknownVerb(String),
652
653 #[error("unconfigured: {0} is not set")]
654 Unconfigured(String),
655
656 #[error("unknown embedding model: {0}")]
657 UnknownModel(String),
658
659 #[error("embedding: {0}")]
660 Embedding(#[from] lattice_embed::EmbedError),
661
662 #[error("ambiguous: {0}")]
663 Ambiguous(String),
664
665 #[error("fusion: {0}")]
666 Fusion(#[from] khive_fusion::FuseError),
667
668 #[error("unknown fusion strategy: {0}")]
672 UnknownFusionStrategy(String),
673
674 #[error("internal: {0}")]
675 Internal(String),
676
677 #[error("audit batch incompatible event store: {0}")]
683 IncompatibleEventStore(String),
684
685 #[error("guarded edge write refused: {0}")]
686 GuardedWriteFailed(GuardedWriteFailure),
687
688 #[error("missing pack dependency: {0}")]
689 MissingPackDependency(MissingPackDependency),
690
691 #[error("missing pack dependencies: {0}")]
692 MissingPackDependencies(MissingPackDependencies),
693
694 #[error("{0}")]
695 CircularPackDependency(CircularPackDependency),
696
697 #[error("pack '{name}' registered twice (indices {first_idx} and {second_idx})")]
698 PackRedeclared {
699 name: String,
700 first_idx: usize,
701 second_idx: usize,
702 },
703
704 #[error(
708 "verb collision: verb {verb:?} declared by both pack {first_pack:?} and pack \
709 {second_pack:?}; rename one handler or use Visibility::Subhandler for internal verbs"
710 )]
711 VerbCollision {
712 verb: String,
713 first_pack: String,
714 second_pack: String,
715 },
716
717 #[error(
719 "pack {pack:?} handler {verb:?} declares request-envelope parameter {param:?}; rename the verb argument"
720 )]
721 ReservedEnvelopeParam {
722 pack: String,
723 verb: String,
724 param: String,
725 },
726
727 #[error("{}", .0.message)]
745 RefusedWithReceipt(Box<ReceiptRefusal>),
746
747 #[error(transparent)]
751 RefusedWithEvents { context: RefusalEventContext },
752
753 #[error(transparent)]
756 WithResolution { context: ResolutionFailureContext },
757
758 #[error("permission denied for verb {verb:?}: {reason}")]
768 PermissionDenied {
769 verb: String,
770 reason: String,
771 receipt: Box<DenialReceipt>,
772 },
773
774 #[error("gate unavailable for verb {verb:?}: {reason}")]
780 GateUnavailable { verb: String, reason: String },
781
782 #[error("{0}")]
786 Khive(khive_types::KhiveError),
787
788 #[error("not found in this namespace")]
793 NamespaceMismatch { id: uuid::Uuid },
794
795 #[error("ambiguous prefix {prefix:?}: matches {}", format_uuid_list(matches))]
801 AmbiguousPrefix {
802 prefix: String,
803 matches: Vec<uuid::Uuid>,
804 },
805
806 #[error(
811 "cross-backend merge is not supported: \
812 into_id {into_id} is on backend '{into_backend}', \
813 from_id {from_id} is on backend '{from_backend}'. \
814 Both entities must be on the same backend to merge."
815 )]
816 CrossBackendMergeUnsupported {
817 into_id: uuid::Uuid,
818 from_id: uuid::Uuid,
819 into_backend: String,
820 from_backend: String,
821 },
822
823 #[error("unknown remote: {name:?}")]
826 UnknownRemote { name: String },
827
828 #[error("remote cache missing for remote={remote:?} namespace={namespace:?}")]
830 RemoteCacheMissing { remote: String, namespace: String },
831
832 #[error("ambiguous id {id:?}: matched {count} records")]
834 AmbiguousId { id: String, count: usize },
835
836 #[error("cross-namespace write denied: cannot write to remote namespace {namespace:?}")]
838 CrossNamespaceWrite { namespace: String },
839
840 #[error("remote fetch error for remote={remote:?}: {message}")]
843 RemoteFetchError { remote: String, message: String },
844
845 #[error(
851 "write budget exceeded: max_new_entries={max_new_entries}, \
852 attempted_new_entries={attempted_new_entries}"
853 )]
854 WriteBudgetExceeded {
855 max_new_entries: u64,
856 attempted_new_entries: u64,
857 },
858
859 #[error("write blocked: {0}")]
865 SecretDetected(crate::secret_gate::SecretMatch),
866
867 #[error("{operation} exceeded its {budget_ms}ms deadline (elapsed {elapsed_ms}ms)")]
881 DeadlineExceeded {
882 operation: String,
883 budget_ms: u64,
884 elapsed_ms: u64,
885 },
886}
887
888impl From<khive_db::SqliteError> for RuntimeError {
892 fn from(error: khive_db::SqliteError) -> Self {
893 Self::Storage(match error {
894 khive_db::SqliteError::RequestReadStopped(error) => error,
895 error => error.into_storage_error(khive_storage::StorageCapability::Sql, "runtime"),
896 })
897 }
898}
899
900impl RuntimeError {
901 pub fn with_resolution(self, resolution: ResolutionFacts) -> Self {
903 Self::WithResolution {
904 context: ResolutionFailureContext {
905 source: Box::new(self),
906 resolution,
907 },
908 }
909 }
910
911 pub fn with_refusal_events(self, mut recordings: Vec<RefusalEventRecording>) -> Self {
914 if recordings.is_empty() {
915 return self;
916 }
917 match self {
918 Self::RefusedWithEvents { mut context } => {
919 context.recordings.append(&mut recordings);
920 Self::RefusedWithEvents { context }
921 }
922 source => Self::RefusedWithEvents {
923 context: RefusalEventContext {
924 source: Box::new(source),
925 recordings,
926 },
927 },
928 }
929 }
930
931 pub fn refusal_source(&self) -> &Self {
934 let mut error = self;
935 loop {
936 error = match error {
937 Self::RefusedWithEvents { context } => &context.source,
938 Self::WithResolution { context } => context.source.as_ref(),
939 Self::LintFix { source, .. } => source.as_ref(),
940 _ => return error,
941 };
942 }
943 }
944
945 pub fn is_stream_policy_refusal(&self) -> bool {
952 matches!(self.refusal_source(), Self::Khive(error)
953 if error.kind() == khive_types::ErrorKind::Conflict
954 && error.details().and_then(|details| details.get("reason"))
955 == Some("stream_member"))
956 }
957
958 pub fn permission_denied(verb: impl Into<String>, reason: impl Into<String>) -> Self {
960 Self::PermissionDenied {
961 verb: verb.into(),
962 reason: reason.into(),
963 receipt: Box::new(DenialReceipt::not_audited()),
964 }
965 }
966
967 pub fn internal_with_context(context: impl fmt::Display, source: impl fmt::Display) -> Self {
972 Self::Internal(format!("{context}: {source}"))
973 }
974
975 pub fn channel_ingest_failure_class(&self) -> ChannelIngestFailureClass {
983 let source = self.refusal_source();
984 let reason = source.variant_name();
985 if source.retryable_failure_context().is_some() {
986 ChannelIngestFailureClass::Retryable { reason }
987 } else if matches!(source, Self::SecretDetected(_)) {
988 ChannelIngestFailureClass::Permanent { reason }
989 } else {
990 ChannelIngestFailureClass::Unknown { reason }
991 }
992 }
993
994 fn variant_name(&self) -> &'static str {
996 match self {
997 Self::AuditObligation { .. } => "AuditObligation",
998 Self::Storage(_) => "Storage",
999 Self::Query(_) => "Query",
1000 Self::NotFound(_) => "NotFound",
1001 Self::InvalidInput(_) => "InvalidInput",
1002 Self::UnknownVerb(_) => "UnknownVerb",
1003 Self::Unconfigured(_) => "Unconfigured",
1004 Self::UnknownModel(_) => "UnknownModel",
1005 Self::Embedding(_) => "Embedding",
1006 Self::Ambiguous(_) => "Ambiguous",
1007 Self::Fusion(_) => "Fusion",
1008 Self::UnknownFusionStrategy(_) => "UnknownFusionStrategy",
1009 Self::Internal(_) => "Internal",
1010 Self::GuardedWriteFailed(_) => "GuardedWriteFailed",
1011 Self::MissingPackDependency(_) => "MissingPackDependency",
1012 Self::MissingPackDependencies(_) => "MissingPackDependencies",
1013 Self::CircularPackDependency(_) => "CircularPackDependency",
1014 Self::PackRedeclared { .. } => "PackRedeclared",
1015 Self::VerbCollision { .. } => "VerbCollision",
1016 Self::ReservedEnvelopeParam { .. } => "ReservedEnvelopeParam",
1017 Self::PermissionDenied { .. } => "PermissionDenied",
1018 Self::GateUnavailable { .. } => "GateUnavailable",
1019 Self::Khive(_) => "Khive",
1020 Self::NamespaceMismatch { .. } => "NamespaceMismatch",
1021 Self::AmbiguousPrefix { .. } => "AmbiguousPrefix",
1022 Self::CrossBackendMergeUnsupported { .. } => "CrossBackendMergeUnsupported",
1023 Self::UnknownRemote { .. } => "UnknownRemote",
1024 Self::RemoteCacheMissing { .. } => "RemoteCacheMissing",
1025 Self::AmbiguousId { .. } => "AmbiguousId",
1026 Self::CrossNamespaceWrite { .. } => "CrossNamespaceWrite",
1027 Self::RemoteFetchError { .. } => "RemoteFetchError",
1028 Self::WriteBudgetExceeded { .. } => "WriteBudgetExceeded",
1029 Self::SecretDetected(_) => "SecretDetected",
1030 Self::DeadlineExceeded { .. } => "DeadlineExceeded",
1031 Self::IncompatibleEventStore(_) => "IncompatibleEventStore",
1032 Self::RefusedWithReceipt { .. } => "RefusedWithReceipt",
1033 Self::RefusedWithEvents { context } => context.source.variant_name(),
1034 Self::WithResolution { context } => context.source.variant_name(),
1035 Self::LintFix { source, .. } => source.variant_name(),
1036 }
1037 }
1038
1039 pub fn writer_pool_checkout_timeout_context(&self) -> Option<WriterPoolCheckoutTimeoutContext> {
1046 let (sqlite_error, capability, operation) = match self.refusal_source() {
1047 Self::Storage(khive_storage::StorageError::Driver {
1048 capability,
1049 operation,
1050 source,
1051 }) => (
1052 source.downcast_ref::<khive_db::SqliteError>()?,
1053 Some(*capability),
1054 Some(operation.to_string()),
1055 ),
1056 _ => return None,
1057 };
1058
1059 let khive_db::SqliteError::WriterPoolCheckoutTimeout { timeout } = sqlite_error else {
1060 return None;
1061 };
1062 Some(WriterPoolCheckoutTimeoutContext {
1063 timeout: *timeout,
1064 capability,
1065 operation,
1066 })
1067 }
1068
1069 pub fn admission_failure_context(&self) -> Option<AdmissionFailureContext> {
1074 let source = self.refusal_source();
1075 if let Some(context) = self.writer_pool_checkout_timeout_context() {
1076 return Some(AdmissionFailureContext {
1077 stage: WRITER_POOL_CHECKOUT_TIMEOUT_STAGE,
1078 timeout: context.timeout,
1079 capability: context.capability,
1080 operation: context.operation,
1081 pool_identity: None,
1082 scope: None,
1083 retry_after_ms: None,
1084 });
1085 }
1086 if let Self::Storage(khive_storage::StorageError::WriteQueueFull { timeout_ms }) = source {
1087 return Some(AdmissionFailureContext {
1088 stage: WRITER_QUEUE_SATURATED_STAGE,
1089 timeout: Duration::from_millis(*timeout_ms),
1090 capability: None,
1091 operation: None,
1092 pool_identity: None,
1093 scope: Some(WRITER_ADMISSION_SCOPE),
1094 retry_after_ms: Some(*timeout_ms),
1095 });
1096 }
1097 if let Self::Storage(khive_storage::StorageError::AdmissionTimeout {
1098 operation,
1099 timeout_ms,
1100 pool_identity,
1101 }) = source
1102 {
1103 return Some(AdmissionFailureContext {
1104 stage: STORAGE_ADMISSION_TIMEOUT_STAGE,
1105 timeout: Duration::from_millis(*timeout_ms),
1106 capability: None,
1107 operation: Some(operation.to_string()),
1108 pool_identity: pool_identity.clone(),
1109 scope: None,
1110 retry_after_ms: None,
1111 });
1112 }
1113 None
1114 }
1115
1116 pub fn writer_task_failure_context(&self) -> Option<WriterTaskFailureContext> {
1120 let Self::Storage(error) = self.refusal_source() else {
1121 return None;
1122 };
1123 match error.without_sqlite_write_stage() {
1124 khive_storage::StorageError::WriterTaskRequestFailed {
1125 request_state,
1126 source,
1127 } => Some(WriterTaskFailureContext {
1128 stage: WRITER_TASK_REQUEST_FAILED_STAGE,
1129 request_state: *request_state,
1130 task_terminated: false,
1131 retryable: source.is_retryable(),
1132 }),
1133 khive_storage::StorageError::WriterTaskTerminated { request_state, .. } => {
1134 Some(WriterTaskFailureContext {
1135 stage: WRITER_TASK_TERMINATED_STAGE,
1136 request_state: *request_state,
1137 task_terminated: true,
1138 retryable: false,
1139 })
1140 }
1141 _ => None,
1142 }
1143 }
1144
1145 pub fn retryable_failure_context(&self) -> Option<RetryableFailureContext> {
1148 let mut enclosing = self;
1149 loop {
1150 match enclosing {
1151 Self::LintFix { .. } => return None,
1152 Self::RefusedWithEvents { context } => enclosing = &context.source,
1153 Self::WithResolution { context } => enclosing = &context.source,
1154 _ => break,
1155 }
1156 }
1157
1158 let source = self.refusal_source();
1159 if let Some(context) = self.admission_failure_context() {
1160 return Some(context.into());
1161 }
1162 if let Self::Storage(error) = source {
1163 if let khive_storage::StorageError::WriterTaskBusy { timeout_ms } =
1164 error.without_sqlite_write_stage()
1165 {
1166 return Some(RetryableFailureContext {
1167 stage: WRITER_TASK_BEGIN_BUSY_STAGE,
1168 timeout: Duration::from_millis(*timeout_ms),
1169 capability: None,
1170 operation: Some("writer_task_begin".to_string()),
1171 pool_identity: None,
1172 scope: None,
1173 retry_after_ms: None,
1174 });
1175 }
1176 }
1177 if let Self::Storage(khive_storage::StorageError::ReadTransactionAgeEvicted {
1178 operation,
1179 max_age_secs,
1180 }) = source
1181 {
1182 return Some(RetryableFailureContext {
1183 stage: READ_TX_AGE_EVICTED_STAGE,
1184 timeout: Duration::from_secs(*max_age_secs),
1185 capability: Some(khive_storage::StorageCapability::Sql),
1186 operation: Some(operation.to_string()),
1187 pool_identity: None,
1188 scope: None,
1189 retry_after_ms: None,
1190 });
1191 }
1192 if let Self::Storage(
1193 khive_storage::StorageError::ReadTransactionAgeEvictionCleanupFailed {
1194 operation,
1195 max_age_secs,
1196 ..
1197 },
1198 ) = source
1199 {
1200 return Some(RetryableFailureContext {
1201 stage: READ_TX_AGE_EVICTED_STAGE,
1202 timeout: Duration::from_secs(*max_age_secs),
1203 capability: Some(khive_storage::StorageCapability::Sql),
1204 operation: Some(operation.to_string()),
1205 pool_identity: None,
1206 scope: None,
1207 retry_after_ms: None,
1208 });
1209 }
1210 None
1211 }
1212}
1213
1214pub fn fts_text_leg_or_err<T>(
1221 result: Result<Vec<T>, RuntimeError>,
1222 context: &'static str,
1223 query: &str,
1224) -> RuntimeResult<Vec<T>> {
1225 match result {
1226 Ok(hits) => Ok(hits),
1227 Err(RuntimeError::Storage(se)) if se.is_fts5_syntax_error() => {
1228 tracing::warn!(
1229 error = %se,
1230 query = %query,
1231 context,
1232 "FTS text leg failed on a parser syntax error; failing loud (#569)"
1233 );
1234 Err(RuntimeError::InvalidInput(format!(
1235 "{context}: FTS query could not be parsed: {se}"
1236 )))
1237 }
1238 Err(e) => Err(e),
1239 }
1240}
1241
1242fn format_uuid_list(uuids: &[uuid::Uuid]) -> String {
1243 uuids
1244 .iter()
1245 .map(uuid::Uuid::to_string)
1246 .collect::<Vec<_>>()
1247 .join(", ")
1248}
1249
1250impl From<khive_types::EntityTypeError> for RuntimeError {
1254 fn from(e: khive_types::EntityTypeError) -> Self {
1255 Self::InvalidInput(e.to_string())
1256 }
1257}
1258
1259impl From<khive_types::KhiveError> for RuntimeError {
1260 fn from(e: khive_types::KhiveError) -> Self {
1261 Self::Khive(e)
1262 }
1263}
1264
1265#[cfg(test)]
1266#[path = "error_refusal_event_context_tests.rs"]
1267mod refusal_event_context_tests;
1268
1269#[cfg(test)]
1270#[path = "error_stream_policy_refusal_tests.rs"]
1271mod stream_policy_refusal_tests;
1272
1273#[cfg(test)]
1274#[path = "error_channel_ingest_failure_class_tests.rs"]
1275mod channel_ingest_failure_class_tests;