#[derive(Debug, Clone)]
pub struct PgConnection {
pub host: String,
pub port: u16,
pub user: String,
pub password: String,
pub database: String,
pub connect_timeout_secs: u32,
}
impl PgConnection {
pub fn env(&self) -> Vec<(String, String)> {
vec![
("PGHOST".to_string(), self.host.clone()),
("PGPORT".to_string(), self.port.to_string()),
("PGUSER".to_string(), self.user.clone()),
("PGPASSWORD".to_string(), self.password.clone()),
("PGDATABASE".to_string(), self.database.clone()),
(
"PGCONNECT_TIMEOUT".to_string(),
self.connect_timeout_secs.to_string(),
),
]
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SqlInvocation {
pub argv: Vec<String>,
pub env: Vec<(String, String)>,
}
const PG_EXCLUDE_EXTENSION_MIN_MAJOR: u32 = 17;
pub fn pg_server_major(server_version_num: &str) -> Option<u32> {
server_version_num
.trim()
.parse::<u32>()
.ok()
.map(|n| n / 10000)
}
pub fn pg_dump_invocation(
conn: &PgConnection,
data_only: bool,
server_version_num: &str,
exclude_extensions: &[String],
no_publications: bool,
) -> SqlInvocation {
let mut argv = vec![
"pg_dump".to_string(),
"--no-owner".to_string(),
"--no-privileges".to_string(),
];
if no_publications {
argv.push("--no-publications".to_string());
}
let narrow = !exclude_extensions.is_empty()
&& pg_server_major(server_version_num).is_some_and(|m| m >= PG_EXCLUDE_EXTENSION_MIN_MAJOR);
if narrow {
for ext in exclude_extensions {
argv.push(format!("--exclude-extension={ext}"));
}
} else {
argv.push("--no-comments".to_string());
}
argv.push("-Fc".to_string());
if data_only {
argv.push("--data-only".to_string());
argv.push("--exclude-table-data=boatramp_migrations.*".to_string());
}
SqlInvocation {
argv,
env: conn.env(),
}
}
pub fn pg_restore_invocation(conn: &PgConnection, data_only: bool) -> SqlInvocation {
let mut argv = vec![
"pg_restore".to_string(),
"--no-owner".to_string(),
"--no-privileges".to_string(),
"--exit-on-error".to_string(),
"-d".to_string(),
conn.database.clone(),
];
if data_only {
argv.push("--data-only".to_string());
}
SqlInvocation {
argv,
env: conn.env(),
}
}
pub fn psql_invocation(conn: &PgConnection, sql: &str) -> SqlInvocation {
let argv = vec![
"psql".to_string(),
"-v".to_string(),
"ON_ERROR_STOP=1".to_string(),
"-tAc".to_string(),
sql.to_string(),
];
SqlInvocation {
argv,
env: conn.env(),
}
}
pub fn pg_restore_shadow_invocation(conn: &PgConnection, shadow_db: &str) -> SqlInvocation {
let argv = vec![
"pg_restore".to_string(),
"--no-owner".to_string(),
"--no-privileges".to_string(),
"--exit-on-error".to_string(),
"--single-transaction".to_string(),
"-d".to_string(),
shadow_db.to_string(),
];
SqlInvocation {
argv,
env: conn.env(),
}
}
#[derive(Debug, Clone)]
pub struct MysqlConnection {
pub host: String,
pub port: u16,
pub user: String,
pub password: String,
pub database: String,
pub connect_timeout_secs: u32,
}
impl MysqlConnection {
pub fn env(&self) -> Vec<(String, String)> {
vec![("MYSQL_PWD".to_string(), self.password.clone())]
}
pub fn conn_args(&self) -> Vec<String> {
vec![
format!("--host={}", self.host),
format!("--port={}", self.port),
format!("--user={}", self.user),
]
}
fn connect_timeout_arg(&self) -> String {
format!("--connect-timeout={}", self.connect_timeout_secs)
}
}
pub fn mysqldump_invocation(conn: &MysqlConnection, data_only: bool) -> SqlInvocation {
let mut argv = vec![
"mysqldump".to_string(),
"--single-transaction".to_string(),
"--no-tablespaces".to_string(),
"--set-gtid-purged=OFF".to_string(),
];
argv.extend(conn.conn_args());
if data_only {
argv.push("--no-create-info".to_string());
argv.push(conn.database.clone());
} else {
argv.push("--databases".to_string());
argv.push(conn.database.clone());
}
SqlInvocation {
argv,
env: conn.env(),
}
}
pub fn mysql_restore_invocation(conn: &MysqlConnection, data_only: bool) -> SqlInvocation {
let mut argv = vec!["mysql".to_string()];
argv.extend(conn.conn_args());
argv.push(conn.connect_timeout_arg());
if data_only {
argv.push(conn.database.clone());
}
SqlInvocation {
argv,
env: conn.env(),
}
}
pub fn mysqldump_tables_only_invocation(conn: &MysqlConnection) -> SqlInvocation {
let mut argv = vec![
"mysqldump".to_string(),
"--single-transaction".to_string(),
"--no-tablespaces".to_string(),
"--set-gtid-purged=OFF".to_string(),
];
argv.extend(conn.conn_args());
argv.push(conn.database.clone());
SqlInvocation {
argv,
env: conn.env(),
}
}
pub fn mysqldump_capture_base_invocation(conn: &MysqlConnection) -> SqlInvocation {
let mut argv = vec![
"mysqldump".to_string(),
"--single-transaction".to_string(),
"--no-tablespaces".to_string(),
"--source-data=2".to_string(),
"--set-gtid-purged=COMMENTED".to_string(),
];
argv.extend(conn.conn_args());
argv.push(conn.database.clone());
SqlInvocation {
argv,
env: conn.env(),
}
}
pub fn mysqlbinlog_capture_invocation(
conn: &MysqlConnection,
live_db: &str,
shadow_db: &str,
base_gtid: &str,
start_log: &str,
) -> SqlInvocation {
let argv = vec![
"mysqlbinlog".to_string(),
"--read-from-remote-server".to_string(),
format!("--host={}", conn.host),
format!("--port={}", conn.port),
format!("--user={}", conn.user),
format!("--rewrite-db={live_db}->{shadow_db}"),
format!("--database={shadow_db}"),
"--skip-gtids".to_string(),
format!("--exclude-gtids={base_gtid}"),
"--to-last-log".to_string(),
start_log.to_string(),
];
SqlInvocation {
argv,
env: conn.env(),
}
}
pub fn mysql_shadow_invocation(conn: &MysqlConnection, shadow_db: &str) -> SqlInvocation {
let mut argv = vec!["mysql".to_string()];
argv.extend(conn.conn_args());
argv.push(conn.connect_timeout_arg());
argv.push("--local-infile=0".to_string());
argv.push(shadow_db.to_string());
SqlInvocation {
argv,
env: conn.env(),
}
}
pub fn mysql_admin_invocation(conn: &MysqlConnection) -> SqlInvocation {
let mut argv = vec!["mysql".to_string()];
argv.extend(conn.conn_args());
argv.push(conn.connect_timeout_arg());
argv.push("-N".to_string());
argv.push("--batch".to_string());
SqlInvocation {
argv,
env: conn.env(),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn conn() -> PgConnection {
PgConnection {
host: "10.0.0.5".to_string(),
port: 5432,
user: "svc".to_string(),
password: "sup3r-s3cret-pw".to_string(),
database: "app".to_string(),
connect_timeout_secs: 10,
}
}
#[test]
fn data_only_toggles_the_flag_on_both_sides() {
let dump_do = pg_dump_invocation(&conn(), true, "160004", &[], false);
let dump_full = pg_dump_invocation(&conn(), false, "160004", &[], false);
assert!(dump_do.argv.iter().any(|a| a == "--data-only"));
assert!(!dump_full.argv.iter().any(|a| a == "--data-only"));
assert!(
dump_do
.argv
.iter()
.any(|a| a == "--exclude-table-data=boatramp_migrations.*")
);
assert!(
!dump_full
.argv
.iter()
.any(|a| a == "--exclude-table-data=boatramp_migrations.*")
);
let restore_do = pg_restore_invocation(&conn(), true);
let restore_full = pg_restore_invocation(&conn(), false);
assert!(restore_do.argv.iter().any(|a| a == "--data-only"));
assert!(!restore_full.argv.iter().any(|a| a == "--data-only"));
for inv in [&restore_do, &restore_full] {
let pos = inv
.argv
.iter()
.position(|a| a == "-d")
.expect("pg_restore argv must carry -d");
assert_eq!(
inv.argv.get(pos + 1).map(String::as_str),
Some(conn().database.as_str()),
"the target database must follow -d"
);
}
}
#[test]
fn pg_server_major_parses_the_version_num_banner() {
assert_eq!(pg_server_major("160004"), Some(16));
assert_eq!(pg_server_major("170000"), Some(17));
assert_eq!(pg_server_major(" 180002 "), Some(18));
assert_eq!(pg_server_major("not-a-number"), None);
}
#[test]
fn no_publications_toggles_the_dump_flag() {
let with = pg_dump_invocation(&conn(), false, "160004", &[], true);
assert!(with.argv.iter().any(|a| a == "--no-publications"));
let without = pg_dump_invocation(&conn(), false, "160004", &[], false);
assert!(!without.argv.iter().any(|a| a == "--no-publications"));
}
#[test]
fn exclude_extension_narrows_no_comments_only_on_pg17_with_a_shadow_allowlist() {
let exts = vec!["vector".to_string(), "pg_trgm".to_string()];
let has = |inv: &SqlInvocation, a: &str| inv.argv.iter().any(|x| x == a);
let has_exclude = |inv: &SqlInvocation, e: &str| {
inv.argv
.iter()
.any(|x| x == &format!("--exclude-extension={e}"))
};
let pg17 = pg_dump_invocation(&conn(), false, "170000", &exts, false);
assert!(!has(&pg17, "--no-comments"), "PG17 keeps object comments");
assert!(has_exclude(&pg17, "vector") && has_exclude(&pg17, "pg_trgm"));
let pg16 = pg_dump_invocation(&conn(), false, "160004", &exts, false);
assert!(
has(&pg16, "--no-comments"),
"PG16 has no --exclude-extension → blunt strip"
);
assert!(
!pg16
.argv
.iter()
.any(|a| a.starts_with("--exclude-extension")),
"PG16's pg_dump does not know --exclude-extension"
);
let sidecar = pg_dump_invocation(&conn(), false, "170000", &[], false);
assert!(has(&sidecar, "--no-comments"));
assert!(
!sidecar
.argv
.iter()
.any(|a| a.starts_with("--exclude-extension"))
);
let unknown = pg_dump_invocation(&conn(), false, "garbage", &exts, false);
assert!(has(&unknown, "--no-comments"));
assert!(
!unknown
.argv
.iter()
.any(|a| a.starts_with("--exclude-extension"))
);
}
#[test]
fn shadow_builders_split_trusted_ddl_from_untrusted_apply() {
let c = conn();
let ddl = psql_invocation(&c, "CREATE DATABASE \"app__shadow_1\" OWNER \"owner\"");
assert_eq!(ddl.argv[0], "psql");
assert!(ddl.argv.iter().any(|a| a == "ON_ERROR_STOP=1"));
assert!(
ddl.argv
.iter()
.any(|a| a == "CREATE DATABASE \"app__shadow_1\" OWNER \"owner\"")
);
let apply = pg_restore_shadow_invocation(&c, "app__shadow_1");
assert_eq!(apply.argv[0], "pg_restore");
assert!(apply.argv.iter().any(|a| a == "--single-transaction"));
assert!(apply.argv.iter().any(|a| a == "--exit-on-error"));
assert!(apply.argv.iter().any(|a| a == "--no-owner"));
assert!(apply.argv.iter().any(|a| a == "--no-privileges"));
assert!(
!apply.argv.iter().any(|a| a == "--data-only"),
"the chunked pg shadow path is FULL-restore only (default #1)"
);
assert!(
!apply
.argv
.iter()
.any(|a| a == "-j" || a.starts_with("--jobs")),
"NEVER parallel-restore a piped archive"
);
let pos = apply.argv.iter().position(|a| a == "-d").expect("-d");
assert_eq!(
apply.argv.get(pos + 1).map(String::as_str),
Some("app__shadow_1"),
"the shadow db (NOT conn.database) follows -d"
);
for inv in [&ddl, &apply] {
assert!(
inv.env
.iter()
.any(|(k, v)| k == "PGPASSWORD" && v == &c.password)
);
for a in &inv.argv {
assert!(
!a.contains(&c.password),
"credential leaked into argv: {a:?}"
);
}
}
}
#[test]
fn mysql_shadow_builders_are_tables_only_and_target_the_shadow() {
let c = myconn();
let dump = mysqldump_tables_only_invocation(&c);
assert_eq!(dump.argv[0], "mysqldump");
assert!(!dump.argv.iter().any(|a| a == "--databases"));
assert!(!dump.argv.iter().any(|a| a == "--no-create-info"));
assert!(!dump.argv.iter().any(|a| a.starts_with("--connect-timeout")));
assert!(
dump.argv.iter().any(|a| a == "--set-gtid-purged=OFF"),
"tables-only dump must suppress gtid/sql_log_bin statements: {:?}",
dump.argv
);
assert!(
dump.argv.last().map(String::as_str) == Some("app"),
"the db is positional (tables only): {:?}",
dump.argv
);
let apply = mysql_shadow_invocation(&c, "app__shadow_1");
assert_eq!(apply.argv[0], "mysql");
assert!(apply.argv.iter().any(|a| a == "app__shadow_1"));
assert!(!apply.argv.iter().any(|a| a == "app"));
assert!(apply.argv.iter().any(|a| a == "--local-infile=0"));
let admin = mysql_admin_invocation(&c);
assert_eq!(admin.argv[0], "mysql");
assert!(admin.argv.iter().any(|a| a == "-N"));
assert!(admin.argv.iter().any(|a| a == "--batch"));
assert!(
!admin
.argv
.iter()
.any(|a| a == "app" || a == "app__shadow_1"),
"the admin connection names no default database: {:?}",
admin.argv
);
for inv in [&dump, &apply, &admin] {
assert!(
inv.env
.iter()
.any(|(k, v)| k == "MYSQL_PWD" && v == &c.password)
);
assert!(inv.argv.iter().any(|a| a == &format!("--host={}", c.host)));
for a in &inv.argv {
assert!(
!a.contains(&c.password),
"credential leaked into argv: {a:?}"
);
}
}
}
#[test]
fn mysql_true_delta_capture_and_replay_builders() {
let c = myconn();
let base = mysqldump_capture_base_invocation(&c);
assert_eq!(base.argv[0], "mysqldump");
assert!(base.argv.iter().any(|a| a == "--source-data=2"));
assert!(base.argv.iter().any(|a| a == "--set-gtid-purged=COMMENTED"));
assert!(!base.argv.iter().any(|a| a == "--databases")); assert_eq!(base.argv.last().map(String::as_str), Some("app"));
let cap = mysqlbinlog_capture_invocation(
&c,
"app",
"app__shadow_1",
"uuid:1-9",
"mysql-bin.000007",
);
assert_eq!(cap.argv[0], "mysqlbinlog");
assert!(cap.argv.iter().any(|a| a == "--read-from-remote-server"));
assert!(
!cap.argv.iter().any(|a| a == "--raw"),
"capture is NON-raw (db-scoped SQL text), never --raw (whole-server disclosure): {:?}",
cap.argv
);
assert!(
cap.argv
.iter()
.any(|a| a == "--rewrite-db=app->app__shadow_1")
);
assert!(
cap.argv.iter().any(|a| a == "--database=app__shadow_1"),
"--database must filter on the REWRITTEN (shadow) name: {:?}",
cap.argv
);
assert!(cap.argv.iter().any(|a| a == "--skip-gtids"));
assert!(cap.argv.iter().any(|a| a == "--exclude-gtids=uuid:1-9"));
assert!(cap.argv.iter().any(|a| a == "--to-last-log"));
assert_eq!(
cap.argv.last().map(String::as_str),
Some("mysql-bin.000007")
);
assert!(cap.argv.iter().any(|a| a == &format!("--user={}", c.user)));
for inv in [&base, &cap] {
assert!(
inv.env
.iter()
.any(|(k, v)| k == "MYSQL_PWD" && v == &c.password)
);
for a in &inv.argv {
assert!(
!a.contains(&c.password),
"credential leaked into argv: {a:?}"
);
assert!(!a.contains("://"), "no DSN in argv: {a:?}");
}
}
}
#[test]
fn credential_is_in_the_env_never_in_argv() {
let c = conn();
for inv in [
pg_dump_invocation(&c, true, "160004", &[], false),
pg_dump_invocation(&c, false, "170000", &["vector".to_string()], false),
pg_restore_invocation(&c, false),
] {
assert!(
inv.env
.iter()
.any(|(k, v)| k == "PGPASSWORD" && v == &c.password),
"PGPASSWORD must be set in the environment"
);
for a in &inv.argv {
assert!(
!a.contains(&c.password),
"credential leaked into argv: {a:?}"
);
}
for a in &inv.argv {
assert!(
!a.contains("://"),
"a DSN must never be built into argv: {a:?}"
);
}
}
}
#[test]
fn env_carries_the_discrete_connection_and_a_bounded_timeout() {
let e = conn().env();
let get = |k: &str| e.iter().find(|(kk, _)| kk == k).map(|(_, v)| v.clone());
assert_eq!(get("PGHOST").as_deref(), Some("10.0.0.5"));
assert_eq!(get("PGPORT").as_deref(), Some("5432"));
assert_eq!(get("PGUSER").as_deref(), Some("svc"));
assert_eq!(get("PGDATABASE").as_deref(), Some("app"));
assert_eq!(get("PGCONNECT_TIMEOUT").as_deref(), Some("10"));
}
fn myconn() -> MysqlConnection {
MysqlConnection {
host: "10.0.0.6".to_string(),
port: 3306,
user: "svc".to_string(),
password: "sup3r-s3cret-pw".to_string(),
database: "app".to_string(),
connect_timeout_secs: 10,
}
}
#[test]
fn mysql_data_only_toggles_and_credential_stays_in_env() {
let dump_do = mysqldump_invocation(&myconn(), true);
let dump_full = mysqldump_invocation(&myconn(), false);
assert!(dump_do.argv.iter().any(|a| a == "--no-create-info"));
assert!(!dump_full.argv.iter().any(|a| a == "--no-create-info"));
assert!(dump_full.argv.iter().any(|a| a == "--databases"));
assert!(!dump_do.argv.iter().any(|a| a == "--databases"));
for d in [&dump_do, &dump_full] {
assert!(
d.argv.iter().any(|a| a == "--set-gtid-purged=OFF"),
"mysqldump must carry --set-gtid-purged=OFF (gtid-safety): {:?}",
d.argv
);
}
for d in [&dump_do, &dump_full] {
assert!(
!d.argv.iter().any(|a| a.starts_with("--connect-timeout")),
"mysqldump must not carry --connect-timeout"
);
}
let restore_do = mysql_restore_invocation(&myconn(), true);
let restore_full = mysql_restore_invocation(&myconn(), false);
assert!(
restore_full
.argv
.iter()
.any(|a| a.starts_with("--connect-timeout")),
"the mysql restore client should carry --connect-timeout"
);
assert!(restore_do.argv.iter().any(|a| a == "app"));
assert!(!restore_full.argv.iter().any(|a| a == "app"));
let c = myconn();
for inv in [
mysqldump_invocation(&c, true),
mysql_restore_invocation(&c, false),
] {
assert!(
inv.env
.iter()
.any(|(k, v)| k == "MYSQL_PWD" && v == &c.password),
"MYSQL_PWD must be set in the environment"
);
assert!(
inv.argv.iter().any(|a| a == &format!("--host={}", c.host)),
"the connection host must be explicit in argv: {:?}",
inv.argv
);
for a in &inv.argv {
assert!(
!a.contains(&c.password),
"credential leaked into argv: {a:?}"
);
assert!(
!a.contains("://"),
"a DSN must never be built into argv: {a:?}"
);
}
}
}
#[test]
fn mysql_env_carries_the_discrete_connection() {
let c = myconn();
let e = c.env();
assert_eq!(e.len(), 1, "env carries only MYSQL_PWD");
assert_eq!(e[0].0, "MYSQL_PWD");
assert_eq!(e[0].1, "sup3r-s3cret-pw");
let a = c.conn_args();
assert!(a.iter().any(|x| x == "--host=10.0.0.6"));
assert!(a.iter().any(|x| x == "--port=3306"));
assert!(a.iter().any(|x| x == "--user=svc"));
assert!(
!a.iter().any(|x| x.contains("sup3r-s3cret-pw")),
"password never in conn_args"
);
}
}