feat: PG-Sync V2 — ops-based outbox, WAL pipeline, trigger generator

Cargo.toml

@@ -15,9 +15,10 @@ default = []
 server = ["dep:axum", "dep:tower-http", "dep:tokio", "dep:tokio-util", "dep:prometheus"]
 loadtest = ["ureq"]
 replay = ["ureq"]
-pg-sync = ["dep:sqlx", "dep:clap", "dep:reqwest", "dep:chrono", "dep:tokio", "dep:axum", "dep:tower-http", "dep:futures-core", "dep:futures-util", "dep:bytes"]
+pg-sync = ["dep:sqlx", "dep:clap", "dep:reqwest", "dep:chrono", "dep:tokio", "dep:axum", "dep:tower-http", "dep:futures-core", "dep:futures-util", "dep:bytes", "dep:serde_yaml"]
 simd = ["roaring/simd"]
 heap-prof = ["dep:tikv-jemallocator", "dep:tikv-jemalloc-ctl"]
+serde_yaml = ["dep:serde_yaml"]
 
 [dependencies]
 # Bitmap indexes
@@ -85,6 +86,7 @@ thiserror = "2"
 # Logging
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+serde_yaml = { version = "0.9.34", optional = true }
 
 [dev-dependencies]
 # Property-based testing

src/lib.rs

@@ -1,6 +1,10 @@
 pub mod bitmap_fs;
 pub mod bound_store;
 pub mod bucket_diff_log;
+#[cfg(feature = "pg-sync")]
+pub mod ops_processor;
+#[cfg(feature = "pg-sync")]
+pub mod ops_wal;
 pub mod cache;
 pub mod capture;
 pub mod concurrency;

src/metrics.rs

@@ -116,6 +116,13 @@ pub struct Metrics {
     pub pgsync_cycle_seconds: HistogramVec,
     pub pgsync_rows_fetched_total: IntCounterVec,
     pub pgsync_cursor_position: IntGaugeVec,
+
+    // V2 sync metrics (unified namespace with source label)
+    pub sync_cursor_position: IntGaugeVec,
+    pub sync_max_id: IntGaugeVec,
+    pub sync_lag_rows: IntGaugeVec,
+    pub sync_ops_total: IntCounterVec,
+    pub sync_wal_bytes: IntGaugeVec,
 }
 
 impl Metrics {
@@ -570,6 +577,28 @@ impl Metrics {
         )
         .unwrap();
 
+        // V2 sync metrics (unified namespace)
+        let sync_cursor_position = IntGaugeVec::new(
+            Opts::new("bitdex_sync_cursor_position", "Current sync cursor position"),
+            &["source"],
+        ).unwrap();
+        let sync_max_id = IntGaugeVec::new(
+            Opts::new("bitdex_sync_max_id", "Max ops table ID (for lag calculation)"),
+            &["source"],
+        ).unwrap();
+        let sync_lag_rows = IntGaugeVec::new(
+            Opts::new("bitdex_sync_lag_rows", "Number of ops rows behind"),
+            &["source"],
+        ).unwrap();
+        let sync_ops_total = IntCounterVec::new(
+            Opts::new("bitdex_sync_ops_total", "Total ops received from sync sources"),
+            &["source"],
+        ).unwrap();
+        let sync_wal_bytes = IntGaugeVec::new(
+            Opts::new("bitdex_sync_wal_bytes", "Current WAL file size in bytes"),
+            &["source"],
+        ).unwrap();
+
         // Register all metrics
         registry.register(Box::new(alive_documents.clone())).unwrap();
         registry.register(Box::new(slot_high_water.clone())).unwrap();
@@ -671,6 +700,11 @@ impl Metrics {
         registry.register(Box::new(pgsync_cycle_seconds.clone())).unwrap();
         registry.register(Box::new(pgsync_rows_fetched_total.clone())).unwrap();
         registry.register(Box::new(pgsync_cursor_position.clone())).unwrap();
+        registry.register(Box::new(sync_cursor_position.clone())).unwrap();
+        registry.register(Box::new(sync_max_id.clone())).unwrap();
+        registry.register(Box::new(sync_lag_rows.clone())).unwrap();
+        registry.register(Box::new(sync_ops_total.clone())).unwrap();
+        registry.register(Box::new(sync_wal_bytes.clone())).unwrap();
 
         Self {
             registry,
@@ -746,6 +780,11 @@ impl Metrics {
             pgsync_cycle_seconds,
             pgsync_rows_fetched_total,
             pgsync_cursor_position,
+            sync_cursor_position,
+            sync_max_id,
+            sync_lag_rows,
+            sync_ops_total,
+            sync_wal_bytes,
         }
     }