use anyhow::{Context, bail};
use clap::{Parser, Subcommand};
use pgsync_core::ErrorKind;
use pgsync_core::config::Config;
use pgsync_postgres::job::{self, JobContext};
use pgsync_postgres::source::{CheckStatus, PreflightMode};
use std::path::PathBuf;
use tokio_util::sync::CancellationToken;
use tracing::{error, info};
mod ui;
#[derive(Parser)]
#[command(
name = "pgsync",
version,
about = "PostgreSQL logical replication engine (PGRun)"
)]
struct Cli {
#[arg(short, long, env = "PGSYNC_CONFIG", default_value = "pgsync.yaml")]
config: PathBuf,
#[arg(long, env = "PGSYNC_JSON")]
json: bool,
#[arg(long, env = "PGSYNC_LOG")]
log: Option<String>,
#[arg(long)]
quiet: bool,
#[arg(long)]
show_keys: bool,
#[command(subcommand)]
command: Command,
}
#[derive(Subcommand)]
enum Command {
Check,
Bootstrap {
#[arg(long)]
restart_snapshot: bool,
},
Run,
Status,
Cleanup {
#[arg(long)]
yes: bool,
},
}
fn init_logging(json: bool, filter: &str) {
let mut filter = filter.to_owned();
if !filter.contains("tokio_postgres") {
filter.push_str(",tokio_postgres=warn");
}
let filter = tracing_subscriber::EnvFilter::try_new(&filter)
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info,tokio_postgres=warn"));
let builder = tracing_subscriber::fmt()
.with_env_filter(filter)
.with_target(false)
.with_writer(std::io::stderr);
if json {
builder.json().flatten_event(true).init();
} else {
builder.init();
}
}
fn shutdown_token() -> CancellationToken {
let token = CancellationToken::new();
let t = token.clone();
tokio::spawn(async move {
let ctrl_c = tokio::signal::ctrl_c();
#[cfg(unix)]
{
let mut term =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
.expect("SIGTERM handler");
tokio::select! {
_ = ctrl_c => {}
_ = term.recv() => {}
}
}
#[cfg(not(unix))]
{
let _ = ctrl_c.await;
}
info!("shutdown requested; finishing the current transaction");
t.cancel();
});
token
}
#[tokio::main]
async fn main() {
let cli = Cli::parse();
let tty = std::io::IsTerminal::is_terminal(&std::io::stdout());
let pretty = !cli.json && !cli.quiet && tty;
let default_log = if pretty { "warn" } else { "info" };
let log = cli.log.clone().unwrap_or_else(|| default_log.to_owned());
init_logging(cli.json, &log);
let code = match real_main(cli, pretty).await {
Ok(()) => 0,
Err(e) => {
error!("{e:#}");
match e.downcast_ref::<pgsync_core::Error>().map(|e| e.kind()) {
Some(ErrorKind::Config) => 2,
Some(ErrorKind::Preflight) => 3,
Some(ErrorKind::ResyncRequired) => 4,
Some(ErrorKind::Unsupported) | Some(ErrorKind::Consistency) => 5,
_ => 1,
}
}
};
std::process::exit(code);
}
async fn real_main(cli: Cli, pretty: bool) -> anyhow::Result<()> {
let text = std::fs::read_to_string(&cli.config)
.with_context(|| format!("reading {}", cli.config.display()))?;
let config = Config::from_yaml(&text)?;
let cancel = shutdown_token();
let mut ctx = JobContext::new(config, cancel)?;
if pretty {
let color = std::env::var("NO_COLOR").is_err();
ctx.progress = std::sync::Arc::new(ui::Terminal::new(cli.show_keys, color));
}
info!(job = %ctx.config.job.id, source = %ctx.source.redacted(), target = %ctx.target.redacted(), "pgsync {}", env!("CARGO_PKG_VERSION"));
match cli.command {
Command::Check => {
let (out, _) = job::check(&ctx, PreflightMode::Bootstrap).await?;
if cli.json {
println!("{}", serde_json::to_string_pretty(&out)?);
} else {
for (label, report) in [("source", &out.source), ("target", &out.target)] {
println!("{label}:");
for c in &report.checks {
let tag = match c.status {
CheckStatus::Pass => "PASS",
CheckStatus::Warn => "WARN",
CheckStatus::Fail => "FAIL",
};
println!(" [{tag}] {:<28} {}", c.name, c.detail);
}
}
println!(
"{}",
if out.ok {
"OK: prerequisites satisfied"
} else {
"FAILED: fix the items marked FAIL"
}
);
}
if !out.ok {
bail!(pgsync_core::Error::preflight("preflight failed"));
}
}
Command::Bootstrap { restart_snapshot } => {
job::bootstrap(&ctx, restart_snapshot).await?;
}
Command::Run => {
job::run(&ctx).await?;
}
Command::Status => {
let out = job::status(&ctx).await?;
if cli.json {
println!("{}", serde_json::to_string_pretty(&out)?);
} else {
match &out.job {
None => println!("job {:?}: not bootstrapped", ctx.config.job.id),
Some(j) => {
println!("job {}: {}", j.job_id, j.state);
println!(
" source system {} db {}",
j.source_system_id, j.source_dbname
);
println!(" slot {} (owned: {})", j.slot, j.slot_owned);
println!(" publication {}", j.publication);
println!(
" consistent pt {}",
j.consistent_point
.map(|l| l.to_string())
.unwrap_or_default()
);
println!(" committed lsn {}", j.committed_lsn);
println!(" acked lsn {}", j.acked_lsn);
println!(" txns applied {}", j.txns_applied);
if let Some(s) = out.seconds_since_last_commit {
println!(" last commit {s:.0}s ago (source commit time)");
}
if let Some(e) = &j.last_error {
println!(" last error {e}");
}
}
}
if let Some(src) = &out.source {
if let Some(v) = &src.version {
println!("source: {}", v.split(" on ").next().unwrap_or(v));
}
}
match &out.runtime {
None => println!("process: no heartbeat recorded"),
Some(r) => {
let liveness = if r.stopped_at.is_some() {
format!("stopped at {}", r.stopped_at.as_deref().unwrap_or(""))
} else if r.stale {
format!(
"STALE (no heartbeat for {:.0}s; process probably gone)",
r.age_seconds
)
} else {
format!("alive ({:.0}s ago)", r.age_seconds)
};
println!(
"process: pid {} on {} · pgsync {} · {}",
r.pid, r.hostname, r.pgsync_version, liveness
);
println!(
" connection {} received {} server wal end {} apply lag {} bytes",
r.connection, r.received_lsn, r.server_wal_end, r.apply_lag_bytes
);
println!(
" throughput {:.1} txn/s {:.0} rows/s (txns {} rows {} reconnects {})",
r.txns_per_s,
r.rows_per_s,
r.txns_applied,
r.rows_applied,
r.reconnects
);
println!(
" memory {}",
r.rss_bytes
.map(|b| format!("{:.1} MB rss", b as f64 / 1048576.0))
.unwrap_or_else(|| "n/a".into())
);
if let Some(t) = &r.last_apply_at {
println!(" last apply {t}");
}
}
}
println!(
"snapshot: {}/{} tables copied, {} rows, {} bytes",
out.snapshot.tables_copied,
out.snapshot.tables_total,
out.snapshot.rows_copied,
out.snapshot.bytes_copied
);
if let Some(a) = &out.action {
println!("action: {a}");
}
match &out.slot {
None => println!("slot: missing on source"),
Some(s) => println!(
"slot: active={} wal_status={} confirmed_flush={} restart={} invalidation={}",
s.active,
s.wal_status.clone().unwrap_or_default(),
s.confirmed_flush_lsn
.map(|l| l.to_string())
.unwrap_or_default(),
s.restart_lsn.map(|l| l.to_string()).unwrap_or_default(),
s.invalidation_reason
.clone()
.unwrap_or_else(|| "none".into())
),
}
if let Some(l) = out.source_wal_lsn {
println!(
"source wal: {l} slot lag: {} bytes retained wal: {} bytes",
out.slot_lag_bytes.unwrap_or(0),
out.retained_wal_bytes.unwrap_or(0)
);
}
for t in &out.tables {
println!(
" table {:<32} {} rows={} bytes={}",
t.table, t.state, t.rows_copied, t.bytes_copied
);
}
}
}
Command::Cleanup { yes } => {
if !yes {
bail!(
"cleanup drops the replication slot owned by job {:?} on the source and deletes its metadata on the target; re-run with --yes to confirm",
ctx.config.job.id
);
}
for a in job::cleanup(&ctx).await? {
println!("{a}");
}
}
}
Ok(())
}