1use std::borrow::Cow;
4use std::error::Error as StdError;
5use std::fmt;
6
7use thiserror::Error;
8
9use crate::blob::ContentRef;
10use crate::capability::StorageCapability;
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub enum WriterTaskRequestState {
21 NotStarted,
23 TransactionRolledBack,
28 SideEffectsUnknown,
32}
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
36#[serde(rename_all = "snake_case")]
37pub enum SqliteWriteStage {
38 #[serde(rename = "sqlite_begin_busy")]
39 Begin,
40 #[serde(rename = "sqlite_statement_failure")]
41 Statement,
42 #[serde(rename = "sqlite_commit_failure")]
43 Commit,
44}
45
46impl SqliteWriteStage {
47 pub const fn as_str(self) -> &'static str {
48 match self {
49 Self::Begin => "sqlite_begin_busy",
50 Self::Statement => "sqlite_statement_failure",
51 Self::Commit => "sqlite_commit_failure",
52 }
53 }
54}
55
56#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
58pub struct SqliteWriteFailure {
59 pub stage: SqliteWriteStage,
60 pub primary_code: i32,
61 pub extended_code: i32,
62 #[serde(default)]
64 pub settlement_unknown: bool,
65}
66
67impl fmt::Display for WriterTaskRequestState {
68 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
69 f.write_str(match self {
70 Self::NotStarted => "not_started",
71 Self::TransactionRolledBack => "transaction_rolled_back",
72 Self::SideEffectsUnknown => "side_effects_unknown",
73 })
74 }
75}
76
77#[derive(Debug, Clone, Copy, PartialEq, Eq)]
79pub enum CapacityUnavailablePhase {
80 Identity,
81 Lock,
82 LockContended,
83 Probe,
84}
85
86impl CapacityUnavailablePhase {
87 pub const fn as_str(self) -> &'static str {
88 match self {
89 Self::Identity => "identity",
90 Self::Lock => "lock",
91 Self::LockContended => "lock_contended",
92 Self::Probe => "probe",
93 }
94 }
95}
96
97impl fmt::Display for CapacityUnavailablePhase {
98 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
99 f.write_str(self.as_str())
100 }
101}
102
103#[derive(Debug, Error)]
105pub enum StorageError {
106 #[error("{source}")]
107 SqliteWrite {
108 failure: SqliteWriteFailure,
109 #[source]
110 source: Box<StorageError>,
111 },
112 #[error("{capability:?} resource not found: {resource} ({key})")]
113 NotFound {
114 capability: StorageCapability,
115 resource: &'static str,
116 key: String,
117 },
118
119 #[error("{capability:?} resource already exists: {resource} ({key})")]
120 AlreadyExists {
121 capability: StorageCapability,
122 resource: &'static str,
123 key: String,
124 },
125
126 #[error("conflict in {capability:?} during {operation}: {message}")]
127 Conflict {
128 capability: StorageCapability,
129 operation: Cow<'static, str>,
130 message: String,
131 },
132
133 #[error("invalid input for {capability:?} during {operation}: {message}")]
134 InvalidInput {
135 capability: StorageCapability,
136 operation: Cow<'static, str>,
137 message: String,
138 },
139
140 #[error("unsupported operation for {capability:?}: {operation} ({message})")]
141 Unsupported {
142 capability: StorageCapability,
143 operation: Cow<'static, str>,
144 message: String,
145 },
146
147 #[error(
152 "blob {content_ref} exceeds the {max_bytes}-byte read limit (observed at least {observed_at_least} bytes)"
153 )]
154 BlobTooLarge {
155 content_ref: ContentRef,
156 max_bytes: u64,
157 observed_at_least: u64,
158 },
159
160 #[error(
163 "blob {content_ref} metadata reports {metadata_bytes} bytes but the complete body contains {actual_bytes} bytes"
164 )]
165 BlobSizeMismatch {
166 content_ref: ContentRef,
167 metadata_bytes: u64,
168 actual_bytes: u64,
169 },
170
171 #[error("blob digest mismatch: expected {expected}, computed {actual}")]
174 BlobDigestMismatch {
175 expected: ContentRef,
176 actual: ContentRef,
177 },
178
179 #[error("pool failure during {operation}: {message}")]
180 Pool {
181 operation: Cow<'static, str>,
182 message: String,
183 },
184
185 #[error("timeout during {operation}")]
186 Timeout { operation: Cow<'static, str> },
187
188 #[error("admission timeout during {operation} after {timeout_ms}ms{pool}", pool = match .pool_identity {
194 Some(identity) => format!(" (pool: {identity})"),
195 None => String::new(),
196 })]
197 AdmissionTimeout {
198 operation: Cow<'static, str>,
199 timeout_ms: u64,
201 pool_identity: Option<String>,
203 },
204
205 #[error("sql transaction failure during {operation}: {message}")]
206 Transaction {
207 operation: Cow<'static, str>,
208 message: String,
209 },
210
211 #[error(
220 "cached read-only transaction exceeded the maximum read-transaction age \
221 ({max_age_secs}s) during {operation} and was rolled back; retry to open a fresh \
222 read snapshot"
223 )]
224 ReadTransactionAgeEvicted {
225 operation: Cow<'static, str>,
226 max_age_secs: u64,
227 },
228
229 #[error(
241 "cached read-only transaction exceeded the maximum read-transaction age \
242 ({max_age_secs}s) during {operation} but could not be cleanly rolled back \
243 ({message}); the connection was discarded, retry to open a fresh read snapshot"
244 )]
245 ReadTransactionAgeEvictionCleanupFailed {
246 operation: Cow<'static, str>,
247 max_age_secs: u64,
248 message: String,
249 },
250
251 #[error("serialization failure in {capability:?}: {message}")]
252 Serialization {
253 capability: StorageCapability,
254 message: String,
255 },
256
257 #[error("index maintenance failure in {capability:?}: {message}")]
258 IndexMaintenance {
259 capability: StorageCapability,
260 message: String,
261 },
262
263 #[error("backend driver error in {capability:?} during {operation}: {source}")]
264 Driver {
265 capability: StorageCapability,
266 operation: Cow<'static, str>,
267 #[source]
268 source: Box<dyn StdError + Send + Sync>,
269 },
270
271 #[error("write queue full: timed out after {timeout_ms}ms waiting for writer task capacity")]
276 WriteQueueFull { timeout_ms: u64 },
277
278 #[error(
283 "writer task could not begin within {timeout_ms}ms because SQLite remained busy; request was not executed"
284 )]
285 WriterTaskBusy { timeout_ms: u64 },
286
287 #[error("writer task request failed (request_state={request_state}): {source}")]
293 WriterTaskRequestFailed {
294 request_state: WriterTaskRequestState,
295 #[source]
296 source: Box<StorageError>,
297 },
298
299 #[error("writer task terminated (request_state={request_state})")]
306 WriterTaskTerminated {
307 request_state: WriterTaskRequestState,
308 sqlite_full_codes: Option<(i32, i32)>,
311 },
312
313 #[error("internal storage error: {0}")]
316 Internal(String),
317
318 #[error(
323 "KHIVE_WRITE_QUEUE=1 but no Tokio runtime context is available to spawn the writer task"
324 )]
325 WriterTaskNoRuntime,
326
327 #[error(
331 "refusing write on {capability:?} at {volume}: {available_bytes} bytes available, \
332 the {floor_bytes}-byte free-space floor plus {required_headroom_bytes} bytes of \
333 operation headroom would be violated"
334 )]
335 CapacityFloor {
336 capability: StorageCapability,
337 volume: String,
338 available_bytes: u64,
339 floor_bytes: u64,
340 required_headroom_bytes: u64,
341 },
342
343 #[error("capacity admission unavailable for {capability:?} in {phase} phase: {message}")]
345 CapacityUnavailable {
346 capability: StorageCapability,
347 phase: CapacityUnavailablePhase,
348 message: String,
349 },
350}
351
352impl StorageError {
353 pub fn sqlite_write_failure(&self) -> Option<SqliteWriteFailure> {
355 match self {
356 Self::SqliteWrite { failure, .. } => Some(*failure),
357 Self::WriterTaskRequestFailed { source, .. } => source.sqlite_write_failure(),
358 _ => None,
359 }
360 }
361
362 pub fn without_sqlite_write_stage(&self) -> &Self {
364 match self {
365 Self::SqliteWrite { source, .. } => source.without_sqlite_write_stage(),
366 _ => self,
367 }
368 }
369
370 pub fn with_sqlite_write_failure(self, failure: SqliteWriteFailure) -> Self {
372 if self.sqlite_write_failure().is_some() {
373 self
374 } else {
375 Self::SqliteWrite {
376 failure,
377 source: Box::new(self),
378 }
379 }
380 }
381 pub fn writer_task_terminated(request_state: WriterTaskRequestState) -> Self {
385 Self::WriterTaskTerminated {
386 request_state,
387 sqlite_full_codes: None,
388 }
389 }
390
391 pub fn driver(
393 capability: StorageCapability,
394 operation: impl Into<Cow<'static, str>>,
395 source: impl StdError + Send + Sync + 'static,
396 ) -> Self {
397 Self::Driver {
398 capability,
399 operation: operation.into(),
400 source: Box::new(source),
401 }
402 }
403
404 pub fn capability(&self) -> Option<StorageCapability> {
406 match self {
407 Self::NotFound { capability, .. }
408 | Self::AlreadyExists { capability, .. }
409 | Self::Conflict { capability, .. }
410 | Self::InvalidInput { capability, .. }
411 | Self::Unsupported { capability, .. }
412 | Self::Serialization { capability, .. }
413 | Self::IndexMaintenance { capability, .. }
414 | Self::Driver { capability, .. }
415 | Self::CapacityFloor { capability, .. }
416 | Self::CapacityUnavailable { capability, .. } => Some(*capability),
417 Self::BlobTooLarge { .. }
418 | Self::BlobSizeMismatch { .. }
419 | Self::BlobDigestMismatch { .. } => Some(StorageCapability::Blob),
420 Self::WriterTaskRequestFailed { source, .. } | Self::SqliteWrite { source, .. } => {
421 source.capability()
422 }
423 Self::Pool { .. }
424 | Self::Timeout { .. }
425 | Self::AdmissionTimeout { .. }
426 | Self::Transaction { .. }
427 | Self::ReadTransactionAgeEvicted { .. }
428 | Self::ReadTransactionAgeEvictionCleanupFailed { .. }
429 | Self::WriteQueueFull { .. }
430 | Self::WriterTaskBusy { .. }
431 | Self::WriterTaskTerminated { .. }
432 | Self::Internal(..)
433 | Self::WriterTaskNoRuntime => None,
434 }
435 }
436
437 pub fn is_retryable(&self) -> bool {
439 if let Self::WriterTaskRequestFailed { source, .. } | Self::SqliteWrite { source, .. } =
440 self
441 {
442 return source.is_retryable();
443 }
444 matches!(
445 self,
446 Self::Pool { .. }
447 | Self::Timeout { .. }
448 | Self::AdmissionTimeout { .. }
449 | Self::Transaction { .. }
450 | Self::ReadTransactionAgeEvicted { .. }
451 | Self::ReadTransactionAgeEvictionCleanupFailed { .. }
452 | Self::WriteQueueFull { .. }
453 | Self::WriterTaskBusy { .. }
454 | Self::CapacityUnavailable {
455 phase: CapacityUnavailablePhase::LockContended,
456 ..
457 }
458 )
459 }
460
461 pub fn is_fts5_syntax_error(&self) -> bool {
477 if let Self::WriterTaskRequestFailed { source, .. } | Self::SqliteWrite { source, .. } =
478 self
479 {
480 return source.is_fts5_syntax_error();
481 }
482 let Self::Driver {
483 capability,
484 operation,
485 source,
486 } = self
487 else {
488 return false;
489 };
490 if *capability != StorageCapability::Text || operation.as_ref() != "fts_search" {
491 return false;
492 }
493 let msg = source.to_string();
494 msg.contains("fts5: syntax error")
495 || msg.contains("fts5: parser stack overflow")
496 || msg.contains("fts5: column queries are not supported")
497 || msg.contains("fts5: phrase queries are not supported (detail")
498 || msg.contains("fts5: NEAR queries are not supported (detail")
499 }
500
501 pub fn is_unique_constraint_violation(&self) -> bool {
514 if let Self::WriterTaskRequestFailed { source, .. } | Self::SqliteWrite { source, .. } =
515 self
516 {
517 return source.is_unique_constraint_violation();
518 }
519 let Self::Driver {
520 capability,
521 operation,
522 source,
523 } = self
524 else {
525 return false;
526 };
527 if *capability != StorageCapability::Sql {
528 return false;
529 }
530 if !matches!(
531 operation.as_ref(),
532 "execute" | "pool_writer.execute" | "tx.execute"
533 ) {
534 return false;
535 }
536 source.to_string().contains("UNIQUE constraint failed")
537 }
538}
539
540#[cfg(test)]
541mod tests {
542 use super::*;
543 use std::fmt;
544
545 #[derive(Debug)]
546 struct FakeSource(String);
547
548 impl fmt::Display for FakeSource {
549 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
550 write!(f, "{}", self.0)
551 }
552 }
553
554 impl StdError for FakeSource {}
555
556 fn driver_err(operation: &'static str, message: &str) -> StorageError {
557 StorageError::driver(
558 StorageCapability::Text,
559 operation,
560 FakeSource(message.into()),
561 )
562 }
563
564 #[test]
565 fn writer_task_request_state_display_is_stable() {
566 assert_eq!(
567 WriterTaskRequestState::NotStarted.to_string(),
568 "not_started"
569 );
570 assert_eq!(
571 WriterTaskRequestState::TransactionRolledBack.to_string(),
572 "transaction_rolled_back"
573 );
574 assert_eq!(
575 WriterTaskRequestState::SideEffectsUnknown.to_string(),
576 "side_effects_unknown"
577 );
578 }
579
580 #[test]
581 fn writer_task_busy_is_retryable_without_claiming_queue_rejection() {
582 let error = StorageError::WriterTaskBusy { timeout_ms: 175 };
583 assert!(error.is_retryable());
584 assert_eq!(
585 error.to_string(),
586 "writer task could not begin within 175ms because SQLite remained busy; request was not executed"
587 );
588 assert_eq!(error.capability(), None);
589 }
590
591 #[test]
592 fn capacity_retryability_distinguishes_contention_from_refusal() {
593 for (phase, label, retryable) in [
594 (None, "floor", false),
595 (Some(CapacityUnavailablePhase::Identity), "identity", false),
596 (Some(CapacityUnavailablePhase::Lock), "lock", false),
597 (
598 Some(CapacityUnavailablePhase::LockContended),
599 "lock_contended",
600 true,
601 ),
602 (Some(CapacityUnavailablePhase::Probe), "probe", false),
603 ] {
604 let error = match phase {
605 Some(phase) => {
606 assert_eq!(phase.as_str(), label);
607 assert_eq!(phase.to_string(), label);
608 StorageError::CapacityUnavailable {
609 capability: StorageCapability::Sql,
610 phase,
611 message: "capacity admission unavailable".into(),
612 }
613 }
614 None => StorageError::CapacityFloor {
615 capability: StorageCapability::Sql,
616 volume: "volume".into(),
617 available_bytes: 99,
618 floor_bytes: 100,
619 required_headroom_bytes: 0,
620 },
621 };
622 assert_eq!(error.is_retryable(), retryable, "{label}");
623 let wrapped = StorageError::WriterTaskRequestFailed {
624 request_state: WriterTaskRequestState::TransactionRolledBack,
625 source: Box::new(error),
626 };
627 assert_eq!(wrapped.is_retryable(), retryable, "wrapped {label}");
628 }
629 }
630
631 #[test]
632 fn writer_task_request_failure_preserves_source_policy_and_rollback_state() {
633 let error = StorageError::WriterTaskRequestFailed {
634 request_state: WriterTaskRequestState::TransactionRolledBack,
635 source: Box::new(StorageError::Pool {
636 operation: "writer_task_commit".into(),
637 message: "commit refused".into(),
638 }),
639 };
640
641 assert_eq!(error.capability(), None);
642 assert!(
643 error.is_retryable(),
644 "rollback finality must not discard the source error's retry policy"
645 );
646 assert_eq!(
647 error.to_string(),
648 "writer task request failed (request_state=transaction_rolled_back): pool failure during writer_task_commit: commit refused"
649 );
650 assert_eq!(
651 StdError::source(&error).map(ToString::to_string),
652 Some("pool failure during writer_task_commit: commit refused".to_string()),
653 "the original typed storage error must remain the public source"
654 );
655 }
656
657 #[test]
658 fn writer_task_request_failure_does_not_invent_retryability() {
659 let error = StorageError::WriterTaskRequestFailed {
660 request_state: WriterTaskRequestState::TransactionRolledBack,
661 source: Box::new(StorageError::InvalidInput {
662 capability: StorageCapability::Notes,
663 operation: "append_note".into(),
664 message: "deterministic refusal".into(),
665 }),
666 };
667
668 assert!(!error.is_retryable());
669 assert_eq!(error.capability(), Some(StorageCapability::Notes));
670 }
671
672 #[test]
673 fn writer_task_terminated_is_uncapability_scoped_and_not_retryable() {
674 for request_state in [
675 WriterTaskRequestState::NotStarted,
676 WriterTaskRequestState::TransactionRolledBack,
677 WriterTaskRequestState::SideEffectsUnknown,
678 ] {
679 let error = StorageError::writer_task_terminated(request_state);
680 assert_eq!(error.capability(), None);
681 assert!(!error.is_retryable());
682 assert_eq!(
683 error.to_string(),
684 format!("writer task terminated (request_state={request_state})")
685 );
686 }
687 }
688
689 #[test]
690 fn blob_integrity_errors_are_blob_scoped_and_not_retryable() {
691 let requested = crate::blob::ContentRef::from_hex("a".repeat(64)).unwrap();
692 let actual = crate::blob::ContentRef::from_hex("b".repeat(64)).unwrap();
693 let errors = [
694 StorageError::BlobTooLarge {
695 content_ref: requested.clone(),
696 max_bytes: 8,
697 observed_at_least: 9,
698 },
699 StorageError::BlobSizeMismatch {
700 content_ref: requested.clone(),
701 metadata_bytes: 7,
702 actual_bytes: 8,
703 },
704 StorageError::BlobDigestMismatch {
705 expected: requested,
706 actual,
707 },
708 ];
709
710 for error in errors {
711 assert_eq!(error.capability(), Some(StorageCapability::Blob));
712 assert!(!error.is_retryable());
713 }
714 }
715
716 #[test]
717 fn fts5_syntax_error_at_fts_search_is_classified_as_syntax_error() {
718 let e = driver_err("fts_search", "fts5: syntax error near \"@\"");
719 assert!(e.is_fts5_syntax_error());
720 }
721
722 #[test]
723 fn fts5_parser_stack_overflow_is_classified_as_syntax_error() {
724 let e = driver_err("fts_search", "fts5: parser stack overflow");
725 assert!(e.is_fts5_syntax_error());
726 }
727
728 #[test]
729 fn fts5_unsupported_column_query_is_classified_as_syntax_error() {
730 let e = driver_err(
731 "fts_search",
732 "fts5: column queries are not supported (detail=none)",
733 );
734 assert!(e.is_fts5_syntax_error());
735 }
736
737 #[test]
738 fn timeout_is_not_classified_as_syntax_error() {
739 let e = StorageError::Timeout {
740 operation: "fts_search".into(),
741 };
742 assert!(!e.is_fts5_syntax_error());
743 }
744
745 #[test]
746 fn pool_failure_is_not_classified_as_syntax_error() {
747 let e = StorageError::Pool {
748 operation: "fts_search".into(),
749 message: "pool exhausted".into(),
750 };
751 assert!(!e.is_fts5_syntax_error());
752 }
753
754 #[test]
755 fn driver_error_at_non_search_operation_is_not_classified_as_syntax_error() {
756 let e = driver_err("open_fts_reader", "fts5: syntax error near \"@\"");
757 assert!(!e.is_fts5_syntax_error());
758 }
759
760 #[test]
761 fn driver_error_with_unrelated_message_is_not_classified_as_syntax_error() {
762 let e = driver_err("fts_search", "disk I/O error");
763 assert!(!e.is_fts5_syntax_error());
764 }
765
766 #[test]
767 fn fts5_phrase_detail_query_is_classified_as_syntax_error() {
768 let e = driver_err(
769 "fts_search",
770 "fts5: phrase queries are not supported (detail!=full)",
771 );
772 assert!(e.is_fts5_syntax_error());
773 }
774
775 #[test]
776 fn fts5_near_detail_query_is_classified_as_syntax_error() {
777 let e = driver_err(
778 "fts_search",
779 "fts5: NEAR queries are not supported (detail!=full)",
780 );
781 assert!(e.is_fts5_syntax_error());
782 }
783
784 #[test]
785 fn unprefixed_detail_message_is_not_classified_as_syntax_error() {
786 let e = driver_err(
787 "fts_search",
788 "phrase queries are not supported (detail!=full)",
789 );
790 assert!(!e.is_fts5_syntax_error());
791 }
792
793 #[test]
794 fn fts5_shadow_table_corruption_is_not_classified_as_syntax_error() {
795 let e = driver_err(
796 "fts_search",
797 "fts5: error creating shadow table notes_content: no such table",
798 );
799 assert!(!e.is_fts5_syntax_error());
800 }
801
802 #[test]
803 fn non_text_capability_is_not_classified_as_syntax_error() {
804 let e = StorageError::Driver {
805 capability: StorageCapability::Vectors,
806 operation: "fts_search".into(),
807 source: Box::new(FakeSource("fts5: syntax error near \"@\"".into())),
808 };
809 assert!(!e.is_fts5_syntax_error());
810 }
811
812 fn driver_err_sql(operation: &'static str, message: &str) -> StorageError {
813 StorageError::driver(
814 StorageCapability::Sql,
815 operation,
816 FakeSource(message.into()),
817 )
818 }
819
820 #[test]
821 fn unique_constraint_failure_at_execute_sql_capability_is_classified() {
822 let e = driver_err_sql(
823 "execute",
824 "UNIQUE constraint failed: brain_serve_ledger.namespace, \
825 brain_serve_ledger.target_id, brain_serve_ledger.query_class, \
826 brain_serve_ledger.served_at",
827 );
828 assert!(e.is_unique_constraint_violation());
829 }
830
831 #[test]
832 fn unique_constraint_failure_at_pool_writer_execute_is_classified() {
833 let e = driver_err_sql("pool_writer.execute", "UNIQUE constraint failed: t.id");
834 assert!(e.is_unique_constraint_violation());
835 }
836
837 #[test]
838 fn unique_constraint_message_at_non_execute_operation_is_not_classified() {
839 let e = driver_err_sql("query_row", "UNIQUE constraint failed: t.id");
840 assert!(!e.is_unique_constraint_violation());
841 }
842
843 #[test]
844 fn non_unique_driver_error_at_execute_is_not_classified() {
845 let e = driver_err_sql("execute", "disk I/O error");
846 assert!(!e.is_unique_constraint_violation());
847 }
848
849 #[test]
850 fn non_sql_capability_is_not_classified_as_unique_violation() {
851 let e = driver_err("execute", "UNIQUE constraint failed: t.id");
852 assert!(!e.is_unique_constraint_violation());
853 }
854
855 #[test]
856 fn timeout_is_not_classified_as_unique_violation() {
857 let e = StorageError::Timeout {
858 operation: "execute".into(),
859 };
860 assert!(!e.is_unique_constraint_violation());
861 }
862}