23416749279

Committed 23 Mar 2026 12:51AM UTC coverage: 43.642% (-8.7%) from 52.336%

Build # 23416749279

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #60 from vnvo/chaos

Chaos testing: observability improvements, pipeline lifecycle fixes, and chaos testing expansion

Coverage Stats

135 of 3691 new or added lines in 45 files covered. (3.66%)

23 existing lines in 14 files now uncovered.

8940 of 20485 relevant lines covered (43.64%)

181.12 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 and loop control.

use common::RetryOutcome;
use deltaforge_core::SourceError;
use thiserror::Error;
use tracing::{error, warn};

// =============================================================================
// PostgresSourceError - for helper functions (NOT the event loop)
// =============================================================================

#[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("checkpoint error: {0}")]
    Checkpoint(String),

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

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

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

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

pub type PostgresSourceResult<T> = Result<T, PostgresSourceError>;

impl From<PostgresSourceError> for SourceError {
    fn from(e: PostgresSourceError) -> Self {
        match e {
            PostgresSourceError::InvalidDsn(msg) => SourceError::Incompatible {
                details: msg.into(),
            },
            PostgresSourceError::ReplicationConnect(msg) => {
                SourceError::Connect {
                    details: msg.into(),
                }
            }
            PostgresSourceError::Auth(msg) => SourceError::Auth {
                details: msg.into(),
            },
            PostgresSourceError::Checkpoint(msg) => SourceError::Checkpoint {
                details: msg.into(),
            },
            PostgresSourceError::LsnParse(msg) => SourceError::Incompatible {
                details: format!("LSN: {msg}").into(),
            },
            PostgresSourceError::Schema {
                schema,
                table,
                details,
            } => SourceError::Schema {
                details: format!("{schema}.{table}: {details}").into(),
            },
            PostgresSourceError::Io(e) => SourceError::Io(e),
            PostgresSourceError::Database(msg) => {
                SourceError::Other(anyhow::anyhow!("{msg}"))
            }
        }
    }
}

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

// =============================================================================
// LoopControl - for event loop control flow
// =============================================================================

#[derive(Debug)]
pub enum LoopControl {
    Reconnect,
    ReloadSchema {
        schema: Option<String>,
        table: Option<String>,
    },
    Stop,
    Fail(SourceError),
}

impl LoopControl {
    /// Classify RetryOutcome from watchdog/retry_async.
    pub fn from_replication_outcome<E: std::fmt::Display>(
        outcome: RetryOutcome<E>,
    ) -> Self {
        match outcome {
            RetryOutcome::Cancelled => Self::Stop,
            RetryOutcome::Timeout { .. } => Self::Reconnect,
            RetryOutcome::Failed(e)
            | RetryOutcome::Exhausted { last_error: e, .. } => {
                Self::from_error_message(&e.to_string())
            }
        }
    }

    /// Classify PgWireError (main event loop errors).
    #[allow(dead_code)]
    pub fn from_pgwire_error(err: pgwire_replication::PgWireError) -> Self {
        use pgwire_replication::PgWireError::*;

        match err {
            Io(ref msg) if msg.contains("permission denied") => {
                error!(error = %msg, "permission denied");
                Self::Fail(SourceError::Auth {
                    details: msg.clone().into(),
                })
            }
            Io(msg) => {
                warn!(error = %msg, "IO error, will reconnect");
                Self::Reconnect
            }

            Auth(msg) => {
                error!(error = %msg, "authentication failed");
                Self::Fail(SourceError::Auth {
                    details: msg.into(),
                })
            }

            Tls(msg) => {
                error!(error = %msg, "TLS error");
                Self::Fail(SourceError::Incompatible {
                    details: format!("TLS: {msg}").into(),
                })
            }

            Server(ref msg) => Self::classify_server_error(msg),

            Protocol(ref msg) if is_schema_error(msg) => {
                warn!(error = %msg, "schema error, will reload");
                Self::ReloadSchema {
                    schema: None,
                    table: None,
                }
            }
            Protocol(msg) => {
                warn!(error = %msg, "protocol error, will reconnect");
                Self::Reconnect
            }

            Task(msg) => {
                warn!(error = %msg, "task error, will reconnect");
                Self::Reconnect
            }

            Internal(msg) => {
                error!(error = %msg, "internal error");
                Self::Fail(SourceError::Other(anyhow::anyhow!(
                    "pgwire: {}",
                    msg
                )))
            }
        }
    }

