20766227608

Committed 07 Jan 2026 12:14AM UTC coverage: 46.53% (-2.4%) from 48.884%

Build # 20766227608

Build Type

push

github

Committed by

vnvo

Commit Message

add postgres to the summary links too

Run Details

3667 of 7881 relevant lines covered (46.53%)

2.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/crates/sources/src/postgres/postgres_errors.rs

//! PostgreSQL source error types.

use deltaforge_core::SourceError;
use thiserror::Error;

#[derive(Debug, Error)]
pub enum PostgresSourceError {
    #[error("invalid postgres dsn: {0}")]
    InvalidDsn(String),

    #[error("replication connection failed: {0}")]
    ReplicationConnect(String),

    #[error("auth failed: {0}")]
    Auth(String),

    #[error("publication '{publication}' not found")]
    PublicationNotFound { publication: String },

    #[error("replication slot '{slot}' not found")]
    SlotNotFound { slot: String },

    #[error("schema resolution failed for table {schema}.{table}: {details}")]
    Schema {
        schema: String,
        table: String,
        details: String,
    },

    #[error("checkpoint load failed: {0}")]
    Checkpoint(String),

    #[error("LSN parse error: {0}")]
    LsnParse(String),

    #[error("pgoutput protocol error: {0}")]
    Protocol(String),

    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),

    #[error("database error: {0}")]
    Database(String),

    #[error("replication error: {0}")]
    Replication(String),
}

impl From<PostgresSourceError> for SourceError {
    fn from(e: PostgresSourceError) -> Self {
        match e {
            PostgresSourceError::InvalidDsn(dsn) => SourceError::Incompatible {
                details: format!("invalid PostgreSQL DSN: {dsn}").into(),
            },
            PostgresSourceError::ReplicationConnect(msg) => {
                SourceError::Connect {
                    details: msg.into(),
                }
            }
            PostgresSourceError::Auth(msg) => SourceError::Auth {
                details: msg.into(),
            },
            PostgresSourceError::PublicationNotFound { publication } => {
                SourceError::Incompatible {
                    details: format!("publication '{publication}' not found")
                        .into(),
                }
            }
            PostgresSourceError::SlotNotFound { slot } => {
                SourceError::Incompatible {
                    details: format!("replication slot '{slot}' not found")
                        .into(),
                }
            }
            PostgresSourceError::Schema {
                schema,
                table,
                details,
            } => SourceError::Schema {
                details: format!("{schema}.{table}: {details}").into(),
            },
            PostgresSourceError::Checkpoint(msg) => SourceError::Checkpoint {
                details: msg.into(),
            },
            PostgresSourceError::LsnParse(msg) => SourceError::Incompatible {
                details: format!("LSN parse error: {msg}").into(),
            },
            PostgresSourceError::Protocol(msg) => {
                SourceError::Other(anyhow::anyhow!(msg))
            }
            PostgresSourceError::Io(e) => SourceError::Io(e),
            PostgresSourceError::Database(msg) => {
                SourceError::Other(anyhow::anyhow!("database error: {msg}"))
            }
            PostgresSourceError::Replication(msg) => {
                SourceError::Other(anyhow::anyhow!("replication error: {msg}"))
            }
        }
    }
}

impl From<tokio_postgres::Error> for PostgresSourceError {
    fn from(e: tokio_postgres::Error) -> Self {
        PostgresSourceError::Database(e.to_string())
    }
}

impl From<pgwire_replication::PgWireError> for PostgresSourceError {
    fn from(e: pgwire_replication::PgWireError) -> Self {
        PostgresSourceError::Replication(e.to_string())
    }
}

1	//! PostgreSQL source error types.
2
3	use deltaforge_core::SourceError;
4	use thiserror::Error;
5
6	#[derive(Debug, Error)]
7	pub enum PostgresSourceError {
8	#[error("invalid postgres dsn: {0}")]
9	InvalidDsn(String),
10
11	#[error("replication connection failed: {0}")]
12	ReplicationConnect(String),
13
14	#[error("auth failed: {0}")]
15	Auth(String),
16
17	#[error("publication '{publication}' not found")]
18	PublicationNotFound { publication: String },
19
20	#[error("replication slot '{slot}' not found")]
21	SlotNotFound { slot: String },
22
23	#[error("schema resolution failed for table {schema}.{table}: {details}")]
24	Schema {
25	schema: String,
26	table: String,
27	details: String,
28	},
29
30	#[error("checkpoint load failed: {0}")]
31	Checkpoint(String),
32
33	#[error("LSN parse error: {0}")]
34	LsnParse(String),
35
36	#[error("pgoutput protocol error: {0}")]
37	Protocol(String),
38
39	#[error("I/O error: {0}")]
40	Io(#[from] std::io::Error),
41
42	#[error("database error: {0}")]
43	Database(String),
44
45	#[error("replication error: {0}")]
46	Replication(String),
47	}
48
49	impl From<PostgresSourceError> for SourceError {
50	fn from(e: PostgresSourceError) -> Self {	×
51	match e {	×
52	PostgresSourceError::InvalidDsn(dsn) => SourceError::Incompatible {	×
53	details: format!("invalid PostgreSQL DSN: {dsn}").into(),	×
54	},	×
55	PostgresSourceError::ReplicationConnect(msg) => {	×
56	SourceError::Connect {	×
57	details: msg.into(),	×
58	}	×
59	}
60	PostgresSourceError::Auth(msg) => SourceError::Auth {	×
61	details: msg.into(),	×
62	},	×
63	PostgresSourceError::PublicationNotFound { publication } => {	×
64	SourceError::Incompatible {	×
65	details: format!("publication '{publication}' not found")	×
66	.into(),	×
67	}	×
68	}
69	PostgresSourceError::SlotNotFound { slot } => {	×
70	SourceError::Incompatible {	×
71	details: format!("replication slot '{slot}' not found")	×
72	.into(),	×
73	}	×
74	}
75	PostgresSourceError::Schema {
76	schema,	×
77	table,	×
78	details,	×
79	} => SourceError::Schema {	×
80	details: format!("{schema}.{table}: {details}").into(),	×
81	},	×
82	PostgresSourceError::Checkpoint(msg) => SourceError::Checkpoint {	×
83	details: msg.into(),	×
84	},	×
85	PostgresSourceError::LsnParse(msg) => SourceError::Incompatible {	×
86	details: format!("LSN parse error: {msg}").into(),	×
87	},	×
88	PostgresSourceError::Protocol(msg) => {	×
89	SourceError::Other(anyhow::anyhow!(msg))	×
90	}
91	PostgresSourceError::Io(e) => SourceError::Io(e),	×
92	PostgresSourceError::Database(msg) => {	×
93	SourceError::Other(anyhow::anyhow!("database error: {msg}"))	×
94	}
95	PostgresSourceError::Replication(msg) => {	×
96	SourceError::Other(anyhow::anyhow!("replication error: {msg}"))	×
97	}
98	}
99	}	×
100	}
101
102	impl From<tokio_postgres::Error> for PostgresSourceError {
103	fn from(e: tokio_postgres::Error) -> Self {	×
104	PostgresSourceError::Database(e.to_string())	×
105	}	×
106	}
107
108	impl From<pgwire_replication::PgWireError> for PostgresSourceError {
109	fn from(e: pgwire_replication::PgWireError) -> Self {	×
110	PostgresSourceError::Replication(e.to_string())	×
111	}	×
112	}

vnvo / deltaforge / 20766227608

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous