pgsync_postgres/
errors.rs

1//! Mapping of PostgreSQL errors to PGSync error kinds.
2
3use pgsync_core::{Error, ErrorKind};
4
5/// Classify a SQLSTATE code.
6pub fn kind_for_sqlstate(code: &str) -> ErrorKind {
7    match code {
8        // connection exceptions, admin shutdown, cannot connect now, crash shutdown
9        c if c.starts_with("08") => ErrorKind::Transient,
10        "57P01" | "57P02" | "57P03" | "57P04" | "57014" => ErrorKind::Transient,
11        // too many connections / conflicts
12        "53300" | "40001" | "40P01" | "55P03" => ErrorKind::Transient,
13        // object_in_use: slot already active (stale walsender after our crash) -> retry
14        "55006" => ErrorKind::Transient,
15        // authentication / authorization
16        c if c.starts_with("28") => ErrorKind::Config,
17        "42501" => ErrorKind::Preflight,
18        // undefined object (slot / publication missing)
19        "42704" | "42P01" | "42703" => ErrorKind::ResyncRequired,
20        // feature not supported
21        "0A000" => ErrorKind::Unsupported,
22        // insufficient resources (disk, memory)
23        c if c.starts_with("53") => ErrorKind::Resource,
24        // data exceptions / integrity violations: divergence or unsupported data
25        c if c.starts_with("22") || c.starts_with("23") => ErrorKind::Consistency,
26        _ => ErrorKind::Internal,
27    }
28}
29
30/// Convert a `tokio_postgres::Error` into a categorised [`Error`].
31pub fn from_pg(context: &str, e: tokio_postgres::Error) -> Error {
32    if let Some(db) = e.as_db_error() {
33        let kind = kind_for_sqlstate(db.code().code());
34        let mut msg = format!(
35            "{context}: {} (SQLSTATE {})",
36            db.message(),
37            db.code().code()
38        );
39        if let Some(d) = db.detail() {
40            msg.push_str(&format!(": {d}"));
41        }
42        return Error::new(kind, msg);
43    }
44    if e.is_closed() {
45        return Error::transient(format!("{context}: connection closed"));
46    }
47    let s = e.to_string();
48    // tokio-postgres hides the TLS failure reason (unknown issuer, name mismatch, ...) in
49    // the source chain; surface it, and treat it as configuration, not a retryable fault.
50    if s.contains("TLS handshake") {
51        return Error::config(format!("{context}: {}", error_chain(&e)));
52    }
53    // tokio-postgres wraps I/O and timeout errors without a DbError.
54    let kind = if s.contains("timed out")
55        || s.contains("timeout")
56        || s.contains("error communicating")
57        || s.contains("connection reset")
58        || s.contains("broken pipe")
59        || s.contains("connection refused")
60        || s.contains("db error: FATAL")
61        || s.contains("unexpected EOF")
62        || s.contains("error connecting")
63        || s.contains("Connection refused")
64        || s.contains("dns")
65    {
66        ErrorKind::Transient
67    } else {
68        ErrorKind::Internal
69    };
70    Error::with_source(kind, context.to_owned(), e)
71}
72
73/// `Display` of an error followed by every `source()` in its chain.
74pub fn error_chain(e: &(dyn std::error::Error + 'static)) -> String {
75    let mut out = e.to_string();
76    let mut cur = e.source();
77    while let Some(s) = cur {
78        out.push_str(": ");
79        out.push_str(&s.to_string());
80        cur = s.source();
81    }
82    out
83}
84
85pub fn from_io(context: &str, e: std::io::Error) -> Error {
86    Error::with_source(ErrorKind::Transient, context.to_owned(), e)
87}
88
89#[cfg(test)]
90mod tests {
91    use super::*;
92
93    #[test]
94    fn classifies_common_sqlstates() {
95        assert_eq!(kind_for_sqlstate("08006"), ErrorKind::Transient);
96        assert_eq!(kind_for_sqlstate("57P01"), ErrorKind::Transient);
97        assert_eq!(kind_for_sqlstate("55006"), ErrorKind::Transient);
98        assert_eq!(kind_for_sqlstate("28P01"), ErrorKind::Config);
99        assert_eq!(kind_for_sqlstate("42501"), ErrorKind::Preflight);
100        assert_eq!(kind_for_sqlstate("42704"), ErrorKind::ResyncRequired);
101        assert_eq!(kind_for_sqlstate("23505"), ErrorKind::Consistency);
102        assert_eq!(kind_for_sqlstate("22P02"), ErrorKind::Consistency);
103        assert_eq!(kind_for_sqlstate("0A000"), ErrorKind::Unsupported);
104        assert_eq!(kind_for_sqlstate("53100"), ErrorKind::Resource);
105        assert_eq!(kind_for_sqlstate("XX000"), ErrorKind::Internal);
106    }
107}
pgsync_postgres/errors.rs

pgsync_postgres/
errors.rs