    /// Classify tokio_postgres errors (control plane queries).
    pub fn from_tokio_postgres_error(err: &tokio_postgres::Error) -> Self {
        if let Some(db_err) = err.as_db_error() {
            let code = db_err.code().code();
            if code.starts_with("28") {
                return Self::Fail(SourceError::Auth {
                    details: err.to_string().into(),
                });
            }
            if code.starts_with("42") {
                return Self::ReloadSchema {
                    schema: None,
                    table: None,
                };
            }
            if code.starts_with("08")
                || code.starts_with("53")
                || code.starts_with("57")
            {
                return Self::Reconnect;
            }
        }
        Self::from_error_message(&err.to_string())
    }

    fn classify_server_error(msg: &str) -> Self {
        let lower = msg.to_lowercase();

        // Auth errors (28xxx)
        if lower.contains("28p01")
            || lower.contains("28000")
            || lower.contains("password authentication failed")
            || lower.contains("no pg_hba.conf entry")
        {
            error!(error = %msg, "auth failed");
            return Self::Fail(SourceError::Auth {
                details: msg.to_string().into(),
            });
        }

        // Schema errors (42xxx)
        if lower.contains("42p01")
            || lower.contains("42703")
            || lower.contains("42704")
            || is_schema_error(&lower)
        {
            warn!(error = %msg, "schema error, will reload");
            return Self::ReloadSchema {
                schema: None,
                table: None,
            };
        }

        // Slot errors
        if lower.contains("replication slot") {
            if lower.contains("does not exist") {
                error!(error = %msg, "slot not found");
                return Self::Fail(SourceError::Checkpoint {
                    details: msg.to_string().into(),
                });
            }
            if lower.contains("already active") {
                warn!(error = %msg, "slot in use, will retry");
                return Self::Reconnect;
            }
        }

        // WAL removed
        if lower.contains("requested wal segment") && lower.contains("removed")
        {
            error!(error = %msg, "WAL removed");
            return Self::Fail(SourceError::Checkpoint {
                details: msg.to_string().into(),
            });
        }

        // Admin shutdown (57P01) - this is retryable
        if lower.contains("57p01")
            || lower.contains("terminating connection")
            || lower.contains("administrator command")
        {
            warn!(error = %msg, "connection terminated by admin, will reconnect");
            return Self::Reconnect;
        }

        warn!(error = %msg, "server error, will reconnect");
        Self::Reconnect
    }

    fn from_error_message(msg: &str) -> Self {
        let lower = msg.to_lowercase();

        // Invalid DSN/connection string is a config error, not retryable
        if lower.contains("invalid connection string")
            || lower.contains("invalid dsn")
        {
            error!(error = %msg, "invalid connection string - check DSN format");
            return Self::Fail(SourceError::Incompatible {
                details: msg.to_string().into(),
            });
        }

        if is_auth_error(&lower) {
            return Self::Fail(SourceError::Auth {
                details: msg.to_string().into(),
            });
        }
        if is_connection_error(&lower) {
            return Self::Reconnect;
        }
        if is_schema_error(&lower) {
            return Self::ReloadSchema {
                schema: None,
                table: None,
            };
        }
        Self::Fail(SourceError::Other(anyhow::anyhow!("{}", msg)))
    }

    pub fn is_retryable(&self) -> bool {
        matches!(self, Self::Reconnect | Self::ReloadSchema { .. })
    }
}

fn is_auth_error(s: &str) -> bool {
    s.contains("password authentication failed")
        || s.contains("permission denied")
        || s.contains("no pg_hba.conf entry")
        || s.contains("insufficient privilege")
}

