pgsync_core/
config.rs

1//! Job configuration: YAML schema, defaults, and validation.
2//!
3//! Connection URLs are never stored in the file; they are read from the environment
4//! variables named by `url_env` so that secrets stay out of config and logs.
5
6use crate::error::{Error, Result};
7use serde::{Deserialize, Serialize};
8use std::collections::BTreeSet;
9use std::fmt;
10use std::time::Duration;
11
12#[derive(Debug, Clone, Deserialize, Serialize)]
13#[serde(deny_unknown_fields)]
14pub struct Config {
15    pub job: JobConfig,
16    pub source: SourceConfig,
17    pub target: TargetConfig,
18    pub replication: ReplicationConfig,
19    #[serde(default)]
20    pub schema: SchemaConfig,
21    #[serde(default)]
22    pub resources: ResourcesConfig,
23    #[serde(default)]
24    pub safety: SafetyConfig,
25}
26
27#[derive(Debug, Clone, Deserialize, Serialize)]
28#[serde(deny_unknown_fields)]
29pub struct JobConfig {
30    /// Stable identifier; also the primary key of the checkpoint row in the target.
31    pub id: String,
32}
33
34#[derive(Debug, Clone, Deserialize, Serialize)]
35#[serde(deny_unknown_fields)]
36pub struct SourceConfig {
37    /// Environment variable holding the source connection URL.
38    pub url_env: String,
39    /// Existing publication to stream from.
40    pub publication: String,
41    /// Logical replication slot name (created by `bootstrap`, used by `run`).
42    pub slot: String,
43}
44
45#[derive(Debug, Clone, Deserialize, Serialize)]
46#[serde(deny_unknown_fields)]
47pub struct TargetConfig {
48    pub url_env: String,
49}
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
52#[serde(rename_all = "snake_case")]
53pub enum ReplicationMode {
54    /// Initial snapshot followed by continuous CDC (default).
55    SnapshotAndCdc,
56    /// Reserved. Rejected by validation: pgsync cannot prove that an existing target copy is
57    /// consistent with a freshly created slot, so there is no safe way to skip the snapshot.
58    CdcOnly,
59}
60
61#[derive(Debug, Clone, Deserialize, Serialize)]
62#[serde(deny_unknown_fields)]
63pub struct ReplicationConfig {
64    #[serde(default = "default_mode")]
65    pub mode: ReplicationMode,
66    /// Fully qualified `schema.table` names. Must match the publication contents.
67    pub tables: Vec<TableName>,
68}
69
70fn default_mode() -> ReplicationMode {
71    ReplicationMode::SnapshotAndCdc
72}
73
74#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, Default)]
75#[serde(rename_all = "snake_case")]
76pub enum SchemaMode {
77    /// Run `pg_dump --schema-only` for the selected tables and apply it to the target.
78    #[default]
79    PgDump,
80    /// The target schema already exists; PGSync verifies columns and proceeds.
81    Existing,
82}
83
84#[derive(Debug, Clone, Deserialize, Serialize, Default)]
85#[serde(deny_unknown_fields)]
86pub struct SchemaConfig {
87    #[serde(default)]
88    pub mode: SchemaMode,
89    /// Path to the `pg_dump` binary (default: resolved from `PATH`).
90    #[serde(default)]
91    pub pg_dump_path: Option<String>,
92}
93
94#[derive(Debug, Clone, Deserialize, Serialize)]
95#[serde(deny_unknown_fields, default)]
96pub struct ResourcesConfig {
97    pub memory_budget_mb: u64,
98    pub buffer_size_kb: u64,
99    pub max_inflight_bytes: u64,
100    pub channel_capacity: usize,
101    pub max_batch_rows: usize,
102    pub max_batch_bytes: u64,
103    /// Largest single replication message accepted from the server (one row change can be
104    /// up to 1 GB in PostgreSQL). Larger messages stop the job with a `resource` error
105    /// before anything is buffered; nothing is acknowledged. `None` derives a value from
106    /// the remaining memory budget, capped at 32 MiB.
107    pub max_message_bytes: Option<u64>,
108    pub max_parallel_copies: usize,
109    #[serde(with = "duration_text")]
110    pub connection_timeout: Duration,
111    pub retry_backoff: RetryBackoff,
112    #[serde(with = "duration_text")]
113    pub snapshot_timeout: Duration,
114    #[serde(with = "duration_text")]
115    pub feedback_interval: Duration,
116    pub max_replication_lag_bytes: u64,
117}
118
119impl Default for ResourcesConfig {
120    fn default() -> Self {
121        ResourcesConfig {
122            memory_budget_mb: 128,
123            buffer_size_kb: 256,
124            max_inflight_bytes: 16 * 1024 * 1024,
125            channel_capacity: 4096,
126            max_batch_rows: 500,
127            max_batch_bytes: 4 * 1024 * 1024,
128            max_message_bytes: None,
129            max_parallel_copies: 2,
130            connection_timeout: Duration::from_secs(10),
131            retry_backoff: RetryBackoff::default(),
132            snapshot_timeout: Duration::from_secs(6 * 3600),
133            feedback_interval: Duration::from_secs(10),
134            max_replication_lag_bytes: 1024 * 1024 * 1024,
135        }
136    }
137}
138
139#[derive(Debug, Clone, Deserialize, Serialize)]
140#[serde(deny_unknown_fields, default)]
141pub struct RetryBackoff {
142    #[serde(with = "duration_text")]
143    pub initial: Duration,
144    #[serde(with = "duration_text")]
145    pub max: Duration,
146    pub multiplier: f64,
147    /// `0` means retry forever.
148    pub max_attempts: u32,
149}
150
151pub const MAX_MESSAGE_BYTES_CAP: u64 = 32 * 1024 * 1024;
152
153impl ResourcesConfig {
154    /// Memory reserved for channel, batch and COPY buffers.
155    pub fn fixed_budget_bytes(&self) -> u64 {
156        self.max_inflight_bytes
157            + self.max_batch_bytes
158            + (self.max_parallel_copies as u64) * 2 * self.buffer_size_kb * 1024
159    }
160
161    /// Effective message limit: the configured value, or half of what the budget leaves
162    /// after the fixed buffers (a message is held twice while it is applied: decoded and
163    /// serialised as statement parameters), capped at 32 MiB and never below 1 MiB.
164    pub fn effective_max_message_bytes(&self) -> u64 {
165        match self.max_message_bytes {
166            Some(v) => v,
167            None => {
168                let budget = self.memory_budget_mb * 1024 * 1024;
169                let spare = budget.saturating_sub(self.fixed_budget_bytes()) / 2;
170                spare.clamp(1024 * 1024, MAX_MESSAGE_BYTES_CAP)
171            }
172        }
173    }
174}
175
176impl Default for RetryBackoff {
177    fn default() -> Self {
178        RetryBackoff {
179            initial: Duration::from_secs(1),
180            max: Duration::from_secs(30),
181            multiplier: 2.0,
182            max_attempts: 0,
183        }
184    }
185}
186
187#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, Default)]
188#[serde(rename_all = "snake_case")]
189pub enum WalRetentionAction {
190    #[default]
191    Warn,
192    Stop,
193}
194
195#[derive(Debug, Clone, Deserialize, Serialize)]
196#[serde(deny_unknown_fields, default)]
197pub struct SafetyConfig {
198    pub require_tls: bool,
199    /// Accept any server certificate. Only honoured for loopback hosts.
200    pub insecure_skip_tls_verify: bool,
201    /// PEM file with CA certificate(s) to trust in addition to the system store.
202    pub tls_ca_file: Option<String>,
203    pub auto_create_publication: bool,
204    pub auto_drop_slot: bool,
205    pub wal_retention_alarm_bytes: u64,
206    pub wal_retention_action: WalRetentionAction,
207    #[serde(with = "duration_text")]
208    pub monitor_interval: Duration,
209    /// How often the running job writes its runtime row (`pgsync.job_runtime`) to the
210    /// target, for `pgsync status` and external monitoring.
211    #[serde(with = "duration_text")]
212    pub heartbeat_interval: Duration,
213}
214
215impl Default for SafetyConfig {
216    fn default() -> Self {
217        SafetyConfig {
218            require_tls: true,
219            insecure_skip_tls_verify: false,
220            tls_ca_file: None,
221            auto_create_publication: false,
222            auto_drop_slot: false,
223            wal_retention_alarm_bytes: 1024 * 1024 * 1024,
224            wal_retention_action: WalRetentionAction::Warn,
225            monitor_interval: Duration::from_secs(30),
226            heartbeat_interval: Duration::from_secs(10),
227        }
228    }
229}
230
231/// A `schema.table` pair. Serialised as `schema.table`; quoted identifiers are not
232/// supported in Beta 1 (names must be plain lower-case identifiers).
233#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
234pub struct TableName {
235    pub schema: String,
236    pub name: String,
237}
238
239impl TableName {
240    pub fn new(schema: impl Into<String>, name: impl Into<String>) -> Self {
241        TableName {
242            schema: schema.into(),
243            name: name.into(),
244        }
245    }
246
247    pub fn parse(s: &str) -> Result<Self> {
248        let (schema, name) = s
249            .split_once('.')
250            .ok_or_else(|| Error::config(format!("table {s:?} must be written as schema.table")))?;
251        for part in [schema, name] {
252            if !is_plain_identifier(part) {
253                return Err(Error::config(format!(
254                    "table {s:?}: {part:?} is not a plain identifier (lower-case letters, digits, underscore)"
255                )));
256            }
257        }
258        Ok(TableName::new(schema, name))
259    }
260
261    /// Double-quoted, escaped `"schema"."table"` for use in SQL.
262    pub fn quoted(&self) -> String {
263        format!("{}.{}", quote_ident(&self.schema), quote_ident(&self.name))
264    }
265}
266
267impl fmt::Display for TableName {
268    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
269        write!(f, "{}.{}", self.schema, self.name)
270    }
271}
272
273impl Serialize for TableName {
274    fn serialize<S: serde::Serializer>(&self, s: S) -> std::result::Result<S::Ok, S::Error> {
275        s.collect_str(self)
276    }
277}
278
279impl<'de> Deserialize<'de> for TableName {
280    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> std::result::Result<Self, D::Error> {
281        let s = String::deserialize(d)?;
282        TableName::parse(&s).map_err(|e| serde::de::Error::custom(e.message().to_owned()))
283    }
284}
285
286pub fn is_plain_identifier(s: &str) -> bool {
287    let mut chars = s.chars();
288    match chars.next() {
289        Some(c) if c.is_ascii_lowercase() || c == '_' => {}
290        _ => return false,
291    }
292    chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_') && s.len() <= 63
293}
294
295/// Quote an SQL identifier, doubling embedded quotes.
296pub fn quote_ident(s: &str) -> String {
297    let mut out = String::with_capacity(s.len() + 2);
298    out.push('"');
299    for c in s.chars() {
300        if c == '"' {
301            out.push('"');
302        }
303        out.push(c);
304    }
305    out.push('"');
306    out
307}
308
309/// Quote an SQL string literal (single quotes doubled). Used only for replication
310/// commands, which do not support parameters.
311pub fn quote_literal(s: &str) -> String {
312    let mut out = String::with_capacity(s.len() + 2);
313    out.push('\'');
314    for c in s.chars() {
315        if c == '\'' {
316            out.push('\'');
317        }
318        out.push(c);
319    }
320    out.push('\'');
321    out
322}
323
324impl Config {
325    pub fn from_yaml(text: &str) -> Result<Config> {
326        let cfg: Config = serde_yaml_ng::from_str(text)
327            .map_err(|e| Error::config(format!("invalid configuration: {e}")))?;
328        cfg.validate()?;
329        Ok(cfg)
330    }
331
332    pub fn validate(&self) -> Result<()> {
333        let bad = |m: String| Err(Error::config(m));
334        if self.job.id.is_empty() || self.job.id.len() > 64 {
335            return bad("job.id must be 1-64 characters".into());
336        }
337        if !self
338            .job
339            .id
340            .chars()
341            .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
342        {
343            return bad("job.id may only contain letters, digits, '-' and '_'".into());
344        }
345        for (field, value) in [
346            ("source.url_env", &self.source.url_env),
347            ("target.url_env", &self.target.url_env),
348        ] {
349            if value.is_empty() {
350                return bad(format!("{field} must name an environment variable"));
351            }
352        }
353        if !is_plain_identifier(&self.source.publication) {
354            return bad(format!(
355                "source.publication {:?} must be a plain identifier",
356                self.source.publication
357            ));
358        }
359        if !is_slot_name(&self.source.slot) {
360            return bad(format!(
361                "source.slot {:?} must contain only lower-case letters, digits and underscores (max 63)",
362                self.source.slot
363            ));
364        }
365        if self.replication.tables.is_empty() {
366            return bad("replication.tables must list at least one table".into());
367        }
368        if self.replication.mode == ReplicationMode::CdcOnly {
369            return bad(
370                "replication.mode: cdc_only is not supported in this version: pgsync cannot verify \
371                 that the target already holds a copy consistent with a new slot's snapshot, and \
372                 streaming into an unverified copy would diverge silently. Use snapshot_and_cdc."
373                    .into(),
374            );
375        }
376        let mut seen = BTreeSet::new();
377        for t in &self.replication.tables {
378            if !seen.insert(t) {
379                return bad(format!("replication.tables lists {t} more than once"));
380            }
381        }
382        let r = &self.resources;
383        if r.memory_budget_mb < 16 {
384            return bad("resources.memory_budget_mb must be at least 16".into());
385        }
386        if r.buffer_size_kb == 0 || r.buffer_size_kb > 64 * 1024 {
387            return bad("resources.buffer_size_kb must be between 1 and 65536".into());
388        }
389        if r.max_inflight_bytes < 64 * 1024 {
390            return bad("resources.max_inflight_bytes must be at least 65536".into());
391        }
392        if r.channel_capacity == 0 {
393            return bad("resources.channel_capacity must be positive".into());
394        }
395        if r.max_batch_rows == 0 || r.max_batch_bytes == 0 {
396            return bad("resources.max_batch_rows and max_batch_bytes must be positive".into());
397        }
398        if r.max_parallel_copies == 0 || r.max_parallel_copies > 64 {
399            return bad("resources.max_parallel_copies must be between 1 and 64".into());
400        }
401        let budget = r.memory_budget_mb * 1024 * 1024;
402        let needed = r.fixed_budget_bytes();
403        if needed > budget {
404            return bad(format!(
405                "resources.memory_budget_mb ({} MiB) is smaller than max_inflight_bytes + max_batch_bytes + copy buffers ({} bytes)",
406                r.memory_budget_mb, needed
407            ));
408        }
409        if let Some(m) = r.max_message_bytes {
410            if m < 64 * 1024 {
411                return bad("resources.max_message_bytes must be at least 65536".into());
412            }
413            if needed + 2 * m > budget {
414                return bad(format!(
415                    "resources.memory_budget_mb ({} MiB) cannot hold the fixed buffers ({} bytes) plus two copies of a max_message_bytes message ({} bytes)",
416                    r.memory_budget_mb, needed, m
417                ));
418            }
419        }
420        if r.connection_timeout.is_zero() {
421            return bad("resources.connection_timeout must be positive".into());
422        }
423        if r.retry_backoff.initial.is_zero() || r.retry_backoff.max < r.retry_backoff.initial {
424            return bad("resources.retry_backoff: initial must be positive and <= max".into());
425        }
426        if !(r.retry_backoff.multiplier >= 1.0 && r.retry_backoff.multiplier <= 10.0) {
427            return bad("resources.retry_backoff.multiplier must be between 1 and 10".into());
428        }
429        if r.snapshot_timeout.is_zero() {
430            return bad("resources.snapshot_timeout must be positive".into());
431        }
432        if r.feedback_interval.is_zero() || r.feedback_interval > Duration::from_secs(60) {
433            return bad("resources.feedback_interval must be between 1ms and 60s".into());
434        }
435        if r.max_replication_lag_bytes == 0 {
436            return bad("resources.max_replication_lag_bytes must be positive".into());
437        }
438        let s = &self.safety;
439        if s.wal_retention_alarm_bytes == 0 {
440            return bad("safety.wal_retention_alarm_bytes must be positive".into());
441        }
442        if s.monitor_interval.is_zero() {
443            return bad("safety.monitor_interval must be positive".into());
444        }
445        if s.heartbeat_interval.is_zero() || s.heartbeat_interval > Duration::from_secs(300) {
446            return bad("safety.heartbeat_interval must be between 1ms and 5m".into());
447        }
448        Ok(())
449    }
450}
451
452pub fn is_slot_name(s: &str) -> bool {
453    !s.is_empty()
454        && s.len() <= 63
455        && s.chars()
456            .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_')
457}
458
459/// Human durations such as `500ms`, `10s`, `5m`, `2h`.
460pub fn parse_duration(s: &str) -> std::result::Result<Duration, String> {
461    let s = s.trim();
462    let split = s.find(|c: char| !c.is_ascii_digit()).unwrap_or(s.len());
463    let (num, unit) = s.split_at(split);
464    let n: u64 = num.parse().map_err(|_| format!("invalid duration {s:?}"))?;
465    let d = match unit.trim() {
466        "ms" => Duration::from_millis(n),
467        "s" | "" => Duration::from_secs(n),
468        "m" => Duration::from_secs(n * 60),
469        "h" => Duration::from_secs(n * 3600),
470        _ => return Err(format!("invalid duration unit in {s:?} (use ms, s, m, h)")),
471    };
472    Ok(d)
473}
474
475pub fn format_duration(d: Duration) -> String {
476    let ms = d.as_millis();
477    if ms % 3_600_000 == 0 && ms > 0 {
478        format!("{}h", ms / 3_600_000)
479    } else if ms % 60_000 == 0 && ms > 0 {
480        format!("{}m", ms / 60_000)
481    } else if ms % 1000 == 0 {
482        format!("{}s", ms / 1000)
483    } else {
484        format!("{ms}ms")
485    }
486}
487
488mod duration_text {
489    use super::*;
490    use serde::{Deserializer, Serializer};
491
492    pub fn serialize<S: Serializer>(d: &Duration, s: S) -> std::result::Result<S::Ok, S::Error> {
493        s.serialize_str(&format_duration(*d))
494    }
495
496    pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> std::result::Result<Duration, D::Error> {
497        let s = String::deserialize(d)?;
498        parse_duration(&s).map_err(serde::de::Error::custom)
499    }
500}
501
502#[cfg(test)]
503mod tests {
504    use super::*;
505
506    const MINIMAL: &str = r#"
507job:
508  id: demo
509source:
510  url_env: SRC
511  publication: pub1
512  slot: slot_1
513target:
514  url_env: DST
515replication:
516  tables: [public.users, public.orders]
517"#;
518
519    #[test]
520    fn minimal_config_gets_defaults() {
521        let c = Config::from_yaml(MINIMAL).unwrap();
522        assert_eq!(c.replication.mode, ReplicationMode::SnapshotAndCdc);
523        assert_eq!(c.resources.memory_budget_mb, 128);
524        assert_eq!(c.resources.connection_timeout, Duration::from_secs(10));
525        assert!(c.safety.require_tls);
526        assert_eq!(c.schema.mode, SchemaMode::PgDump);
527        assert_eq!(c.replication.tables[0], TableName::new("public", "users"));
528    }
529
530    #[test]
531    fn spec_example_parses() {
532        let yaml = r#"
533job:
534  id: demo-replication
535source:
536  url_env: PGSYNC_SOURCE_URL
537  publication: pgsync_demo
538  slot: pgsync_demo_slot
539target:
540  url_env: PGSYNC_TARGET_URL
541replication:
542  mode: snapshot_and_cdc
543  tables:
544    - public.users
545    - public.orders
546resources:
547  memory_budget_mb: 128
548  buffer_size_kb: 256
549  max_inflight_bytes: 16777216
550  max_parallel_copies: 2
551  connection_timeout: 5s
552  retry_backoff:
553    initial: 500ms
554    max: 1m
555  snapshot_timeout: 2h
556safety:
557  require_tls: true
558  auto_create_publication: false
559  auto_drop_slot: false
560"#;
561        let c = Config::from_yaml(yaml).unwrap();
562        assert_eq!(
563            c.resources.retry_backoff.initial,
564            Duration::from_millis(500)
565        );
566        assert_eq!(c.resources.retry_backoff.max, Duration::from_secs(60));
567        assert_eq!(c.resources.snapshot_timeout, Duration::from_secs(7200));
568        assert_eq!(c.resources.connection_timeout, Duration::from_secs(5));
569    }
570
571    #[test]
572    fn rejects_unknown_fields_and_bad_values() {
573        let cases = [
574            (
575                MINIMAL.replace("tables: [public.users, public.orders]", "tables: []"),
576                "at least one",
577            ),
578            (
579                MINIMAL.replace("public.orders", "public.users"),
580                "more than once",
581            ),
582            (MINIMAL.replace("slot_1", "Slot-1"), "source.slot"),
583            (
584                format!("{MINIMAL}\n")
585                    .replace("replication:\n", "replication:\n  mode: cdc_only\n"),
586                "cdc_only is not supported",
587            ),
588            (MINIMAL.replace("id: demo", "id: 'demo job'"), "job.id"),
589            (
590                format!("{MINIMAL}\nresources:\n  memory_budget_mb: 4\n"),
591                "memory_budget_mb",
592            ),
593            (
594                format!("{MINIMAL}\nresources:\n  max_parallel_copies: 0\n"),
595                "max_parallel_copies",
596            ),
597            (
598                format!("{MINIMAL}\nresources:\n  connection_timeout: 10x\n"),
599                "duration",
600            ),
601            (
602                format!("{MINIMAL}\nresources:\n  typo: 1\n"),
603                "unknown field",
604            ),
605            (
606                format!(
607                    "{MINIMAL}\nresources:\n  memory_budget_mb: 16\n  max_inflight_bytes: 20000000\n"
608                ),
609                "smaller than",
610            ),
611        ];
612        for (yaml, expect) in cases {
613            let err = Config::from_yaml(&yaml).expect_err(&format!("should reject: {expect}"));
614            assert!(err.message().contains(expect), "{expect}: got {err}");
615        }
616    }
617
618    #[test]
619    fn message_limit_is_derived_from_the_budget_or_validated() {
620        let c = Config::from_yaml(MINIMAL).unwrap();
621        // 128 MiB - (16 + 4 + 1) MiB = 107 MiB, half is 53.5 MiB, capped at 32 MiB.
622        assert_eq!(c.resources.effective_max_message_bytes(), 32 * 1024 * 1024);
623        let c =
624            Config::from_yaml(&format!("{MINIMAL}\nresources:\n  memory_budget_mb: 24\n")).unwrap();
625        assert_eq!(c.resources.effective_max_message_bytes(), 1536 * 1024);
626        let c = Config::from_yaml(&format!(
627            "{MINIMAL}\nresources:\n  max_message_bytes: 8388608\n"
628        ))
629        .unwrap();
630        assert_eq!(c.resources.effective_max_message_bytes(), 8 * 1024 * 1024);
631        let err = Config::from_yaml(&format!(
632            "{MINIMAL}\nresources:\n  max_message_bytes: 67108864\n"
633        ))
634        .unwrap_err();
635        assert!(err.message().contains("two copies"), "{err}");
636    }
637
638    #[test]
639    fn table_name_parsing_and_quoting() {
640        assert!(TableName::parse("users").is_err());
641        assert!(TableName::parse("Public.users").is_err());
642        assert!(TableName::parse("public.users;drop").is_err());
643        let t = TableName::parse("app_1.orders_2").unwrap();
644        assert_eq!(t.quoted(), "\"app_1\".\"orders_2\"");
645        assert_eq!(quote_ident("we\"ird"), "\"we\"\"ird\"");
646        assert_eq!(quote_literal("it's"), "'it''s'");
647    }
648
649    #[test]
650    fn durations_round_trip() {
651        for (s, d) in [
652            ("500ms", Duration::from_millis(500)),
653            ("10s", Duration::from_secs(10)),
654            ("5m", Duration::from_secs(300)),
655            ("2h", Duration::from_secs(7200)),
656            ("7", Duration::from_secs(7)),
657        ] {
658            assert_eq!(parse_duration(s).unwrap(), d, "{s}");
659        }
660        assert_eq!(format_duration(Duration::from_secs(300)), "5m");
661        assert_eq!(format_duration(Duration::from_millis(1500)), "1500ms");
662        assert!(parse_duration("abc").is_err());
663        assert!(parse_duration("10d").is_err());
664    }
665}
pgsync_core/config.rs

pgsync_core/
config.rs
