23696515415

Committed 28 Mar 2026 11:15PM UTC coverage: 41.639% (-1.7%) from 43.376%

Build # 23696515415

Build Type

push

github

Committed by

web-flow

Commit Message

perf: throughput optimizations, coordinator pipelining, chaos UI improvements  (#61)

* feat: postgres soak test — 120 tables, writer tasks, fault injection, schema alters

* feat: postgres backlog drain, config lab UI, writer fixes

- Add postgres support to backlog drain benchmark (run_pg)
- Add Config Lab tab with A/B comparison runner and presets
- Add PATCH proxy endpoint to chaos UI backend
- Fix pg_writer_loop: reconnect on error, f64->numeric type mismatch
- Fix escape_like: escape underscore, convert glob * to SQL %
- Promote writer errors from debug to warn level
- Add df_base/pipeline fields to SoakSource

* clippy and formatting fixes

* perf: postgres CDC optimizations, chaos UI improvements

Source hot-path optimizations (profiler-guided):
- fast_uuid_v7: atomic counter replaces getrandom syscall (3.1% -> 0.2%)
- Remove #[instrument] from read_next_event/dispatch_event
- try_send with async fallback in send_event
- Arc<Vec<RelationColumn>> and Arc<str> qualified_name per relation
- Arc<PostgresTableSchema> in LoadedSchema cache
- Static PG_VERSION LazyLock, hand-written checkpoint JSON
- Pre-allocated batch vector in coordinator

Chaos UI:
- Config Lab tab with A/B comparison runner and presets
- Service refresh button (--no-deps --force-recreate)
- Dynamic port/URL link badges from docker port
- Grafana in-flight panel color thresholds (green=draining, orange=buffering)
- Wider services card layout

* chore: upgrade pgwire-replication to v0.3, fix clippy

- Update pgwire-replication to v0.3 (buffered WAL reads)
- Adapt Io error handling for Arc<io::Error> (use err.kind() not string match)
- Include ErrorKind in connect error details
- Fix clippy: range patterns, useless format!

* fix formatting

* perf: zero-copy tuples, counter cache, LSN cache - PG matches MySQL

Zero-copy tuple parsing:
- PgColumnValue::Text/Binary now hold Bytes slices (not String/Vec)
- parse_tuple_data uses Bytes::slice() instead of copy + from_utf8_lossy
- DML h... (continued)

Coverage Stats

47 of 1157 new or added lines in 15 files covered. (4.06%)

111 existing lines in 12 files now uncovered.

9176 of 22037 relevant lines covered (41.64%)

168.57 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 err)
                if err.kind() == std::io::ErrorKind::PermissionDenied =>
            {
                error!(error = %err, kind = ?err.kind(), "permission denied");
                Self::Fail(SourceError::Auth {
                    details: format!("{err}").into(),
                })
            }
            Io(ref err)
                if matches!(
                    err.kind(),
                    std::io::ErrorKind::ConnectionRefused
                        | std::io::ErrorKind::ConnectionReset
                        | std::io::ErrorKind::ConnectionAborted
                        | std::io::ErrorKind::BrokenPipe
                        | std::io::ErrorKind::TimedOut
                        | std::io::ErrorKind::UnexpectedEof
                ) =>
            {
                warn!(error = %err, kind = ?err.kind(), "IO error, will reconnect");
                Self::Reconnect
            }
            Io(err) => {
                warn!(error = %err, kind = ?err.kind(), "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 {	×
NEW 122	Io(ref err)	×
NEW 123	if err.kind() == std::io::ErrorKind::PermissionDenied =>	×
124	{
NEW 125	error!(error = %err, kind = ?err.kind(), "permission denied");	×
126	Self::Fail(SourceError::Auth {	×
NEW 127	details: format!("{err}").into(),	×
128	})	×
129	}
NEW 130	Io(ref err)	×
NEW 131	if matches!(	×
NEW 132	err.kind(),	×
133	std::io::ErrorKind::ConnectionRefused
134	\| std::io::ErrorKind::ConnectionReset
135	\| std::io::ErrorKind::ConnectionAborted
136	\| std::io::ErrorKind::BrokenPipe
137	\| std::io::ErrorKind::TimedOut
138	\| std::io::ErrorKind::UnexpectedEof
139	) =>
140	{
NEW 141	warn!(error = %err, kind = ?err.kind(), "IO error, will reconnect");	×
NEW 142	Self::Reconnect	×
143	}
NEW 144	Io(err) => {	×
NEW 145	warn!(error = %err, kind = ?err.kind(), "IO error, will reconnect");	×
UNCOV 146	Self::Reconnect	×
147	}
148
149	Auth(msg) => {	×
150	error!(error = %msg, "authentication failed");	×
151	Self::Fail(SourceError::Auth {	×
152	details: msg.into(),	×
153	})	×
154	}
155
156	Tls(msg) => {	×
157	error!(error = %msg, "TLS error");	×
158	Self::Fail(SourceError::Incompatible {	×
159	details: format!("TLS: {msg}").into(),	×
160	})	×
161	}
162
163	Server(ref msg) => Self::classify_server_error(msg),	×
164
165	Protocol(ref msg) if is_schema_error(msg) => {	×
166	warn!(error = %msg, "schema error, will reload");	×
167	Self::ReloadSchema {	×
168	schema: None,	×
169	table: None,	×
170	}	×
171	}
172	Protocol(msg) => {	×
173	warn!(error = %msg, "protocol error, will reconnect");	×
174	Self::Reconnect	×
175	}
176
177	Task(msg) => {	×
178	warn!(error = %msg, "task error, will reconnect");	×
179	Self::Reconnect	×
180	}
181
182	Internal(msg) => {	×
183	error!(error = %msg, "internal error");	×
184	Self::Fail(SourceError::Other(anyhow::anyhow!(	×
185	"pgwire: {}",	×
186	msg	×
187	)))	×
188	}
189	}
190	}	×
191
192	/// Classify tokio_postgres errors (control plane queries).
193	pub fn from_tokio_postgres_error(err: &tokio_postgres::Error) -> Self {	×
194	if let Some(db_err) = err.as_db_error() {	×
195	let code = db_err.code().code();	×
196	if code.starts_with("28") {	×
197	return Self::Fail(SourceError::Auth {	×
198	details: err.to_string().into(),	×
199	});	×
200	}	×
201	if code.starts_with("42") {	×
202	return Self::ReloadSchema {	×
203	schema: None,	×
204	table: None,	×
205	};	×
206	}	×
207	if code.starts_with("08")	×
208	\|\| code.starts_with("53")	×
209	\|\| code.starts_with("57")	×
210	{
211	return Self::Reconnect;	×
212	}	×
213	}	×
214	Self::from_error_message(&err.to_string())	×
215	}	×
216
217	fn classify_server_error(msg: &str) -> Self {	×
218	let lower = msg.to_lowercase();	×
219
220	// Auth errors (28xxx)
221	if lower.contains("28p01")	×
222	\|\| lower.contains("28000")	×
223	\|\| lower.contains("password authentication failed")	×
224	\|\| lower.contains("no pg_hba.conf entry")	×
225	{
226	error!(error = %msg, "auth failed");	×
227	return Self::Fail(SourceError::Auth {	×
228	details: msg.to_string().into(),	×
229	});	×
230	}	×
231
232	// Schema errors (42xxx)
233	if lower.contains("42p01")	×
234	\|\| lower.contains("42703")	×
235	\|\| lower.contains("42704")	×
236	\|\| is_schema_error(&lower)	×
237	{
238	warn!(error = %msg, "schema error, will reload");	×
239	return Self::ReloadSchema {	×
240	schema: None,	×
241	table: None,	×
242	};	×
243	}	×
244
245	// Slot errors
246	if lower.contains("replication slot") {	×
247	if lower.contains("does not exist") {	×
248	error!(error = %msg, "slot not found");	×
249	return Self::Fail(SourceError::Checkpoint {	×
250	details: msg.to_string().into(),	×
251	});	×
252	}	×
253	if lower.contains("already active") {	×
254	warn!(error = %msg, "slot in use, will retry");	×
255	return Self::Reconnect;	×
256	}	×
257	}	×
258
259	// WAL removed
260	if lower.contains("requested wal segment") && lower.contains("removed")	×
261	{
262	error!(error = %msg, "WAL removed");	×
263	return Self::Fail(SourceError::Checkpoint {	×
264	details: msg.to_string().into(),	×
265	});	×
266	}	×
267
268	// Admin shutdown (57P01) - this is retryable
269	if lower.contains("57p01")	×
270	\|\| lower.contains("terminating connection")	×
271	\|\| lower.contains("administrator command")	×
272	{
273	warn!(error = %msg, "connection terminated by admin, will reconnect");	×
274	return Self::Reconnect;	×
275	}	×
276
277	warn!(error = %msg, "server error, will reconnect");	×
278	Self::Reconnect	×
279	}	×
280
281	fn from_error_message(msg: &str) -> Self {	×
282	let lower = msg.to_lowercase();	×
283
284	// Invalid DSN/connection string is a config error, not retryable
285	if lower.contains("invalid connection string")	×
286	\|\| lower.contains("invalid dsn")	×
287	{
288	error!(error = %msg, "invalid connection string - check DSN format");	×
289	return Self::Fail(SourceError::Incompatible {	×
290	details: msg.to_string().into(),	×
291	});	×
292	}	×
293
294	if is_auth_error(&lower) {	×
295	return Self::Fail(SourceError::Auth {	×
296	details: msg.to_string().into(),	×
297	});	×
298	}	×
299	if is_connection_error(&lower) {	×
300	return Self::Reconnect;	×
301	}	×
302	if is_schema_error(&lower) {	×
303	return Self::ReloadSchema {	×
304	schema: None,	×
305	table: None,	×
306	};	×
307	}	×
308	Self::Fail(SourceError::Other(anyhow::anyhow!("{}", msg)))	×
309	}	×
310
311	pub fn is_retryable(&self) -> bool {	×
312	matches!(self, Self::Reconnect \| Self::ReloadSchema { .. })	×
313	}	×
314	}
315
316	fn is_auth_error(s: &str) -> bool {	×
317	s.contains("password authentication failed")	×
318	\|\| s.contains("permission denied")	×
319	\|\| s.contains("no pg_hba.conf entry")	×
320	\|\| s.contains("insufficient privilege")	×
321	}	×
322
323	fn is_connection_error(s: &str) -> bool {	×
324	s.contains("connection reset")	×
325	\|\| s.contains("connection refused")	×
326	\|\| s.contains("broken pipe")	×
327	\|\| s.contains("connection closed")	×
328	\|\| s.contains("timed out")	×
329	\|\| s.contains("unexpected eof")	×
330	// tokio::io::AsyncReadExt::read_exact returns "early eof" on EOF
331	// (io::ErrorKind::UnexpectedEof). Happens when the remote side closes
332	// the TCP connection gracefully (FIN) mid-message — e.g. Toxiproxy
333	// closing a proxied connection. std::io::Read::read_exact uses
334	// "failed to fill whole buffer" instead, so cover both.
335	\|\| s.contains("early eof")	×
336	\|\| s.contains("failed to fill whole buffer")	×
337	\|\| s.contains("terminating connection")	×
338	\|\| s.contains("administrator command")	×
339	\|\| s.contains("57p01")	×
340	}	×
341
342	fn is_schema_error(s: &str) -> bool {	×
343	(s.contains("relation") \|\| s.contains("column") \|\| s.contains("type"))	×
344	&& s.contains("does not exist")	×
345	}	×

vnvo / deltaforge / 23696515415

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