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("{failure}")]
622 AuditObligation {
623 #[source]
625 failure: Box<AuditObligationFailure>,
626 domain_result: serde_json::Value,
628 },
629
630 #[error("storage: {0}")]
631 Storage(#[from] khive_storage::StorageError),
632
633 #[error("query: {0}")]
634 Query(#[from] khive_query::QueryError),
635
636 #[error("not found: {0}")]
637 NotFound(String),
638
639 #[error("invalid input: {0}")]
640 InvalidInput(String),
641
642 #[error("invalid input: {0}")]
643 UnknownVerb(String),
644
645 #[error("unconfigured: {0} is not set")]
646 Unconfigured(String),
647
648 #[error("unknown embedding model: {0}")]
649 UnknownModel(String),
650
651 #[error("embedding: {0}")]
652 Embedding(#[from] lattice_embed::EmbedError),
653
654 #[error("ambiguous: {0}")]
655 Ambiguous(String),
656
657 #[error("fusion: {0}")]
658 Fusion(#[from] khive_fusion::FuseError),
659
660 #[error("unknown fusion strategy: {0}")]
664 UnknownFusionStrategy(String),
665
666 #[error("internal: {0}")]
667 Internal(String),
668
669 #[error("audit batch incompatible event store: {0}")]
675 IncompatibleEventStore(String),
676
677 #[error("guarded edge write refused: {0}")]
678 GuardedWriteFailed(GuardedWriteFailure),
679
680 #[error("missing pack dependency: {0}")]
681 MissingPackDependency(MissingPackDependency),
682
683 #[error("missing pack dependencies: {0}")]
684 MissingPackDependencies(MissingPackDependencies),
685
686 #[error("{0}")]
687 CircularPackDependency(CircularPackDependency),
688
689 #[error("pack '{name}' registered twice (indices {first_idx} and {second_idx})")]
690 PackRedeclared {
691 name: String,
692 first_idx: usize,
693 second_idx: usize,
694 },
695
696 #[error(
700 "verb collision: verb {verb:?} declared by both pack {first_pack:?} and pack \
701 {second_pack:?}; rename one handler or use Visibility::Subhandler for internal verbs"
702 )]
703 VerbCollision {
704 verb: String,
705 first_pack: String,
706 second_pack: String,
707 },
708
709 #[error(
711 "pack {pack:?} handler {verb:?} declares request-envelope parameter {param:?}; rename the verb argument"
712 )]
713 ReservedEnvelopeParam {
714 pack: String,
715 verb: String,
716 param: String,
717 },
718
719 #[error("{}", .0.message)]
737 RefusedWithReceipt(Box<ReceiptRefusal>),
738
739 #[error(transparent)]
743 RefusedWithEvents { context: RefusalEventContext },
744
745 #[error(transparent)]
748 WithResolution { context: ResolutionFailureContext },
749
750 #[error("permission denied for verb {verb:?}: {reason}")]
760 PermissionDenied {
761 verb: String,
762 reason: String,
763 receipt: Box<DenialReceipt>,
764 },
765
766 #[error("gate unavailable for verb {verb:?}: {reason}")]
772 GateUnavailable { verb: String, reason: String },
773
774 #[error("{0}")]
778 Khive(khive_types::KhiveError),
779
780 #[error("not found in this namespace")]
785 NamespaceMismatch { id: uuid::Uuid },
786
787 #[error("ambiguous prefix {prefix:?}: matches {}", format_uuid_list(matches))]
793 AmbiguousPrefix {
794 prefix: String,
795 matches: Vec<uuid::Uuid>,
796 },
797
798 #[error(
803 "cross-backend merge is not supported: \
804 into_id {into_id} is on backend '{into_backend}', \
805 from_id {from_id} is on backend '{from_backend}'. \
806 Both entities must be on the same backend to merge."
807 )]
808 CrossBackendMergeUnsupported {
809 into_id: uuid::Uuid,
810 from_id: uuid::Uuid,
811 into_backend: String,
812 from_backend: String,
813 },
814
815 #[error("unknown remote: {name:?}")]
818 UnknownRemote { name: String },
819
820 #[error("remote cache missing for remote={remote:?} namespace={namespace:?}")]
822 RemoteCacheMissing { remote: String, namespace: String },
823
824 #[error("ambiguous id {id:?}: matched {count} records")]
826 AmbiguousId { id: String, count: usize },
827
828 #[error("cross-namespace write denied: cannot write to remote namespace {namespace:?}")]
830 CrossNamespaceWrite { namespace: String },
831
832 #[error("remote fetch error for remote={remote:?}: {message}")]
835 RemoteFetchError { remote: String, message: String },
836
837 #[error(
843 "write budget exceeded: max_new_entries={max_new_entries}, \
844 attempted_new_entries={attempted_new_entries}"
845 )]
846 WriteBudgetExceeded {
847 max_new_entries: u64,
848 attempted_new_entries: u64,
849 },
850
851 #[error("write blocked: {0}")]
857 SecretDetected(crate::secret_gate::SecretMatch),
858
859 #[error("{operation} exceeded its {budget_ms}ms deadline (elapsed {elapsed_ms}ms)")]
873 DeadlineExceeded {
874 operation: String,
875 budget_ms: u64,
876 elapsed_ms: u64,
877 },
878}
879
880impl From<khive_db::SqliteError> for RuntimeError {
884 fn from(error: khive_db::SqliteError) -> Self {
885 Self::Storage(match error {
886 khive_db::SqliteError::RequestReadStopped(error) => error,
887 error => error.into_storage_error(khive_storage::StorageCapability::Sql, "runtime"),
888 })
889 }
890}
891
892impl RuntimeError {
893 pub fn with_resolution(self, resolution: ResolutionFacts) -> Self {
895 Self::WithResolution {
896 context: ResolutionFailureContext {
897 source: Box::new(self),
898 resolution,
899 },
900 }
901 }
902
903 pub fn with_refusal_events(self, mut recordings: Vec<RefusalEventRecording>) -> Self {
906 if recordings.is_empty() {
907 return self;
908 }
909 match self {
910 Self::RefusedWithEvents { mut context } => {
911 context.recordings.append(&mut recordings);
912 Self::RefusedWithEvents { context }
913 }
914 source => Self::RefusedWithEvents {
915 context: RefusalEventContext {
916 source: Box::new(source),
917 recordings,
918 },
919 },
920 }
921 }
922
923 pub fn refusal_source(&self) -> &Self {
926 let mut error = self;
927 loop {
928 error = match error {
929 Self::RefusedWithEvents { context } => &context.source,
930 Self::WithResolution { context } => context.source.as_ref(),
931 _ => return error,
932 };
933 }
934 }
935
936 pub fn is_stream_policy_refusal(&self) -> bool {
943 matches!(self.refusal_source(), Self::Khive(error)
944 if error.kind() == khive_types::ErrorKind::Conflict
945 && error.details().and_then(|details| details.get("reason"))
946 == Some("stream_member"))
947 }
948
949 pub fn permission_denied(verb: impl Into<String>, reason: impl Into<String>) -> Self {
951 Self::PermissionDenied {
952 verb: verb.into(),
953 reason: reason.into(),
954 receipt: Box::new(DenialReceipt::not_audited()),
955 }
956 }
957
958 pub fn internal_with_context(context: impl fmt::Display, source: impl fmt::Display) -> Self {
963 Self::Internal(format!("{context}: {source}"))
964 }
965
966 pub fn channel_ingest_failure_class(&self) -> ChannelIngestFailureClass {
974 let source = self.refusal_source();
975 let reason = source.variant_name();
976 if source.retryable_failure_context().is_some() {
977 ChannelIngestFailureClass::Retryable { reason }
978 } else if matches!(source, Self::SecretDetected(_)) {
979 ChannelIngestFailureClass::Permanent { reason }
980 } else {
981 ChannelIngestFailureClass::Unknown { reason }
982 }
983 }
984
985 fn variant_name(&self) -> &'static str {
987 match self {
988 Self::AuditObligation { .. } => "AuditObligation",
989 Self::Storage(_) => "Storage",
990 Self::Query(_) => "Query",
991 Self::NotFound(_) => "NotFound",
992 Self::InvalidInput(_) => "InvalidInput",
993 Self::UnknownVerb(_) => "UnknownVerb",
994 Self::Unconfigured(_) => "Unconfigured",
995 Self::UnknownModel(_) => "UnknownModel",
996 Self::Embedding(_) => "Embedding",
997 Self::Ambiguous(_) => "Ambiguous",
998 Self::Fusion(_) => "Fusion",
999 Self::UnknownFusionStrategy(_) => "UnknownFusionStrategy",
1000 Self::Internal(_) => "Internal",
1001 Self::GuardedWriteFailed(_) => "GuardedWriteFailed",
1002 Self::MissingPackDependency(_) => "MissingPackDependency",
1003 Self::MissingPackDependencies(_) => "MissingPackDependencies",
1004 Self::CircularPackDependency(_) => "CircularPackDependency",
1005 Self::PackRedeclared { .. } => "PackRedeclared",
1006 Self::VerbCollision { .. } => "VerbCollision",
1007 Self::ReservedEnvelopeParam { .. } => "ReservedEnvelopeParam",
1008 Self::PermissionDenied { .. } => "PermissionDenied",
1009 Self::GateUnavailable { .. } => "GateUnavailable",
1010 Self::Khive(_) => "Khive",
1011 Self::NamespaceMismatch { .. } => "NamespaceMismatch",
1012 Self::AmbiguousPrefix { .. } => "AmbiguousPrefix",
1013 Self::CrossBackendMergeUnsupported { .. } => "CrossBackendMergeUnsupported",
1014 Self::UnknownRemote { .. } => "UnknownRemote",
1015 Self::RemoteCacheMissing { .. } => "RemoteCacheMissing",
1016 Self::AmbiguousId { .. } => "AmbiguousId",
1017 Self::CrossNamespaceWrite { .. } => "CrossNamespaceWrite",
1018 Self::RemoteFetchError { .. } => "RemoteFetchError",
1019 Self::WriteBudgetExceeded { .. } => "WriteBudgetExceeded",
1020 Self::SecretDetected(_) => "SecretDetected",
1021 Self::DeadlineExceeded { .. } => "DeadlineExceeded",
1022 Self::IncompatibleEventStore(_) => "IncompatibleEventStore",
1023 Self::RefusedWithReceipt { .. } => "RefusedWithReceipt",
1024 Self::RefusedWithEvents { context } => context.source.variant_name(),
1025 Self::WithResolution { context } => context.source.variant_name(),
1026 }
1027 }
1028
1029 pub fn writer_pool_checkout_timeout_context(&self) -> Option<WriterPoolCheckoutTimeoutContext> {
1036 let (sqlite_error, capability, operation) = match self.refusal_source() {
1037 Self::Storage(khive_storage::StorageError::Driver {
1038 capability,
1039 operation,
1040 source,
1041 }) => (
1042 source.downcast_ref::<khive_db::SqliteError>()?,
1043 Some(*capability),
1044 Some(operation.to_string()),
1045 ),
1046 _ => return None,
1047 };
1048
1049 let khive_db::SqliteError::WriterPoolCheckoutTimeout { timeout } = sqlite_error else {
1050 return None;
1051 };
1052 Some(WriterPoolCheckoutTimeoutContext {
1053 timeout: *timeout,
1054 capability,
1055 operation,
1056 })
1057 }
1058
1059 pub fn admission_failure_context(&self) -> Option<AdmissionFailureContext> {
1064 let source = self.refusal_source();
1065 if let Some(context) = self.writer_pool_checkout_timeout_context() {
1066 return Some(AdmissionFailureContext {
1067 stage: WRITER_POOL_CHECKOUT_TIMEOUT_STAGE,
1068 timeout: context.timeout,
1069 capability: context.capability,
1070 operation: context.operation,
1071 pool_identity: None,
1072 scope: None,
1073 retry_after_ms: None,
1074 });
1075 }
1076 if let Self::Storage(khive_storage::StorageError::WriteQueueFull { timeout_ms }) = source {
1077 return Some(AdmissionFailureContext {
1078 stage: WRITER_QUEUE_SATURATED_STAGE,
1079 timeout: Duration::from_millis(*timeout_ms),
1080 capability: None,
1081 operation: None,
1082 pool_identity: None,
1083 scope: Some(WRITER_ADMISSION_SCOPE),
1084 retry_after_ms: Some(*timeout_ms),
1085 });
1086 }
1087 if let Self::Storage(khive_storage::StorageError::AdmissionTimeout {
1088 operation,
1089 timeout_ms,
1090 pool_identity,
1091 }) = source
1092 {
1093 return Some(AdmissionFailureContext {
1094 stage: STORAGE_ADMISSION_TIMEOUT_STAGE,
1095 timeout: Duration::from_millis(*timeout_ms),
1096 capability: None,
1097 operation: Some(operation.to_string()),
1098 pool_identity: pool_identity.clone(),
1099 scope: None,
1100 retry_after_ms: None,
1101 });
1102 }
1103 None
1104 }
1105
1106 pub fn writer_task_failure_context(&self) -> Option<WriterTaskFailureContext> {
1110 let Self::Storage(error) = self.refusal_source() else {
1111 return None;
1112 };
1113 match error.without_sqlite_write_stage() {
1114 khive_storage::StorageError::WriterTaskRequestFailed {
1115 request_state,
1116 source,
1117 } => Some(WriterTaskFailureContext {
1118 stage: WRITER_TASK_REQUEST_FAILED_STAGE,
1119 request_state: *request_state,
1120 task_terminated: false,
1121 retryable: source.is_retryable(),
1122 }),
1123 khive_storage::StorageError::WriterTaskTerminated { request_state, .. } => {
1124 Some(WriterTaskFailureContext {
1125 stage: WRITER_TASK_TERMINATED_STAGE,
1126 request_state: *request_state,
1127 task_terminated: true,
1128 retryable: false,
1129 })
1130 }
1131 _ => None,
1132 }
1133 }
1134
1135 pub fn retryable_failure_context(&self) -> Option<RetryableFailureContext> {
1138 let source = self.refusal_source();
1139 if let Some(context) = self.admission_failure_context() {
1140 return Some(context.into());
1141 }
1142 if let Self::Storage(error) = source {
1143 if let khive_storage::StorageError::WriterTaskBusy { timeout_ms } =
1144 error.without_sqlite_write_stage()
1145 {
1146 return Some(RetryableFailureContext {
1147 stage: WRITER_TASK_BEGIN_BUSY_STAGE,
1148 timeout: Duration::from_millis(*timeout_ms),
1149 capability: None,
1150 operation: Some("writer_task_begin".to_string()),
1151 pool_identity: None,
1152 scope: None,
1153 retry_after_ms: None,
1154 });
1155 }
1156 }
1157 if let Self::Storage(khive_storage::StorageError::ReadTransactionAgeEvicted {
1158 operation,
1159 max_age_secs,
1160 }) = source
1161 {
1162 return Some(RetryableFailureContext {
1163 stage: READ_TX_AGE_EVICTED_STAGE,
1164 timeout: Duration::from_secs(*max_age_secs),
1165 capability: Some(khive_storage::StorageCapability::Sql),
1166 operation: Some(operation.to_string()),
1167 pool_identity: None,
1168 scope: None,
1169 retry_after_ms: None,
1170 });
1171 }
1172 if let Self::Storage(
1173 khive_storage::StorageError::ReadTransactionAgeEvictionCleanupFailed {
1174 operation,
1175 max_age_secs,
1176 ..
1177 },
1178 ) = source
1179 {
1180 return Some(RetryableFailureContext {
1181 stage: READ_TX_AGE_EVICTED_STAGE,
1182 timeout: Duration::from_secs(*max_age_secs),
1183 capability: Some(khive_storage::StorageCapability::Sql),
1184 operation: Some(operation.to_string()),
1185 pool_identity: None,
1186 scope: None,
1187 retry_after_ms: None,
1188 });
1189 }
1190 None
1191 }
1192}
1193
1194pub fn fts_text_leg_or_err<T>(
1201 result: Result<Vec<T>, RuntimeError>,
1202 context: &'static str,
1203 query: &str,
1204) -> RuntimeResult<Vec<T>> {
1205 match result {
1206 Ok(hits) => Ok(hits),
1207 Err(RuntimeError::Storage(se)) if se.is_fts5_syntax_error() => {
1208 tracing::warn!(
1209 error = %se,
1210 query = %query,
1211 context,
1212 "FTS text leg failed on a parser syntax error; failing loud (#569)"
1213 );
1214 Err(RuntimeError::InvalidInput(format!(
1215 "{context}: FTS query could not be parsed: {se}"
1216 )))
1217 }
1218 Err(e) => Err(e),
1219 }
1220}
1221
1222fn format_uuid_list(uuids: &[uuid::Uuid]) -> String {
1223 uuids
1224 .iter()
1225 .map(uuid::Uuid::to_string)
1226 .collect::<Vec<_>>()
1227 .join(", ")
1228}
1229
1230impl From<khive_types::EntityTypeError> for RuntimeError {
1234 fn from(e: khive_types::EntityTypeError) -> Self {
1235 Self::InvalidInput(e.to_string())
1236 }
1237}
1238
1239impl From<khive_types::KhiveError> for RuntimeError {
1240 fn from(e: khive_types::KhiveError) -> Self {
1241 Self::Khive(e)
1242 }
1243}
1244
1245#[cfg(test)]
1246#[path = "error_refusal_event_context_tests.rs"]
1247mod refusal_event_context_tests;
1248
1249#[cfg(test)]
1250#[path = "error_stream_policy_refusal_tests.rs"]
1251mod stream_policy_refusal_tests;
1252
1253#[cfg(test)]
1254#[path = "error_channel_ingest_failure_class_tests.rs"]
1255mod channel_ingest_failure_class_tests;