pgsync-postgres 0.1.1

PostgreSQL adapter for pgsync: replication protocol client, pgoutput decoder, snapshot copy, transactional sink
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

//! Mapping of PostgreSQL errors to PGSync error kinds.

use pgsync_core::{Error, ErrorKind};

/// Classify a SQLSTATE code.
pub fn kind_for_sqlstate(code: &str) -> ErrorKind {
    match code {
        // connection exceptions, admin shutdown, cannot connect now, crash shutdown
        c if c.starts_with("08") => ErrorKind::Transient,
        "57P01" | "57P02" | "57P03" | "57P04" | "57014" => ErrorKind::Transient,
        // too many connections / conflicts
        "53300" | "40001" | "40P01" | "55P03" => ErrorKind::Transient,
        // object_in_use: slot already active (stale walsender after our crash) -> retry
        "55006" => ErrorKind::Transient,
        // authentication / authorization
        c if c.starts_with("28") => ErrorKind::Config,
        "42501" => ErrorKind::Preflight,
        // undefined object (slot / publication missing)
        "42704" | "42P01" | "42703" => ErrorKind::ResyncRequired,
        // feature not supported
        "0A000" => ErrorKind::Unsupported,
        // insufficient resources (disk, memory)
        c if c.starts_with("53") => ErrorKind::Resource,
        // data exceptions / integrity violations: divergence or unsupported data
        c if c.starts_with("22") || c.starts_with("23") => ErrorKind::Consistency,
        _ => ErrorKind::Internal,
    }
}

/// Convert a `tokio_postgres::Error` into a categorised [`Error`].
pub fn from_pg(context: &str, e: tokio_postgres::Error) -> Error {
    if let Some(db) = e.as_db_error() {
        let kind = kind_for_sqlstate(db.code().code());
        let mut msg = format!(
            "{context}: {} (SQLSTATE {})",
            db.message(),
            db.code().code()
        );
        if let Some(d) = db.detail() {
            msg.push_str(&format!(": {d}"));
        }
        return Error::new(kind, msg);
    }
    if e.is_closed() {
        return Error::transient(format!("{context}: connection closed"));
    }
    let s = e.to_string();
    // tokio-postgres hides the TLS failure reason (unknown issuer, name mismatch, ...) in
    // the source chain; surface it, and treat it as configuration, not a retryable fault.
    if s.contains("TLS handshake") {
        return Error::config(format!("{context}: {}", error_chain(&e)));
    }
    // tokio-postgres wraps I/O and timeout errors without a DbError.
    let kind = if s.contains("timed out")
        || s.contains("timeout")
        || s.contains("error communicating")
        || s.contains("connection reset")
        || s.contains("broken pipe")
        || s.contains("connection refused")
        || s.contains("db error: FATAL")
        || s.contains("unexpected EOF")
        || s.contains("error connecting")
        || s.contains("Connection refused")
        || s.contains("dns")
    {
        ErrorKind::Transient
    } else {
        ErrorKind::Internal
    };
    Error::with_source(kind, context.to_owned(), e)
}

/// `Display` of an error followed by every `source()` in its chain.
pub fn error_chain(e: &(dyn std::error::Error + 'static)) -> String {
    let mut out = e.to_string();
    let mut cur = e.source();
    while let Some(s) = cur {
        out.push_str(": ");
        out.push_str(&s.to_string());
        cur = s.source();
    }
    out
}

pub fn from_io(context: &str, e: std::io::Error) -> Error {
    Error::with_source(ErrorKind::Transient, context.to_owned(), e)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn classifies_common_sqlstates() {
        assert_eq!(kind_for_sqlstate("08006"), ErrorKind::Transient);
        assert_eq!(kind_for_sqlstate("57P01"), ErrorKind::Transient);
        assert_eq!(kind_for_sqlstate("55006"), ErrorKind::Transient);
        assert_eq!(kind_for_sqlstate("28P01"), ErrorKind::Config);
        assert_eq!(kind_for_sqlstate("42501"), ErrorKind::Preflight);
        assert_eq!(kind_for_sqlstate("42704"), ErrorKind::ResyncRequired);
        assert_eq!(kind_for_sqlstate("23505"), ErrorKind::Consistency);
        assert_eq!(kind_for_sqlstate("22P02"), ErrorKind::Consistency);
        assert_eq!(kind_for_sqlstate("0A000"), ErrorKind::Unsupported);
        assert_eq!(kind_for_sqlstate("53100"), ErrorKind::Resource);
        assert_eq!(kind_for_sqlstate("XX000"), ErrorKind::Internal);
    }
}