Merge pull request #222 from emanzx/fix/209-sync-listener-config

fix(sync): make sync listener address configurable and bind failure boot-fatal

Author: farhan-syah

docs/getting-started.md

@@ -223,6 +223,7 @@ log_format = "text"               # "text" or "json"
 native = 6433                     # Always-on protocols have defaults
 pgwire = 6432
 http = 6480
+sync = 9090                       # Always on
 resp = 6381                       # Optional: set to enable, omit to disable
 ilp = 8086                        # Optional: set to enable, omit to disable
 
@@ -236,6 +237,10 @@ resp = true
 ilp = false                       # Example: disable TLS for ILP ingest
 ```
 
+Every listener binds during startup, before the server accepts any
+connection. If a configured port is already in use, startup fails with that
+port named — the server never comes up missing a protocol.
+
 **Server settings:**
 
 | Config field       | Environment variable      | Default                                               |
@@ -244,6 +249,7 @@ ilp = false                       # Example: disable TLS for ILP ingest
 | `ports.native`     | `NODEDB_PORT_NATIVE`      | `6433`                                                |
 | `ports.pgwire`     | `NODEDB_PORT_PGWIRE`      | `6432`                                                |
 | `ports.http`       | `NODEDB_PORT_HTTP`        | `6480`                                                |
+| `ports.sync`       | `NODEDB_PORT_SYNC`        | `9090`                                                |
 | `ports.resp`       | `NODEDB_PORT_RESP`        | disabled                                              |
 | `ports.ilp`        | `NODEDB_PORT_ILP`         | disabled                                              |
 | `data_dir`         | `NODEDB_DATA_DIR`         | `~/.nodedb/data` (binary), `/var/lib/nodedb` (Docker) |

docs/protocols.md

@@ -223,7 +223,7 @@ SELECT * FROM cpu WHERE ts > now() - INTERVAL '1 hour';
 
 CRDT sync protocol for NodeDB-Lite clients (mobile, WASM, desktop). Bidirectional delta exchange over WebSocket.
 
-**Port:** 9090 (default)
+**Port:** 9090 (default). Configurable via `[server.ports] sync` or `NODEDB_PORT_SYNC`.
 
 **Flow:**
 
@@ -256,6 +256,7 @@ host = "127.0.0.1"
 pgwire = 6432       # Always on
 native = 6433       # Always on
 http = 6480         # Always on
+sync = 9090         # Always on
 resp = 6381         # Set to enable (omit to disable)
 ilp = 8086          # Set to enable (omit to disable)
 
@@ -269,7 +270,7 @@ resp = true
 ilp = false         # Example: disable TLS for ILP ingest
 ```
 
-Environment variables override config: `NODEDB_PORT_PGWIRE`, `NODEDB_PORT_NATIVE`, `NODEDB_PORT_HTTP`, `NODEDB_PORT_RESP`, `NODEDB_PORT_ILP`.
+Environment variables override config: `NODEDB_PORT_PGWIRE`, `NODEDB_PORT_NATIVE`, `NODEDB_PORT_HTTP`, `NODEDB_PORT_SYNC`, `NODEDB_PORT_RESP`, `NODEDB_PORT_ILP`.
 
 ## Native Protocol Opcodes (SDK Reference)
 

docs/wasm.md

