pgsync_postgres/
source.rs

1//! Source inspection and preflight checks (ordinary SQL connection, read-only).
2
3use crate::client::PgClient;
4use crate::errors::from_pg;
5use crate::replication::slot::{SlotInfo, slot_info_from_json};
6use pgsync_core::config::TableName;
7use pgsync_core::{Error, Lsn, Result};
8use serde::Serialize;
9use std::collections::BTreeSet;
10
11#[derive(Debug, Clone, Serialize)]
12pub struct ServerInfo {
13    pub version_num: i32,
14    pub version: String,
15    pub wal_level: String,
16    pub max_replication_slots: i32,
17    pub max_wal_senders: i32,
18    pub max_slot_wal_keep_size: String,
19    pub wal_sender_timeout: String,
20    pub wal_sender_timeout_ms: i64,
21    pub current_wal_lsn: Lsn,
22    pub slots_in_use: i64,
23    pub system_identifier: String,
24    pub dbname: String,
25}
26
27#[derive(Debug, Clone, Serialize)]
28pub struct RoleInfo {
29    pub name: String,
30    pub replication: bool,
31    pub superuser: bool,
32}
33
34#[derive(Debug, Clone, Serialize)]
35pub struct PublicationTable {
36    pub table: TableName,
37    pub column_list: Option<Vec<String>>,
38    pub row_filter: Option<String>,
39}
40
41#[derive(Debug, Clone, Serialize)]
42pub struct PublicationInfo {
43    pub name: String,
44    pub all_tables: bool,
45    pub insert: bool,
46    pub update: bool,
47    pub delete: bool,
48    pub truncate: bool,
49    pub via_partition_root: bool,
50    pub tables: Vec<PublicationTable>,
51}
52
53#[derive(Debug, Clone, Serialize)]
54pub struct ColumnInfo {
55    pub attnum: i16,
56    pub name: String,
57    pub type_name: String,
58    pub not_null: bool,
59    pub generated: bool,
60    pub identity: String,
61}
62
63#[derive(Debug, Clone, Serialize)]
64pub struct TableInfo {
65    pub table: TableName,
66    pub oid: u32,
67    pub relkind: String,
68    pub replica_identity: String,
69    pub has_primary_key: bool,
70    pub key_columns: Vec<String>,
71    pub columns: Vec<ColumnInfo>,
72    pub estimated_rows: f64,
73    pub size_bytes: i64,
74    pub can_select: bool,
75}
76
77impl TableInfo {
78    pub fn column_names(&self) -> Vec<String> {
79        self.columns
80            .iter()
81            .filter(|c| !c.generated)
82            .map(|c| c.name.clone())
83            .collect()
84    }
85}
86
87pub async fn server_info(c: &PgClient) -> Result<ServerInfo> {
88    let row = c
89        .query_one(
90            "SELECT current_setting('server_version_num')::int, version(), current_setting('wal_level'), \
91             current_setting('max_replication_slots')::int, current_setting('max_wal_senders')::int, \
92             current_setting('max_slot_wal_keep_size'), current_setting('wal_sender_timeout'), \
93             pg_current_wal_lsn()::text, (SELECT count(*) FROM pg_replication_slots), \
94             (SELECT system_identifier::text FROM pg_control_system()), current_database(), \
95             (SELECT setting::bigint FROM pg_settings WHERE name = 'wal_sender_timeout')",
96            &[],
97        )
98        .await
99        .map_err(|e| from_pg("source server info", e))?;
100    let lsn: String = row.get(7);
101    Ok(ServerInfo {
102        version_num: row.get(0),
103        version: row.get(1),
104        wal_level: row.get(2),
105        max_replication_slots: row.get(3),
106        max_wal_senders: row.get(4),
107        max_slot_wal_keep_size: row.get(5),
108        wal_sender_timeout: row.get(6),
109        current_wal_lsn: lsn
110            .parse()
111            .map_err(|e| Error::internal(format!("pg_current_wal_lsn: {e}")))?,
112        slots_in_use: row.get(8),
113        system_identifier: row.get(9),
114        dbname: row.get(10),
115        wal_sender_timeout_ms: row.get(11),
116    })
117}
118
119pub async fn role_info(c: &PgClient) -> Result<RoleInfo> {
120    let row = c
121        .query_one("SELECT rolname::text, rolreplication, rolsuper FROM pg_roles WHERE rolname = current_user", &[])
122        .await
123        .map_err(|e| from_pg("source role info", e))?;
124    Ok(RoleInfo {
125        name: row.get(0),
126        replication: row.get(1),
127        superuser: row.get(2),
128    })
129}
130
131pub async fn publication_info(c: &PgClient, name: &str) -> Result<Option<PublicationInfo>> {
132    let row = c
133        .query_opt(
134            "SELECT puballtables, pubinsert, pubupdate, pubdelete, pubtruncate, pubviaroot FROM pg_publication WHERE pubname = $1",
135            &[&name],
136        )
137        .await
138        .map_err(|e| from_pg("publication info", e))?;
139    let Some(row) = row else { return Ok(None) };
140    let rows = c
141        .query(
142            "SELECT schemaname::text, tablename::text, attnames::text[], rowfilter, \
143             (SELECT count(*) FROM pg_attribute a JOIN pg_class cl ON cl.oid = a.attrelid JOIN pg_namespace n ON n.oid = cl.relnamespace \
144              WHERE n.nspname = pt.schemaname AND cl.relname = pt.tablename AND a.attnum > 0 AND NOT a.attisdropped AND a.attgenerated = '') \
145             FROM pg_publication_tables pt WHERE pubname = $1 ORDER BY 1, 2",
146            &[&name],
147        )
148        .await
149        .map_err(|e| from_pg("publication tables", e))?;
150    let tables = rows
151        .iter()
152        .map(|r| {
153            let attnames: Option<Vec<String>> = r.get(2);
154            let total: i64 = r.get(4);
155            let column_list = attnames.filter(|a| (a.len() as i64) != total);
156            PublicationTable {
157                table: TableName::new(r.get::<_, String>(0), r.get::<_, String>(1)),
158                column_list,
159                row_filter: r.get(3),
160            }
161        })
162        .collect();
163    Ok(Some(PublicationInfo {
164        name: name.to_owned(),
165        all_tables: row.get(0),
166        insert: row.get(1),
167        update: row.get(2),
168        delete: row.get(3),
169        truncate: row.get(4),
170        via_partition_root: row.get(5),
171        tables,
172    }))
173}
174
175pub async fn table_info(c: &PgClient, t: &TableName) -> Result<Option<TableInfo>> {
176    let row = c
177        .query_opt(
178            "SELECT c.oid, c.relkind::text, c.relreplident::text, \
179             EXISTS (SELECT 1 FROM pg_index i WHERE i.indrelid = c.oid AND i.indisprimary), \
180             c.reltuples::float8, pg_table_size(c.oid), has_table_privilege(c.oid, 'SELECT') \
181             FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace \
182             WHERE n.nspname = $1 AND c.relname = $2",
183            &[&t.schema, &t.name],
184        )
185        .await
186        .map_err(|e| from_pg(&format!("table info {t}"), e))?;
187    let Some(row) = row else { return Ok(None) };
188    let oid: u32 = row.get(0);
189    let relkind: String = row.get(1);
190    let replica_identity: String = row.get(2);
191    let cols = c
192        .query(
193            "SELECT a.attnum, a.attname::text, format_type(a.atttypid, a.atttypmod), a.attnotnull, \
194             a.attgenerated <> '', a.attidentity::text \
195             FROM pg_attribute a WHERE a.attrelid = $1 AND a.attnum > 0 AND NOT a.attisdropped ORDER BY a.attnum",
196            &[&oid],
197        )
198        .await
199        .map_err(|e| from_pg(&format!("columns of {t}"), e))?;
200    let columns = cols
201        .iter()
202        .map(|r| ColumnInfo {
203            attnum: r.get(0),
204            name: r.get(1),
205            type_name: r.get(2),
206            not_null: r.get(3),
207            generated: r.get(4),
208            identity: r.get(5),
209        })
210        .collect();
211    let keys = c
212        .query(
213            "SELECT a.attname::text FROM pg_index i \
214             JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY (i.indkey) \
215             WHERE i.indrelid = $1 AND ((i.indisprimary AND $2 = 'd') OR (i.indisreplident AND $2 = 'i')) \
216             ORDER BY a.attnum",
217            &[&oid, &replica_identity],
218        )
219        .await
220        .map_err(|e| from_pg(&format!("key columns of {t}"), e))?;
221    Ok(Some(TableInfo {
222        table: t.clone(),
223        oid,
224        relkind,
225        replica_identity,
226        has_primary_key: row.get(3),
227        key_columns: keys.iter().map(|r| r.get::<_, String>(0)).collect(),
228        columns,
229        estimated_rows: row.get(4),
230        size_bytes: row.get(5),
231        can_select: row.get(6),
232    }))
233}
234
235pub async fn slot_info(c: &PgClient, slot: &str) -> Result<Option<SlotInfo>> {
236    let row = c
237        .query_opt(
238            "SELECT to_jsonb(s)::text FROM pg_replication_slots s WHERE s.slot_name = $1",
239            &[&slot],
240        )
241        .await
242        .map_err(|e| from_pg("slot info", e))?;
243    match row {
244        Some(r) => Ok(Some(slot_info_from_json(&r.get::<_, String>(0))?)),
245        None => Ok(None),
246    }
247}
248
249pub async fn current_wal_lsn(c: &PgClient) -> Result<Lsn> {
250    let row = c
251        .query_one("SELECT pg_current_wal_lsn()::text", &[])
252        .await
253        .map_err(|e| from_pg("pg_current_wal_lsn", e))?;
254    row.get::<_, String>(0)
255        .parse()
256        .map_err(|e| Error::internal(format!("pg_current_wal_lsn: {e}")))
257}
258
259// ---- preflight ----------------------------------------------------------------------
260
261#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
262#[serde(rename_all = "lowercase")]
263pub enum CheckStatus {
264    Pass,
265    Warn,
266    Fail,
267}
268
269#[derive(Debug, Clone, Serialize)]
270pub struct CheckResult {
271    pub name: String,
272    pub status: CheckStatus,
273    pub detail: String,
274}
275
276#[derive(Debug, Clone, Serialize, Default)]
277pub struct PreflightReport {
278    pub checks: Vec<CheckResult>,
279}
280
281impl PreflightReport {
282    pub fn add(&mut self, name: &str, status: CheckStatus, detail: impl Into<String>) {
283        self.checks.push(CheckResult {
284            name: name.to_owned(),
285            status,
286            detail: detail.into(),
287        });
288    }
289    pub fn pass(&mut self, name: &str, detail: impl Into<String>) {
290        self.add(name, CheckStatus::Pass, detail);
291    }
292    pub fn warn(&mut self, name: &str, detail: impl Into<String>) {
293        self.add(name, CheckStatus::Warn, detail);
294    }
295    pub fn fail(&mut self, name: &str, detail: impl Into<String>) {
296        self.add(name, CheckStatus::Fail, detail);
297    }
298    pub fn failed(&self) -> bool {
299        self.checks.iter().any(|c| c.status == CheckStatus::Fail)
300    }
301    pub fn failures(&self) -> Vec<&CheckResult> {
302        self.checks
303            .iter()
304            .filter(|c| c.status == CheckStatus::Fail)
305            .collect()
306    }
307}
308
309/// Which lifecycle step the preflight is for.
310#[derive(Debug, Clone, Copy, PartialEq, Eq)]
311pub enum PreflightMode {
312    /// `check` / `bootstrap`: the slot must not exist yet.
313    Bootstrap,
314    /// `run`: the slot must exist.
315    Run,
316}
317
318pub struct SourcePreflight {
319    pub server: ServerInfo,
320    pub role: RoleInfo,
321    pub publication: Option<PublicationInfo>,
322    pub tables: Vec<TableInfo>,
323    pub slot: Option<SlotInfo>,
324    pub report: PreflightReport,
325}
326
327/// Inspect the source and evaluate every Beta 1 prerequisite. Never modifies the source.
328pub async fn preflight(
329    c: &PgClient,
330    publication: &str,
331    slot: &str,
332    tables: &[TableName],
333    mode: PreflightMode,
334    feedback_interval: std::time::Duration,
335) -> Result<SourcePreflight> {
336    let mut r = PreflightReport::default();
337    let server = server_info(c).await?;
338    let role = role_info(c).await?;
339
340    // Server
341    if server.version_num < 140000 {
342        r.fail(
343            "server_version",
344            format!("{} is not supported (need 16+)", server.version),
345        );
346    } else if server.version_num < 160000 {
347        r.warn(
348            "server_version",
349            format!(
350                "{} is untested; supported versions are 16, 17, 18",
351                server.version
352            ),
353        );
354    } else {
355        r.pass("server_version", server.version.clone());
356    }
357    if server.wal_level == "logical" {
358        r.pass("wal_level", "logical");
359    } else {
360        r.fail(
361            "wal_level",
362            format!(
363                "is {:?}; must be 'logical' (requires a server restart; PGSync never changes it)",
364                server.wal_level
365            ),
366        );
367    }
368    if server.max_replication_slots <= 0 {
369        r.fail("max_replication_slots", "is 0");
370    } else if server.slots_in_use >= server.max_replication_slots as i64
371        && mode == PreflightMode::Bootstrap
372    {
373        r.fail(
374            "max_replication_slots",
375            format!("all {} slots are in use", server.max_replication_slots),
376        );
377    } else {
378        r.pass(
379            "max_replication_slots",
380            format!(
381                "{} ({} in use)",
382                server.max_replication_slots, server.slots_in_use
383            ),
384        );
385    }
386    if server.max_wal_senders <= 0 {
387        r.fail("max_wal_senders", "is 0");
388    } else {
389        r.pass("max_wal_senders", server.max_wal_senders.to_string());
390    }
391    if server.max_slot_wal_keep_size == "-1" {
392        r.warn(
393            "max_slot_wal_keep_size",
394            "unlimited: a stalled consumer can fill the source disk with retained WAL",
395        );
396    } else {
397        r.pass(
398            "max_slot_wal_keep_size",
399            format!(
400                "{} (slot is invalidated beyond this)",
401                server.max_slot_wal_keep_size
402            ),
403        );
404    }
405    let fb_ms = feedback_interval.as_millis() as i64;
406    if server.wal_sender_timeout_ms > 0 && fb_ms * 2 > server.wal_sender_timeout_ms {
407        r.warn(
408            "wal_sender_timeout",
409            format!(
410                "{} is less than twice resources.feedback_interval ({fb_ms} ms): the walsender may drop \
411                 the connection while the applier is blocked; lower feedback_interval",
412                server.wal_sender_timeout
413            ),
414        );
415    } else {
416        r.pass("wal_sender_timeout", server.wal_sender_timeout.clone());
417    }
418
419    // Role
420    if role.replication || role.superuser {
421        r.pass(
422            "role",
423            format!(
424                "{} has REPLICATION{}",
425                role.name,
426                if role.superuser { " (superuser)" } else { "" }
427            ),
428        );
429    } else {
430        r.fail(
431            "role",
432            format!("{} lacks the REPLICATION attribute", role.name),
433        );
434    }
435    if role.superuser {
436        r.warn(
437            "least_privilege",
438            "connecting as a superuser; prefer a dedicated REPLICATION role with SELECT only",
439        );
440    }
441
442    // Publication
443    let publication = publication_info(c, publication).await?;
444    let configured: BTreeSet<&TableName> = tables.iter().collect();
445    match &publication {
446        None => r.fail("publication", format!("publication {publication:?} does not exist (PGSync does not create it on a source without approval)")),
447        Some(p) => {
448            r.pass("publication", format!("{} exists", p.name));
449            let mut missing = vec![];
450            if !p.insert { missing.push("insert") }
451            if !p.update { missing.push("update") }
452            if !p.delete { missing.push("delete") }
453            if missing.is_empty() {
454                r.pass("publication_operations", format!("insert, update, delete{}", if p.truncate { ", truncate" } else { "" }));
455            } else {
456                r.fail("publication_operations", format!("publication does not publish {}; changes would be silently missing", missing.join(", ")));
457            }
458            if p.all_tables {
459                r.warn("publication_all_tables", "FOR ALL TABLES: new tables appear in the stream and will stop replication until configured");
460            }
461            if p.via_partition_root {
462                r.fail("publish_via_partition_root", "not supported in Beta 1");
463            }
464            let published: BTreeSet<&TableName> = p.tables.iter().map(|t| &t.table).collect();
465            let not_published: Vec<String> = configured.difference(&published).map(|t| t.to_string()).collect();
466            let not_configured: Vec<String> = published.difference(&configured).map(|t| t.to_string()).collect();
467            if not_published.is_empty() && not_configured.is_empty() {
468                r.pass("publication_tables", format!("{} tables match replication.tables", published.len()));
469            } else {
470                if !not_published.is_empty() {
471                    r.fail("publication_tables", format!("configured tables not in publication: {}", not_published.join(", ")));
472                }
473                if !not_configured.is_empty() {
474                    r.fail("publication_tables", format!("publication contains tables not in replication.tables: {} (their changes cannot be applied)", not_configured.join(", ")));
475                }
476            }
477            for t in &p.tables {
478                if t.column_list.is_some() {
479                    r.fail("column_list", format!("{}: publication column lists are not supported in Beta 1", t.table));
480                }
481                if t.row_filter.is_some() {
482                    r.fail("row_filter", format!("{}: publication row filters are not supported in Beta 1", t.table));
483                }
484            }
485        }
486    }
487
488    // Tables
489    let mut infos = Vec::new();
490    for t in tables {
491        match table_info(c, t).await? {
492            None => r.fail("table", format!("{t} does not exist")),
493            Some(info) => {
494                let name = format!("table {t}");
495                match info.relkind.as_str() {
496                    "r" => {}
497                    "p" => r.fail(&name, "partitioned tables are not supported in Beta 1"),
498                    "f" => r.fail(&name, "foreign tables cannot be replicated"),
499                    "v" | "m" => r.fail(&name, "views cannot be replicated"),
500                    k => r.fail(&name, format!("unsupported relkind {k:?}")),
501                }
502                if !info.can_select {
503                    r.fail(
504                        &name,
505                        format!(
506                            "role {} lacks SELECT (needed for the initial snapshot)",
507                            role.name
508                        ),
509                    );
510                }
511                match info.replica_identity.as_str() {
512                    "d" if info.has_primary_key => r.pass(&name, format!("primary key ({})", info.key_columns.join(", "))),
513                    "d" => r.fail(&name, "no primary key and REPLICA IDENTITY DEFAULT: UPDATE/DELETE would fail on the source itself"),
514                    "i" => r.pass(&name, format!("REPLICA IDENTITY USING INDEX ({})", info.key_columns.join(", "))),
515                    "f" => r.warn(&name, "REPLICA IDENTITY FULL: supported, but every UPDATE/DELETE ships the whole old row and matches without an index"),
516                    "n" => r.fail(&name, "REPLICA IDENTITY NOTHING: UPDATE/DELETE cannot be replicated"),
517                    other => r.fail(&name, format!("unknown replica identity {other:?}")),
518                }
519                let generated: Vec<_> = info
520                    .columns
521                    .iter()
522                    .filter(|c| c.generated)
523                    .map(|c| c.name.as_str())
524                    .collect();
525                if !generated.is_empty() {
526                    r.pass(
527                        &name,
528                        format!(
529                            "generated columns are recomputed on the target: {}",
530                            generated.join(", ")
531                        ),
532                    );
533                }
534                infos.push(info);
535            }
536        }
537    }
538
539    // Slot
540    let slot_info = slot_info(c, slot).await?;
541    match (&slot_info, mode) {
542        (None, PreflightMode::Bootstrap) => r.pass("slot", format!("{slot:?} does not exist yet; bootstrap will create it")),
543        (Some(s), PreflightMode::Bootstrap) => r.fail("slot", format!("{slot:?} already exists (plugin {:?}, active {}); choose another name or run `pgsync cleanup`", s.plugin, s.active)),
544        (None, PreflightMode::Run) => r.fail("slot", format!("{slot:?} does not exist; the job needs a new bootstrap")),
545        (Some(s), PreflightMode::Run) => {
546            if s.wal_status.as_deref() == Some("lost") || s.invalidation_reason.is_some() {
547                r.fail("slot", format!("{slot:?} is invalid (wal_status {:?}, reason {:?}); resynchronisation required", s.wal_status, s.invalidation_reason));
548            } else {
549                let retained = s.restart_lsn.map(|l| server.current_wal_lsn.bytes_since(l)).unwrap_or(0);
550                r.pass("slot", format!("{slot:?} exists, active={}, wal_status={:?}, retained_wal_bytes={retained}", s.active, s.wal_status));
551            }
552        }
553    }
554
555    Ok(SourcePreflight {
556        server,
557        role,
558        publication,
559        tables: infos,
560        slot: slot_info,
561        report: r,
562    })
563}
pgsync_postgres/source.rs

pgsync_postgres/
source.rs
