runledger_postgres/
error.rs

1use std::{fmt, sync::Arc};
2
3use sqlx::error::ErrorKind;
4
5mod classify;
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub enum QueryErrorCategory {
9    Conflict,
10    Validation,
11    Forbidden,
12    Internal,
13}
14
15/// Stable semantic kinds for database errors that drive runtime policy.
16///
17/// Human- and machine-readable error codes remain available through
18/// [`QueryError::code`]. This enum is deliberately smaller: it covers errors
19/// whose handling must stay compile-checked across Runledger crates instead of
20/// depending on duplicated string literals.
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub enum QueryErrorKind {
23    JobLeaseOwnerMismatch,
24    JobInvalidCompletionProgress,
25    JobInvalidContinuationDelay,
26    JobInvalidRetryTiming,
27    JobUnstartedClaimReleaseNotApplicable,
28    JobWorkflowHandlerContinuationNotEnabled,
29    JobWorkflowRequeueNotSupported,
30    WorkflowReleaseConflict,
31}
32
33/// Query-error fields that are safe to emit at application logging boundaries.
34///
35/// This deliberately excludes the internal message and SQLx source because
36/// either may contain payload values, idempotency keys, or database policy
37/// details. Keeping the safe projection as a distinct type makes it harder for
38/// callers to accidentally widen structured logs with raw diagnostics.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub(crate) struct SanitizedQueryErrorDiagnostics<'a> {
41    code: &'a str,
42    sqlstate: Option<&'a str>,
43    constraint: Option<&'a str>,
44}
45
46impl<'a> SanitizedQueryErrorDiagnostics<'a> {
47    #[must_use]
48    pub(crate) const fn from_code(code: &'a str) -> Self {
49        Self {
50            code,
51            sqlstate: None,
52            constraint: None,
53        }
54    }
55
56    #[must_use]
57    pub(crate) const fn code(self) -> &'a str {
58        self.code
59    }
60
61    #[must_use]
62    pub(crate) const fn sqlstate(self) -> Option<&'a str> {
63        self.sqlstate
64    }
65
66    #[must_use]
67    pub(crate) const fn constraint(self) -> Option<&'a str> {
68        self.constraint
69    }
70}
71
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub struct FrameworkConstraintSpec {
74    category: QueryErrorCategory,
75    code: &'static str,
76    client_message: &'static str,
77}
78
79impl FrameworkConstraintSpec {
80    #[must_use]
81    pub const fn new(
82        category: QueryErrorCategory,
83        code: &'static str,
84        client_message: &'static str,
85    ) -> Self {
86        Self {
87            category,
88            code,
89            client_message,
90        }
91    }
92
93    #[must_use]
94    pub const fn category(&self) -> QueryErrorCategory {
95        self.category
96    }
97
98    #[must_use]
99    pub const fn code(&self) -> &'static str {
100        self.code
101    }
102
103    #[must_use]
104    pub const fn client_message(&self) -> &'static str {
105        self.client_message
106    }
107}
108
109#[derive(Clone)]
110pub struct QueryError {
111    category: QueryErrorCategory,
112    kind: Option<QueryErrorKind>,
113    code: &'static str,
114    client_message: &'static str,
115    sqlstate: Option<String>,
116    constraint: Option<String>,
117    message: String,
118    source: Option<Arc<sqlx::Error>>,
119}
120
121impl QueryError {
122    #[must_use]
123    pub fn from_classified(
124        category: QueryErrorCategory,
125        code: &'static str,
126        client_message: &'static str,
127        internal_message: impl Into<String>,
128    ) -> Self {
129        Self {
130            category,
131            kind: None,
132            code,
133            client_message,
134            sqlstate: None,
135            constraint: None,
136            message: internal_message.into(),
137            source: None,
138        }
139    }
140
141    #[must_use]
142    pub(crate) fn from_classified_with_kind(
143        category: QueryErrorCategory,
144        kind: QueryErrorKind,
145        code: &'static str,
146        client_message: &'static str,
147        internal_message: impl Into<String>,
148    ) -> Self {
149        Self {
150            category,
151            kind: Some(kind),
152            code,
153            client_message,
154            sqlstate: None,
155            constraint: None,
156            message: internal_message.into(),
157            source: None,
158        }
159    }
160
161    #[must_use]
162    pub(crate) fn from_classified_sqlx_with_kind(
163        category: QueryErrorCategory,
164        kind: QueryErrorKind,
165        code: &'static str,
166        client_message: &'static str,
167        internal_message: impl Into<String>,
168        source: sqlx::Error,
169    ) -> Self {
170        let (sqlstate, constraint) = source
171            .as_database_error()
172            .map(|database_error| {
173                (
174                    database_error.code().map(|code| code.into_owned()),
175                    database_error.constraint().map(ToOwned::to_owned),
176                )
177            })
178            .unwrap_or((None, None));
179
180        Self {
181            category,
182            kind: Some(kind),
183            code,
184            client_message,
185            sqlstate,
186            constraint,
187            message: internal_message.into(),
188            source: Some(Arc::new(source)),
189        }
190    }
191
192    #[must_use]
193    pub fn from_sqlx_with_constraint_classifier<F>(
194        error: sqlx::Error,
195        context: Option<&str>,
196        classify_constraint: F,
197    ) -> Self
198    where
199        F: Fn(&str) -> Option<FrameworkConstraintSpec>,
200    {
201        let (sqlstate, constraint, spec, raw_message) = if let Some(db) = error.as_database_error()
202        {
203            let sqlstate = db.code().map(|code| code.into_owned());
204            let constraint = db.constraint().map(ToOwned::to_owned);
205            let spec = classify_query_error_with_constraint_classifier(
206                &db.kind(),
207                sqlstate.as_deref(),
208                constraint.as_deref(),
209                classify_constraint,
210            );
211            (sqlstate, constraint, spec, db.message().to_owned())
212        } else {
213            (
214                None,
215                None,
216                QueryErrorSpec::internal().into(),
217                error.to_string(),
218            )
219        };
220
221        let message = match context {
222            Some(ctx) => format!("{ctx}: {raw_message}"),
223            None => raw_message,
224        };
225
226        Self {
227            category: spec.category(),
228            kind: None,
229            code: spec.code(),
230            client_message: spec.client_message(),
231            sqlstate,
232            constraint,
233            message,
234            source: Some(Arc::new(error)),
235        }
236    }
237
238    pub(crate) fn from_sqlx(error: sqlx::Error, context: Option<&str>) -> Self {
239        Self::from_sqlx_with_constraint_classifier(error, context, |_| None)
240    }
241
242    #[must_use]
243    pub const fn category(&self) -> QueryErrorCategory {
244        self.category
245    }
246
247    /// Returns a stable semantic kind when this error participates in
248    /// cross-crate runtime policy.
249    #[must_use]
250    pub const fn kind(&self) -> Option<QueryErrorKind> {
251        self.kind
252    }
253
254    #[must_use]
255    pub const fn code(&self) -> &'static str {
256        self.code
257    }
258
259    #[must_use]
260    pub const fn client_message(&self) -> &'static str {
261        self.client_message
262    }
263
264    #[must_use]
265    pub fn sqlstate(&self) -> Option<&str> {
266        self.sqlstate.as_deref()
267    }
268
269    #[must_use]
270    pub fn constraint(&self) -> Option<&str> {
271        self.constraint.as_deref()
272    }
273
274    #[must_use]
275    pub(crate) fn sanitized_diagnostics(&self) -> SanitizedQueryErrorDiagnostics<'_> {
276        SanitizedQueryErrorDiagnostics {
277            code: self.code,
278            sqlstate: self.sqlstate(),
279            constraint: self.constraint(),
280        }
281    }
282
283    #[must_use]
284    pub fn internal_message(&self) -> &str {
285        &self.message
286    }
287
288    /// Returns the underlying SQLx error for trusted diagnostics.
289    ///
290    /// Public [`Display`](fmt::Display) and [`Debug`](fmt::Debug) output for
291    /// [`QueryError`] is sanitized, but the returned source may contain raw
292    /// database details. Do not log or expose it on untrusted boundaries without
293    /// redaction.
294    #[must_use]
295    pub fn source_arc(&self) -> Option<Arc<sqlx::Error>> {
296        self.source.clone()
297    }
298
299    #[must_use]
300    pub fn reclassified_with_constraint_classifier<F>(mut self, classify_constraint: F) -> Self
301    where
302        F: Fn(&str) -> Option<FrameworkConstraintSpec>,
303    {
304        let Some(spec) = self.constraint.as_deref().and_then(classify_constraint) else {
305            return self;
306        };
307
308        self.category = spec.category();
309        self.kind = None;
310        self.code = spec.code();
311        self.client_message = spec.client_message();
312        self
313    }
314}
315
316impl fmt::Debug for QueryError {
317    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
318        f.debug_struct("QueryError")
319            .field("category", &self.category)
320            .field("kind", &self.kind)
321            .field("code", &self.code)
322            .field("client_message", &self.client_message)
323            .field("sqlstate", &self.sqlstate)
324            .field("constraint", &self.constraint)
325            .field("has_source", &self.source.is_some())
326            .finish()
327    }
328}
329
330impl fmt::Display for QueryError {
331    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
332        write!(f, "{}", self.client_message)
333    }
334}
335
336impl std::error::Error for QueryError {
337    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
338        self.source
339            .as_deref()
340            .map(|source| source as &(dyn std::error::Error + 'static))
341    }
342}
343
344#[derive(Debug, Clone, Copy)]
345struct QueryErrorSpec {
346    category: QueryErrorCategory,
347    code: &'static str,
348    client_message: &'static str,
349}
350
351impl QueryErrorSpec {
352    const fn conflict(code: &'static str, client_message: &'static str) -> Self {
353        Self {
354            category: QueryErrorCategory::Conflict,
355            code,
356            client_message,
357        }
358    }
359
360    const fn validation(code: &'static str, client_message: &'static str) -> Self {
361        Self {
362            category: QueryErrorCategory::Validation,
363            code,
364            client_message,
365        }
366    }
367
368    const fn forbidden(code: &'static str, client_message: &'static str) -> Self {
369        Self {
370            category: QueryErrorCategory::Forbidden,
371            code,
372            client_message,
373        }
374    }
375
376    const fn internal() -> Self {
377        Self {
378            category: QueryErrorCategory::Internal,
379            code: "db.query_failed",
380            client_message: "Database operation failed.",
381        }
382    }
383}
384
385impl From<QueryErrorSpec> for FrameworkConstraintSpec {
386    fn from(spec: QueryErrorSpec) -> Self {
387        Self::new(spec.category, spec.code, spec.client_message)
388    }
389}
390
391#[must_use]
392pub fn classify_query_error(
393    kind: &ErrorKind,
394    sqlstate: Option<&str>,
395    constraint: Option<&str>,
396) -> FrameworkConstraintSpec {
397    classify_query_error_with_constraint_classifier(kind, sqlstate, constraint, |_| None)
398}
399
400#[must_use]
401pub fn classify_query_error_with_constraint_classifier<F>(
402    kind: &ErrorKind,
403    sqlstate: Option<&str>,
404    constraint: Option<&str>,
405    classify_constraint: F,
406) -> FrameworkConstraintSpec
407where
408    F: Fn(&str) -> Option<FrameworkConstraintSpec>,
409{
410    if let Some(spec) = constraint.and_then(classify_constraint) {
411        return spec;
412    }
413
414    classify_database_error(kind, sqlstate, constraint).into()
415}
416
417fn classify_database_error(
418    kind: &ErrorKind,
419    sqlstate: Option<&str>,
420    constraint: Option<&str>,
421) -> QueryErrorSpec {
422    if let Some(spec) = constraint.and_then(classify_constraint) {
423        return spec;
424    }
425
426    match (kind, sqlstate) {
427        (ErrorKind::UniqueViolation, _) | (_, Some("23505")) => {
428            QueryErrorSpec::conflict("db.unique_violation", "Resource already exists.")
429        }
430        (ErrorKind::ForeignKeyViolation, _) | (_, Some("23503")) => QueryErrorSpec::validation(
431            "db.related_resource_missing",
432            "Related resource does not exist.",
433        ),
434        (_, Some("23001")) => QueryErrorSpec::validation(
435            "db.related_resource_still_referenced",
436            "Related resource is still referenced and cannot be deleted.",
437        ),
438        (ErrorKind::CheckViolation, _) | (_, Some("23514")) => QueryErrorSpec::validation(
439            "db.business_rule_violation",
440            "Request violates a business rule.",
441        ),
442        (ErrorKind::NotNullViolation, _) | (_, Some("23502")) => {
443            QueryErrorSpec::validation("db.required_field_missing", "Required data is missing.")
444        }
445        (_, Some("42501")) => {
446            QueryErrorSpec::forbidden("db.permission_denied", "Operation is not allowed.")
447        }
448        _ => QueryErrorSpec::internal(),
449    }
450}
451
452fn classify_constraint(constraint: &str) -> Option<QueryErrorSpec> {
453    classify::classify_constraint(constraint)
454}
455
456#[must_use]
457pub fn classify_framework_constraint(constraint: &str) -> Option<FrameworkConstraintSpec> {
458    classify_constraint(constraint).map(FrameworkConstraintSpec::from)
459}
460
461#[must_use]
462pub fn has_framework_constraint_classifier(constraint: &str) -> bool {
463    classify_framework_constraint(constraint).is_some()
464}
465
466#[cfg(test)]
467mod tests {
468    use super::*;
469
470    #[test]
471    fn classifies_job_idempotency_constraint() {
472        let spec = classify_database_error(
473            &ErrorKind::UniqueViolation,
474            Some("23505"),
475            Some("uq_job_queue_type_idempotency_org"),
476        );
477        assert_eq!(spec.category, QueryErrorCategory::Conflict);
478        assert_eq!(spec.code, "job.already_enqueued");
479    }
480
481    #[test]
482    fn classifies_global_job_idempotency_constraint() {
483        let spec = classify_database_error(
484            &ErrorKind::UniqueViolation,
485            Some("23505"),
486            Some("uq_job_queue_type_idempotency_global"),
487        );
488        assert_eq!(spec.category, QueryErrorCategory::Conflict);
489        assert_eq!(spec.code, "job.already_enqueued");
490    }
491
492    #[test]
493    fn classifies_workflow_idempotency_constraint() {
494        let spec = classify_database_error(
495            &ErrorKind::UniqueViolation,
496            Some("23505"),
497            Some("uq_workflow_runs_type_idempotency_org"),
498        );
499        assert_eq!(spec.category, QueryErrorCategory::Conflict);
500        assert_eq!(spec.code, "workflow.already_enqueued");
501    }
502
503    #[test]
504    fn classifies_global_workflow_idempotency_constraint() {
505        let spec = classify_database_error(
506            &ErrorKind::UniqueViolation,
507            Some("23505"),
508            Some("uq_workflow_runs_type_idempotency_global"),
509        );
510        assert_eq!(spec.category, QueryErrorCategory::Conflict);
511        assert_eq!(spec.code, "workflow.already_enqueued");
512    }
513
514    #[test]
515    fn classifies_job_definition_fk_constraint() {
516        let spec = classify_database_error(
517            &ErrorKind::ForeignKeyViolation,
518            Some("23503"),
519            Some("fk_job_queue_job_type"),
520        );
521        assert_eq!(spec.category, QueryErrorCategory::Validation);
522        assert_eq!(spec.code, "job.definition_not_found");
523    }
524
525    #[test]
526    fn classifies_job_runtime_config_definition_fk_constraint() {
527        let spec = classify_database_error(
528            &ErrorKind::ForeignKeyViolation,
529            Some("23503"),
530            Some("fk_job_runtime_configs_job_type"),
531        );
532        assert_eq!(spec.category, QueryErrorCategory::Validation);
533        assert_eq!(spec.code, "job.definition_not_found");
534    }
535
536    #[test]
537    fn classifies_job_organization_fk_constraint() {
538        let spec = classify_database_error(
539            &ErrorKind::ForeignKeyViolation,
540            Some("23503"),
541            Some("fk_job_queue_organization"),
542        );
543        assert_eq!(spec.category, QueryErrorCategory::Validation);
544        assert_eq!(spec.code, "job.organization_not_found");
545    }
546
547    #[test]
548    fn classifies_workflow_linkage_symmetry_constraint() {
549        let spec = classify_database_error(
550            &ErrorKind::CheckViolation,
551            Some("23514"),
552            Some("os_workflow_job_linkage_symmetry"),
553        );
554        assert_eq!(spec.category, QueryErrorCategory::Validation);
555        assert_eq!(spec.code, "workflow.linkage_symmetry_violation");
556    }
557
558    #[test]
559    fn classifies_workflow_linkage_symmetry_trigger_table_constraint() {
560        let spec = classify_database_error(
561            &ErrorKind::CheckViolation,
562            Some("23514"),
563            Some("os_workflow_job_linkage_symmetry_trigger_table"),
564        );
565        assert_eq!(spec.category, QueryErrorCategory::Validation);
566        assert_eq!(spec.code, "workflow.linkage_symmetry_trigger_table_invalid");
567    }
568
569    #[test]
570    fn classifies_external_gate_downgrade_blocked_constraint() {
571        let spec = classify_database_error(
572            &ErrorKind::CheckViolation,
573            Some("23514"),
574            Some("os_workflow_external_gate_downgrade_waiting_runs_exist"),
575        );
576        assert_eq!(spec.category, QueryErrorCategory::Validation);
577        assert_eq!(spec.code, "workflow.external_gate_downgrade_blocked");
578    }
579
580    #[test]
581    fn custom_constraint_classifier_takes_precedence() {
582        let spec = classify_query_error_with_constraint_classifier(
583            &ErrorKind::UniqueViolation,
584            Some("23505"),
585            Some("os_custom_override"),
586            |constraint| {
587                (constraint == "os_custom_override").then_some(FrameworkConstraintSpec::new(
588                    QueryErrorCategory::Forbidden,
589                    "custom.override",
590                    "Custom override wins.",
591                ))
592            },
593        );
594        assert_eq!(spec.category(), QueryErrorCategory::Forbidden);
595        assert_eq!(spec.code(), "custom.override");
596        assert_eq!(spec.client_message(), "Custom override wins.");
597    }
598
599    #[test]
600    fn query_error_debug_omits_internal_message() {
601        let error = QueryError::from_classified(
602            QueryErrorCategory::Conflict,
603            "job.idempotency_conflict",
604            "Job enqueue retry conflicts with the existing idempotency key.",
605            "internal context includes secret-idempotency-key",
606        );
607
608        let debug = format!("{error:?}");
609        assert!(debug.contains("job.idempotency_conflict"));
610        assert!(!debug.contains("secret-idempotency-key"));
611
612        let display = error.to_string();
613        assert_eq!(
614            display,
615            "Job enqueue retry conflicts with the existing idempotency key."
616        );
617        assert!(!display.contains("secret-idempotency-key"));
618    }
619
620    #[test]
621    fn query_error_from_sqlx_uses_sanitized_display_and_debug() {
622        let error = QueryError::from_sqlx(
623            sqlx::Error::Protocol("internal secret-idempotency-key detail".into()),
624            Some("sensitive context"),
625        );
626
627        let display = error.to_string();
628        assert_eq!(display, "Database operation failed.");
629        assert!(!display.contains("secret-idempotency-key"));
630
631        let debug = format!("{error:?}");
632        assert!(debug.contains("db.query_failed"));
633        assert!(!debug.contains("secret-idempotency-key"));
634        assert!(error.internal_message().contains("secret-idempotency-key"));
635        assert!(std::error::Error::source(&error).is_some());
636        assert!(error.source_arc().is_some());
637    }
638
639    #[test]
640    fn sanitized_diagnostics_omit_internal_message_and_source() {
641        let error = QueryError::from_sqlx(
642            sqlx::Error::Protocol("database detail includes secret-idempotency-key".into()),
643            Some("sensitive context includes secret-idempotency-key"),
644        );
645
646        let diagnostics = error.sanitized_diagnostics();
647
648        assert_eq!(diagnostics.code(), "db.query_failed");
649        assert_eq!(diagnostics.sqlstate(), None);
650        assert_eq!(diagnostics.constraint(), None);
651        let debug = format!("{diagnostics:?}");
652        assert!(debug.contains("db.query_failed"));
653        assert!(!debug.contains("secret-idempotency-key"));
654    }
655
656    #[test]
657    fn typed_query_error_from_classified_sqlx_preserves_source_without_leaking_display() {
658        let error = QueryError::from_classified_sqlx_with_kind(
659            QueryErrorCategory::Conflict,
660            QueryErrorKind::WorkflowReleaseConflict,
661            "workflow.release_conflict",
662            "Workflow step release conflicted with another workflow mutation.",
663            "internal context includes secret-lock-key",
664            sqlx::Error::Protocol("database detail includes secret-lock-key".into()),
665        );
666
667        assert_eq!(error.category(), QueryErrorCategory::Conflict);
668        assert_eq!(error.kind(), Some(QueryErrorKind::WorkflowReleaseConflict));
669        assert_eq!(error.code(), "workflow.release_conflict");
670        assert_eq!(
671            error.client_message(),
672            "Workflow step release conflicted with another workflow mutation."
673        );
674        assert!(error.internal_message().contains("secret-lock-key"));
675        assert!(error.source_arc().is_some());
676        assert!(std::error::Error::source(&error).is_some());
677
678        let display = error.to_string();
679        assert_eq!(
680            display,
681            "Workflow step release conflicted with another workflow mutation."
682        );
683        assert!(!display.contains("secret-lock-key"));
684
685        let debug = format!("{error:?}");
686        assert!(debug.contains("workflow.release_conflict"));
687        assert!(debug.contains("has_source: true"));
688        assert!(!debug.contains("secret-lock-key"));
689    }
690
691    #[test]
692    fn classifies_permission_denied() {
693        let spec = classify_database_error(&ErrorKind::Other, Some("42501"), None);
694        assert_eq!(spec.category, QueryErrorCategory::Forbidden);
695        assert_eq!(spec.code, "db.permission_denied");
696    }
697
698    #[test]
699    fn falls_back_to_internal_for_unmapped_errors() {
700        let spec = classify_database_error(&ErrorKind::Other, Some("99999"), Some("not_mapped"));
701        assert_eq!(spec.category, QueryErrorCategory::Internal);
702        assert_eq!(spec.code, "db.query_failed");
703    }
704}
runledger_postgres/error.rs

runledger_postgres/
error.rs