@@ -173,7 +173,7 @@ const db = new NodeDbLite();
 
 // Configure sync to Origin
 await db.sync_config({
-  server_url: "ws://localhost:6433",
+  server_url: "ws://localhost:9090",
   auth_token: "your-token",
   auto_sync: true,
   sync_interval_ms: 5000,

nodedb/src/bootstrap/listeners.rs

@@ -4,6 +4,8 @@
 
 use std::sync::Arc;
 
+use anyhow::Context;
+use tokio::net::TcpListener;
 use tracing::info;
 
 use crate::ServerConfig;
@@ -16,9 +18,14 @@ use crate::control::shutdown::ShutdownBus;
 use crate::control::startup::StartupGate;
 use crate::control::state::SharedState;
 
-/// The three protocol listeners passed to [`spawn_protocol_listeners`].
+/// The pre-bound protocol listeners passed to [`spawn_protocol_listeners`].
+///
+/// Every socket here is already bound by [`bind_listeners`], so spawning
+/// cannot fail on a port conflict.
 pub struct ProtocolListeners {
     pub pg_listener: PgListener,
+    pub http_listener: TcpListener,
+    pub sync_listener: TcpListener,
     pub ilp_listener: Option<IlpListener>,
     pub resp_listener: Option<RespListener>,
 }
@@ -34,6 +41,10 @@ pub struct ListenerInfra {
 ///
 /// The native listener is not spawned here — it is run on the main task
 /// by the caller after this returns.
+///
+/// Infallible by construction: every socket was bound by [`bind_listeners`]
+/// before this point, so a port conflict has already aborted boot while
+/// nothing was exposed. Nothing here may silently swallow a bind failure.
 pub async fn spawn_protocol_listeners(
     listeners: ProtocolListeners,
     shared: Arc<SharedState>,
@@ -44,6 +55,8 @@ pub async fn spawn_protocol_listeners(
 ) {
     let ProtocolListeners {
         pg_listener,
+        http_listener,
+        sync_listener,
         ilp_listener,
         resp_listener,
     } = listeners;
@@ -84,10 +97,9 @@ pub async fn spawn_protocol_listeners(
         }
     });
 
-    // HTTP API server.
+    // HTTP API server (on the socket bound by `bind_listeners`).
     let shared_http = Arc::clone(&shared);
     let http_auth_mode = config.auth.mode.clone();
-    let http_listen = config.http_addr();
     let http_tls = if http_tls_enabled {
         config.server.tls.clone()
     } else {
@@ -96,7 +108,7 @@ pub async fn spawn_protocol_listeners(
     let bus_http = shutdown_bus.clone();
     tokio::spawn(async move {
         if let Err(e) = crate::control::server::http::server::run(
-            http_listen,
+            http_listener,
             shared_http,
             http_auth_mode,
             http_tls.as_ref(),
@@ -156,25 +168,22 @@ pub async fn spawn_protocol_listeners(
         });
     }
 
-    // Sync WebSocket listener for NodeDB-Lite clients.
-    let sync_config = crate::control::server::sync::listener::SyncListenerConfig::default();
-    match crate::control::server::sync::listener::start_sync_listener(
+    // Sync WebSocket listener for NodeDB-Lite clients (socket already bound).
+    let sync_config = crate::control::server::sync::listener::SyncListenerConfig {
+        listen_addr: config.sync_addr(),
+        ..Default::default()
+    };
+    let sync_state = crate::control::server::sync::listener::serve_sync_listener(
+        sync_listener,
         sync_config,
         Some(Arc::clone(&shared)),
     )
-    .await
-    {
-        Ok(sync_state) => {
-            info!(
-                addr = %sync_state.config.listen_addr,
-                max_sessions = sync_state.config.max_sessions,
-                "sync WebSocket listener started"
-            );
-        }
-        Err(e) => {
-            tracing::warn!(error = %e, "sync listener failed to start (non-fatal)");
-        }
-    }
+    .await;
+    info!(
+        addr = %sync_state.config.listen_addr,
+        max_sessions = sync_state.config.max_sessions,
+        "sync WebSocket listener started"
+    );
 
     // Signal readiness to systemd and cluster lifecycle.
     if let Some(handle) = cluster_handle {
@@ -184,18 +193,34 @@ pub async fn spawn_protocol_listeners(
     nodedb_cluster::readiness::notify_ready();
 }
 
+/// Every protocol socket, bound before any accept loop starts.
+pub struct BoundListeners {
+    pub native: Listener,
+    pub pgwire: PgListener,
+    pub http: TcpListener,
+    pub sync: TcpListener,
+    pub ilp: Option<IlpListener>,
+    pub resp: Option<RespListener>,
+}
+
 /// Bind all protocol listeners to their configured addresses.
-pub async fn bind_listeners(
-    config: &ServerConfig,
-) -> anyhow::Result<(
-    Listener,
-    PgListener,
-    Option<IlpListener>,
-    Option<RespListener>,
-)> {
+///
+/// This is the single fail-fast point for listener setup: it runs before the
+/// node waits on cluster readiness and before any accept loop is spawned, so
+/// a port conflict on *any* protocol — including HTTP and sync, which serve
+/// from detached tasks — aborts boot while nothing is exposed yet. Never
+/// move a bind out of here into a spawned task; that is how a server ends up
+/// running for days missing a listener behind one warning line.
+pub async fn bind_listeners(config: &ServerConfig) -> anyhow::Result<BoundListeners> {
     let native = crate::control::server::listener::Listener::bind(config.native_addr()).await?;
     let pgwire =
         crate::control::server::pgwire::listener::PgListener::bind(config.pgwire_addr()).await?;
+    let http = TcpListener::bind(config.http_addr())
+        .await
+        .with_context(|| format!("bind HTTP API listener to {}", config.http_addr()))?;
+    let sync = crate::control::server::sync::listener::bind_sync_listener(config.sync_addr())
+        .await
+        .context("sync listener failed to bind")?;
     let ilp = if let Some(ilp_addr) = config.ilp_addr() {
         Some(crate::control::server::ilp_listener::IlpListener::bind(ilp_addr).await?)
     } else {
@@ -206,5 +231,12 @@ pub async fn bind_listeners(
     } else {
         None
     };
-    Ok((native, pgwire, ilp, resp))
+    Ok(BoundListeners {
+        native,
+        pgwire,
+        http,
+        sync,
+        ilp,
+        resp,
+    })
 }

nodedb/src/config/server/config.rs

@@ -39,6 +39,7 @@ use crate::config::EngineConfig;
 /// pgwire = 6432
 /// native = 6433
 /// http   = 6480
+/// sync   = 9090
 ///
 /// [auth]
 /// # ...
@@ -161,6 +162,11 @@ impl ServerConfig {
         self.addr(self.server.ports.http)
     }
 
+    /// Sync WebSocket listen address (NodeDB-Lite clients).
+    pub fn sync_addr(&self) -> SocketAddr {
+        self.addr(self.server.ports.sync)
+    }
+
     /// RESP listen address (None if disabled).
     pub fn resp_addr(&self) -> Option<SocketAddr> {
         self.server.ports.resp.map(|p| self.addr(p))
@@ -200,6 +206,7 @@ mod tests {
         assert_eq!(cfg.server.ports.native, 6433);
         assert_eq!(cfg.server.ports.pgwire, 6432);
         assert_eq!(cfg.server.ports.http, 6480);
+        assert_eq!(cfg.server.ports.sync, 9090);
         assert!(cfg.server.ports.resp.is_none());
         assert!(cfg.server.ports.ilp.is_none());
         assert!(cfg.server.data_plane_cores >= 1);

nodedb/src/config/server/env.rs

@@ -180,6 +180,7 @@ fn apply_bool_env(var: &str, target: &mut bool) {
 /// - `NODEDB_PORT_NATIVE`      — overrides `config.ports.native` (default 6433)
 /// - `NODEDB_PORT_PGWIRE`      — overrides `config.ports.pgwire` (default 6432)
 /// - `NODEDB_PORT_HTTP`        — overrides `config.ports.http` (default 6480)
+/// - `NODEDB_PORT_SYNC`        — overrides `config.ports.sync` (default 9090)
 /// - `NODEDB_PORT_RESP`        — overrides `config.ports.resp` (set to enable RESP)
 /// - `NODEDB_PORT_ILP`         — overrides `config.ports.ilp` (set to enable ILP)
 /// - `NODEDB_DATA_DIR`         — overrides `config.data_dir`
@@ -235,6 +236,7 @@ pub fn apply_env_overrides(config: &mut ServerConfig) {
     apply_port_env("NODEDB_PORT_NATIVE", &mut config.server.ports.native);
     apply_port_env("NODEDB_PORT_PGWIRE", &mut config.server.ports.pgwire);
     apply_port_env("NODEDB_PORT_HTTP", &mut config.server.ports.http);
+    apply_port_env("NODEDB_PORT_SYNC", &mut config.server.ports.sync);
     apply_optional_port_env("NODEDB_PORT_RESP", &mut config.server.ports.resp);
     apply_optional_port_env("NODEDB_PORT_ILP", &mut config.server.ports.ilp);
 
@@ -679,6 +681,29 @@ mod tests {
         unsafe { std::env::remove_var("NODEDB_MEMORY_LIMIT") };
     }
 
+    /// Tests valid and malformed `NODEDB_PORT_SYNC` sequentially to avoid
+    /// env-var races (env vars are process-global, Rust tests run in parallel).
+    #[test]
+    fn env_sync_port_overrides() {
+        // ── Valid value → overrides ports.sync ──
+        unsafe { std::env::set_var("NODEDB_PORT_SYNC", "19090") };
+        let mut cfg = ServerConfig::default();
+        apply_env_overrides(&mut cfg);
+        assert_eq!(cfg.server.ports.sync, 19090);
+
+        // ── Malformed value → ports.sync unchanged ──
+        unsafe { std::env::set_var("NODEDB_PORT_SYNC", "notaport") };
+        let mut cfg = ServerConfig::default();
+        let before = cfg.server.ports.sync;
+        apply_env_overrides(&mut cfg);
+        assert_eq!(
+            cfg.server.ports.sync, before,
+            "malformed value must not change config"
+        );
+
+        unsafe { std::env::remove_var("NODEDB_PORT_SYNC") };
+    }
+
     #[test]
     fn env_cluster_overrides() {
         // Always start clean.

nodedb/src/config/server/mod.rs

@@ -25,7 +25,7 @@ pub use observability::{
     ObservabilityConfig, OtlpConfig, OtlpExportConfig, OtlpReceiverConfig, PromqlConfig,
     apply_observability_env, validate_feature_availability,
 };
-pub use ports::PortsConfig;
+pub use ports::{DEFAULT_SYNC_PORT, PortsConfig};
 pub use retention::RetentionSettings;
 pub use scheduler::{CronTimezone, SchedulerConfig};
 pub use section::ServerSection;

nodedb/src/config/server/ports.rs

@@ -4,6 +4,13 @@
 
 use serde::{Deserialize, Serialize};
 
+/// Default sync WebSocket port.
+///
+/// Exported because the sync listener's own `SyncListenerConfig::default()`
+/// must agree with the config default — one constant, not two literals that
+/// can drift apart.
+pub const DEFAULT_SYNC_PORT: u16 = 9090;
+
 /// Port configuration for all protocol listeners.
 ///
 /// Always-on protocols have a default port. Optional protocols (RESP, ILP)
@@ -20,6 +27,9 @@ pub struct PortsConfig {
     /// HTTP API port (REST, SSE, WebSocket). Default: 6480.
     #[serde(default = "default_http_port")]
     pub http: u16,
+    /// Sync WebSocket port (NodeDB-Lite clients). Default: 9090.
+    #[serde(default = "default_sync_port")]
+    pub sync: u16,
     /// RESP (Redis-compatible) port. Disabled by default. Set to enable.
     #[serde(default)]
     pub resp: Option<u16>,
@@ -34,6 +44,7 @@ impl Default for PortsConfig {
             native: default_native_port(),
             pgwire: default_pgwire_port(),
             http: default_http_port(),
+            sync: default_sync_port(),
             resp: None,
             ilp: None,
         }
@@ -49,6 +60,9 @@ fn default_pgwire_port() -> u16 {
 fn default_http_port() -> u16 {
     6480
 }
+fn default_sync_port() -> u16 {
+    DEFAULT_SYNC_PORT
+}
 
 #[cfg(test)]
 mod tests {
@@ -60,6 +74,7 @@ mod tests {
         assert_eq!(p.native, 6433);
         assert_eq!(p.pgwire, 6432);
         assert_eq!(p.http, 6480);
+        assert_eq!(p.sync, 9090);
         assert!(p.resp.is_none());
         assert!(p.ilp.is_none());
     }