pgsync_postgres/replication/
slot.rs

1//! Replication commands and slot inspection over a replication connection.
2
3use super::conn::{ReplicationConnection, SimpleResult};
4use pgsync_core::config::{quote_ident, quote_literal};
5use pgsync_core::{Error, Lsn, Result};
6
7#[derive(Debug, Clone)]
8pub struct SystemInfo {
9    pub system_id: String,
10    pub timeline: u64,
11    pub xlogpos: Lsn,
12    pub dbname: Option<String>,
13}
14
15pub async fn identify_system(conn: &mut ReplicationConnection) -> Result<SystemInfo> {
16    let r = conn.simple_query("IDENTIFY_SYSTEM").await?;
17    let get = |c: &str| r.get(0, c).map(str::to_owned);
18    Ok(SystemInfo {
19        system_id: get("systemid")
20            .ok_or_else(|| Error::internal("IDENTIFY_SYSTEM: missing systemid"))?,
21        timeline: get("timeline").and_then(|t| t.parse().ok()).unwrap_or(0),
22        xlogpos: get("xlogpos")
23            .ok_or_else(|| Error::internal("IDENTIFY_SYSTEM: missing xlogpos"))?
24            .parse()
25            .map_err(|e| Error::internal(format!("IDENTIFY_SYSTEM: {e}")))?,
26        dbname: get("dbname"),
27    })
28}
29
30#[derive(Debug, Clone)]
31pub struct CreatedSlot {
32    pub slot_name: String,
33    pub consistent_point: Lsn,
34    pub snapshot_name: String,
35}
36
37/// `CREATE_REPLICATION_SLOT ... LOGICAL pgoutput EXPORT_SNAPSHOT`.
38///
39/// The exported snapshot stays valid only until the next command on `conn`.
40pub async fn create_slot_with_snapshot(
41    conn: &mut ReplicationConnection,
42    slot: &str,
43) -> Result<CreatedSlot> {
44    let cmd = format!(
45        "CREATE_REPLICATION_SLOT {} LOGICAL pgoutput EXPORT_SNAPSHOT",
46        quote_ident(slot)
47    );
48    let r = conn.simple_query(&cmd).await?;
49    let consistent_point: Lsn = r
50        .get(0, "consistent_point")
51        .ok_or_else(|| Error::internal("CREATE_REPLICATION_SLOT: missing consistent_point"))?
52        .parse()
53        .map_err(|e| Error::internal(format!("CREATE_REPLICATION_SLOT: {e}")))?;
54    let snapshot_name = r
55        .get(0, "snapshot_name")
56        .ok_or_else(|| Error::internal("CREATE_REPLICATION_SLOT: missing snapshot_name"))?
57        .to_owned();
58    Ok(CreatedSlot {
59        slot_name: slot.to_owned(),
60        consistent_point,
61        snapshot_name,
62    })
63}
64
65pub async fn drop_slot(conn: &mut ReplicationConnection, slot: &str) -> Result<()> {
66    conn.simple_query(&format!("DROP_REPLICATION_SLOT {} WAIT", quote_ident(slot)))
67        .await?;
68    Ok(())
69}
70
71/// Version-independent view of `pg_replication_slots` for one slot.
72#[derive(Debug, Clone, Default, serde::Serialize)]
73pub struct SlotInfo {
74    pub slot_name: String,
75    pub plugin: Option<String>,
76    pub slot_type: Option<String>,
77    pub database: Option<String>,
78    pub active: bool,
79    pub active_pid: Option<i64>,
80    pub restart_lsn: Option<Lsn>,
81    pub confirmed_flush_lsn: Option<Lsn>,
82    pub wal_status: Option<String>,
83    pub safe_wal_size: Option<i64>,
84    pub invalidation_reason: Option<String>,
85    pub inactive_since: Option<String>,
86}
87
88pub const SLOT_QUERY: &str =
89    "SELECT to_jsonb(s)::text AS j FROM pg_catalog.pg_replication_slots s WHERE s.slot_name = ";
90
91pub fn slot_info_from_json(json: &str) -> Result<SlotInfo> {
92    let v: serde_json::Value = serde_json::from_str(json)
93        .map_err(|e| Error::internal(format!("pg_replication_slots: {e}")))?;
94    let s = |k: &str| v.get(k).and_then(|x| x.as_str()).map(str::to_owned);
95    let lsn = |k: &str| -> Result<Option<Lsn>> {
96        match s(k) {
97            Some(t) => Ok(Some(
98                t.parse()
99                    .map_err(|e| Error::internal(format!("{k}: {e}")))?,
100            )),
101            None => Ok(None),
102        }
103    };
104    Ok(SlotInfo {
105        slot_name: s("slot_name").unwrap_or_default(),
106        plugin: s("plugin"),
107        slot_type: s("slot_type"),
108        database: s("database"),
109        active: v.get("active").and_then(|x| x.as_bool()).unwrap_or(false),
110        active_pid: v.get("active_pid").and_then(|x| x.as_i64()),
111        restart_lsn: lsn("restart_lsn")?,
112        confirmed_flush_lsn: lsn("confirmed_flush_lsn")?,
113        wal_status: s("wal_status"),
114        safe_wal_size: v.get("safe_wal_size").and_then(|x| x.as_i64()),
115        invalidation_reason: s("invalidation_reason"),
116        inactive_since: s("inactive_since"),
117    })
118}
119
120pub async fn read_slot(conn: &mut ReplicationConnection, slot: &str) -> Result<Option<SlotInfo>> {
121    let r: SimpleResult = conn
122        .simple_query(&format!("{SLOT_QUERY}{}", quote_literal(slot)))
123        .await?;
124    match r.get(0, "j") {
125        Some(json) => Ok(Some(slot_info_from_json(json)?)),
126        None => Ok(None),
127    }
128}
129
130/// Build the `START_REPLICATION` command for pgoutput protocol version 1.
131pub fn start_replication_command(slot: &str, from: Lsn, publication: &str) -> String {
132    format!(
133        "START_REPLICATION SLOT {} LOGICAL {} (proto_version '1', publication_names {})",
134        quote_ident(slot),
135        from,
136        quote_literal(publication)
137    )
138}
139
140#[cfg(test)]
141mod tests {
142    use super::*;
143
144    #[test]
145    fn parses_slot_json_with_optional_columns() {
146        let json = r#"{"slot_name":"s1","plugin":"pgoutput","slot_type":"logical","database":"app","active":true,"active_pid":123,"restart_lsn":"0/16B3748","confirmed_flush_lsn":"0/16B3780","wal_status":"reserved","safe_wal_size":null,"two_phase":false}"#;
147        let s = slot_info_from_json(json).unwrap();
148        assert_eq!(s.plugin.as_deref(), Some("pgoutput"));
149        assert!(s.active);
150        assert_eq!(s.restart_lsn, Some(Lsn(0x16B3748)));
151        assert_eq!(s.confirmed_flush_lsn, Some(Lsn(0x16B3780)));
152        assert_eq!(s.invalidation_reason, None);
153        let pg17 = r#"{"slot_name":"s1","active":false,"invalidation_reason":"wal_removed","wal_status":"lost","restart_lsn":null}"#;
154        let s = slot_info_from_json(pg17).unwrap();
155        assert_eq!(s.invalidation_reason.as_deref(), Some("wal_removed"));
156        assert_eq!(s.restart_lsn, None);
157    }
158
159    #[test]
160    fn start_command_quotes_safely() {
161        let c = start_replication_command("my_slot", Lsn(0x16B3748), "pub1");
162        assert_eq!(
163            c,
164            "START_REPLICATION SLOT \"my_slot\" LOGICAL 0/16B3748 (proto_version '1', publication_names 'pub1')"
165        );
166    }
167}
pgsync_postgres/replication/slot.rs

pgsync_postgres/replication/
slot.rs