fn is_connection_error(s: &str) -> bool {
    s.contains("connection reset")
        || s.contains("connection refused")
        || s.contains("broken pipe")
        || s.contains("connection closed")
        || s.contains("timed out")
        || s.contains("unexpected eof")
        // tokio::io::AsyncReadExt::read_exact returns "early eof" on EOF
        // (io::ErrorKind::UnexpectedEof). Happens when the remote side closes
        // the TCP connection gracefully (FIN) mid-message — e.g. Toxiproxy
        // closing a proxied connection. std::io::Read::read_exact uses
        // "failed to fill whole buffer" instead, so cover both.
        || s.contains("early eof")
        || s.contains("failed to fill whole buffer")
        || s.contains("terminating connection")
        || s.contains("administrator command")
        || s.contains("57p01")
}

fn is_schema_error(s: &str) -> bool {
    (s.contains("relation") || s.contains("column") || s.contains("type"))
        && s.contains("does not exist")
}

1	//! PostgreSQL source error types and loop control.
2
3	use common::RetryOutcome;
4	use deltaforge_core::SourceError;
5	use thiserror::Error;
6	use tracing::{error, warn};
7
8	// =============================================================================
9	// PostgresSourceError - for helper functions (NOT the event loop)
10	// =============================================================================
11
12	#[derive(Debug, Error)]
13	pub enum PostgresSourceError {
14	#[error("invalid postgres dsn: {0}")]
15	InvalidDsn(String),
16
17	#[error("replication connection failed: {0}")]
18	ReplicationConnect(String),
19
20	#[error("auth failed: {0}")]
21	Auth(String),
22
23	#[error("checkpoint error: {0}")]
24	Checkpoint(String),
25
26	#[error("LSN parse error: {0}")]
27	LsnParse(String),
28
29	#[error("schema resolution failed for {schema}.{table}: {details}")]
30	Schema {
31	schema: String,
32	table: String,
33	details: String,
34	},
35
36	#[error("I/O error: {0}")]
37	Io(#[from] std::io::Error),
38
39	#[error("database error: {0}")]
40	Database(String),
41	}
42
43	pub type PostgresSourceResult<T> = Result<T, PostgresSourceError>;
44
45	impl From<PostgresSourceError> for SourceError {
46	fn from(e: PostgresSourceError) -> Self {	×
47	match e {	×
48	PostgresSourceError::InvalidDsn(msg) => SourceError::Incompatible {	×
49	details: msg.into(),	×
50	},	×
51	PostgresSourceError::ReplicationConnect(msg) => {	×
52	SourceError::Connect {	×
53	details: msg.into(),	×
54	}	×
55	}
56	PostgresSourceError::Auth(msg) => SourceError::Auth {	×
57	details: msg.into(),	×
58	},	×
59	PostgresSourceError::Checkpoint(msg) => SourceError::Checkpoint {	×
60	details: msg.into(),	×
61	},	×
62	PostgresSourceError::LsnParse(msg) => SourceError::Incompatible {	×
63	details: format!("LSN: {msg}").into(),	×
64	},	×
65	PostgresSourceError::Schema {
66	schema,	×
67	table,	×
68	details,	×
69	} => SourceError::Schema {	×
70	details: format!("{schema}.{table}: {details}").into(),	×
71	},	×
72	PostgresSourceError::Io(e) => SourceError::Io(e),	×
73	PostgresSourceError::Database(msg) => {	×
74	SourceError::Other(anyhow::anyhow!("{msg}"))	×
75	}
76	}
77	}	×
78	}
79
80	impl From<tokio_postgres::Error> for PostgresSourceError {
81	fn from(e: tokio_postgres::Error) -> Self {	×
82	PostgresSourceError::Database(e.to_string())	×
83	}	×
84	}
85
86	// =============================================================================
87	// LoopControl - for event loop control flow
88	// =============================================================================
89
90	#[derive(Debug)]
91	pub enum LoopControl {
92	Reconnect,
93	ReloadSchema {
94	schema: Option<String>,
95	table: Option<String>,
96	},
97	Stop,
98	Fail(SourceError),
99	}
100
101	impl LoopControl {
102	/// Classify RetryOutcome from watchdog/retry_async.
103	pub fn from_replication_outcome<E: std::fmt::Display>(	×
104	outcome: RetryOutcome<E>,	×
105	) -> Self {	×
106	match outcome {	×
107	RetryOutcome::Cancelled => Self::Stop,	×
108	RetryOutcome::Timeout { .. } => Self::Reconnect,	×
109	RetryOutcome::Failed(e)	×
110	\| RetryOutcome::Exhausted { last_error: e, .. } => {	×
111	Self::from_error_message(&e.to_string())	×
112	}
113	}
114	}	×
115
116	/// Classify PgWireError (main event loop errors).
117	#[allow(dead_code)]
118	pub fn from_pgwire_error(err: pgwire_replication::PgWireError) -> Self {	×
119	use pgwire_replication::PgWireError::*;
120
121	match err {	×
122	Io(ref msg) if msg.contains("permission denied") => {	×
123	error!(error = %msg, "permission denied");	×
124	Self::Fail(SourceError::Auth {	×
125	details: msg.clone().into(),	×
126	})	×
127	}
128	Io(msg) => {	×
129	warn!(error = %msg, "IO error, will reconnect");	×
130	Self::Reconnect	×
131	}
132
133	Auth(msg) => {	×
134	error!(error = %msg, "authentication failed");	×
135	Self::Fail(SourceError::Auth {	×
136	details: msg.into(),	×
137	})	×
138	}
139
140	Tls(msg) => {	×
141	error!(error = %msg, "TLS error");	×
142	Self::Fail(SourceError::Incompatible {	×
143	details: format!("TLS: {msg}").into(),	×
144	})	×
145	}
146
147	Server(ref msg) => Self::classify_server_error(msg),	×
148
149	Protocol(ref msg) if is_schema_error(msg) => {	×
150	warn!(error = %msg, "schema error, will reload");	×
151	Self::ReloadSchema {	×
152	schema: None,	×
153	table: None,	×
154	}	×
155	}
156	Protocol(msg) => {	×
157	warn!(error = %msg, "protocol error, will reconnect");	×
158	Self::Reconnect	×
159	}
160
161	Task(msg) => {	×
162	warn!(error = %msg, "task error, will reconnect");	×
163	Self::Reconnect	×
164	}
165
166	Internal(msg) => {	×
167	error!(error = %msg, "internal error");	×
168	Self::Fail(SourceError::Other(anyhow::anyhow!(	×
169	"pgwire: {}",	×
170	msg	×
171	)))	×
172	}
173	}
174	}	×
175
176	/// Classify tokio_postgres errors (control plane queries).
177	pub fn from_tokio_postgres_error(err: &tokio_postgres::Error) -> Self {	×
178	if let Some(db_err) = err.as_db_error() {	×
179	let code = db_err.code().code();	×
180	if code.starts_with("28") {	×
181	return Self::Fail(SourceError::Auth {	×
182	details: err.to_string().into(),	×
183	});	×
184	}	×
185	if code.starts_with("42") {	×
186	return Self::ReloadSchema {	×
187	schema: None,	×
188	table: None,	×
189	};	×
190	}	×
191	if code.starts_with("08")	×
192	\|\| code.starts_with("53")	×
193	\|\| code.starts_with("57")	×
194	{
195	return Self::Reconnect;	×
196	}	×
197	}	×
198	Self::from_error_message(&err.to_string())	×
199	}	×
200
201	fn classify_server_error(msg: &str) -> Self {	×
202	let lower = msg.to_lowercase();	×
203
204	// Auth errors (28xxx)
205	if lower.contains("28p01")	×
206	\|\| lower.contains("28000")	×
207	\|\| lower.contains("password authentication failed")	×
208	\|\| lower.contains("no pg_hba.conf entry")	×
209	{
210	error!(error = %msg, "auth failed");	×
211	return Self::Fail(SourceError::Auth {	×
212	details: msg.to_string().into(),	×
213	});	×
214	}	×
215
216	// Schema errors (42xxx)
217	if lower.contains("42p01")	×
218	\|\| lower.contains("42703")	×
219	\|\| lower.contains("42704")	×
220	\|\| is_schema_error(&lower)	×
221	{
222	warn!(error = %msg, "schema error, will reload");	×
223	return Self::ReloadSchema {	×
224	schema: None,	×
225	table: None,	×
226	};	×
227	}	×
228
229	// Slot errors
230	if lower.contains("replication slot") {	×
231	if lower.contains("does not exist") {	×
232	error!(error = %msg, "slot not found");	×
233	return Self::Fail(SourceError::Checkpoint {	×
234	details: msg.to_string().into(),	×
235	});	×
236	}	×
237	if lower.contains("already active") {	×
238	warn!(error = %msg, "slot in use, will retry");	×
239	return Self::Reconnect;	×
240	}	×
241	}	×
242
243	// WAL removed
244	if lower.contains("requested wal segment") && lower.contains("removed")	×
245	{
246	error!(error = %msg, "WAL removed");	×
247	return Self::Fail(SourceError::Checkpoint {	×
248	details: msg.to_string().into(),	×
249	});	×
250	}	×
251
252	// Admin shutdown (57P01) - this is retryable
253	if lower.contains("57p01")	×
254	\|\| lower.contains("terminating connection")	×
255	\|\| lower.contains("administrator command")	×
256	{
257	warn!(error = %msg, "connection terminated by admin, will reconnect");	×
258	return Self::Reconnect;	×
259	}	×
260
261	warn!(error = %msg, "server error, will reconnect");	×
262	Self::Reconnect	×
263	}	×
264
265	fn from_error_message(msg: &str) -> Self {	×
266	let lower = msg.to_lowercase();	×
267
268	// Invalid DSN/connection string is a config error, not retryable
269	if lower.contains("invalid connection string")	×
270	\|\| lower.contains("invalid dsn")	×
271	{
272	error!(error = %msg, "invalid connection string - check DSN format");	×
273	return Self::Fail(SourceError::Incompatible {	×
274	details: msg.to_string().into(),	×
275	});	×
276	}	×
277
278	if is_auth_error(&lower) {	×
279	return Self::Fail(SourceError::Auth {	×
280	details: msg.to_string().into(),	×
281	});	×
282	}	×
283	if is_connection_error(&lower) {	×
284	return Self::Reconnect;	×
285	}	×
286	if is_schema_error(&lower) {	×
287	return Self::ReloadSchema {	×
288	schema: None,	×
289	table: None,	×
290	};	×
291	}	×
292	Self::Fail(SourceError::Other(anyhow::anyhow!("{}", msg)))	×
293	}	×
294
295	pub fn is_retryable(&self) -> bool {	×
296	matches!(self, Self::Reconnect \| Self::ReloadSchema { .. })	×
297	}	×
298	}
299
300	fn is_auth_error(s: &str) -> bool {	×
301	s.contains("password authentication failed")	×
302	\|\| s.contains("permission denied")	×
303	\|\| s.contains("no pg_hba.conf entry")	×
304	\|\| s.contains("insufficient privilege")	×
305	}	×
306
307	fn is_connection_error(s: &str) -> bool {	×
308	s.contains("connection reset")	×
309	\|\| s.contains("connection refused")	×
310	\|\| s.contains("broken pipe")	×
311	\|\| s.contains("connection closed")	×
312	\|\| s.contains("timed out")	×
313	\|\| s.contains("unexpected eof")	×
314	// tokio::io::AsyncReadExt::read_exact returns "early eof" on EOF
315	// (io::ErrorKind::UnexpectedEof). Happens when the remote side closes
316	// the TCP connection gracefully (FIN) mid-message — e.g. Toxiproxy
317	// closing a proxied connection. std::io::Read::read_exact uses
318	// "failed to fill whole buffer" instead, so cover both.
NEW 319	\|\| s.contains("early eof")	×
NEW 320	\|\| s.contains("failed to fill whole buffer")	×
321	\|\| s.contains("terminating connection")	×
322	\|\| s.contains("administrator command")	×
323	\|\| s.contains("57p01")	×
324	}	×
325
326	fn is_schema_error(s: &str) -> bool {	×
327	(s.contains("relation") \|\| s.contains("column") \|\| s.contains("type"))	×
328	&& s.contains("does not exist")	×
329	}	×

vnvo / deltaforge / 23416749279

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