zerodds_durability_store_postgres/
lib.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3
4//! PostgreSQL cold adapter for the Durability-Service (ADR 0009) — the
5//! shared/long-term `PERSISTENT` tier for large fleets.
6//!
7//! Crate `zerodds-durability-store-postgres`. Safety classification: **STANDARD**.
8//!
9//! Where the sqlite adapter is the single-file default, this one is the tier
10//! for many producers writing into one shared, network-reachable database —
11//! the "PostgreSQL cold tier" of the RAM/SSD/PostgreSQL story. It implements
12//! the same [`DurabilityStore`] contract with identical semantics: one row per
13//! sample, `(topic, instance, sequence)` is the sample identity (a re-send is
14//! an idempotent replace), and retention is bounded ONLY by the topic
15//! [`Contract`] (ADR 0009 inv. 1).
16//!
17//! Connection uses `NoTls` — put the daemon and the database on a trusted
18//! network segment, or front the database with a TLS-terminating proxy; the
19//! connection string is a standard libpq keyword/URI form.
20//!
21//! With the `timescaledb` feature the samples table is a TimescaleDB hypertable
22//! partitioned on `created_nanos` (needs the extension in the target database);
23//! sample identity is preserved by an explicit delete-then-insert on the
24//! `(topic, instance, sequence)` key, so idempotency holds regardless of the
25//! partitioning primary key.
26//!
27//! For analytics, connect to the same database with any SQL/BI tool — the
28//! schema is stable and documented here (ADR 0009: adapters expose their own
29//! native read interface alongside the DDS path).
30
31#![forbid(unsafe_code)]
32
33use std::collections::BTreeMap;
34use std::sync::Mutex;
35use std::time::{Duration, SystemTime, UNIX_EPOCH};
36
37use postgres::types::ToSql;
38use postgres::{Client, NoTls};
39use zerodds_durability_store::{
40    Contract, Cursor, DurabilitySample, DurabilityStore, Page, Result, Selector, StoreError,
41    StoreStats,
42};
43use zerodds_qos::policies::history::HistoryKind;
44
45const DEFAULT_PAGE: usize = 1024;
46
47#[cfg(not(feature = "timescaledb"))]
48const SCHEMA: &str = "
49CREATE TABLE IF NOT EXISTS samples (
50    topic           TEXT     NOT NULL,
51    instance        BYTEA    NOT NULL,
52    sequence        BIGINT   NOT NULL,
53    created_nanos   BIGINT   NOT NULL,
54    payload         BYTEA    NOT NULL,
55    representation  SMALLINT NOT NULL DEFAULT 1,
56    big_endian      BOOLEAN  NOT NULL DEFAULT false,
57    source_guid     BYTEA    NOT NULL DEFAULT '\\x00000000000000000000000000000000',
58    source_sequence BIGINT   NOT NULL DEFAULT -1,
59    PRIMARY KEY (topic, instance, sequence)
60);
61CREATE INDEX IF NOT EXISTS idx_samples_topic ON samples(topic, instance, sequence);
62CREATE TABLE IF NOT EXISTS unregistered (
63    topic     TEXT   NOT NULL,
64    instance  BYTEA  NOT NULL,
65    at_nanos  BIGINT NOT NULL,
66    PRIMARY KEY (topic, instance)
67);
68";
69
70// TimescaleDB requires the partitioning column in every unique index, so the
71// primary key includes `created_nanos`; sample identity on
72// `(topic, instance, sequence)` is upheld by the delete-then-insert in `store`.
73#[cfg(feature = "timescaledb")]
74const SCHEMA: &str = "
75CREATE TABLE IF NOT EXISTS samples (
76    topic           TEXT     NOT NULL,
77    instance        BYTEA    NOT NULL,
78    sequence        BIGINT   NOT NULL,
79    created_nanos   BIGINT   NOT NULL,
80    payload         BYTEA    NOT NULL,
81    representation  SMALLINT NOT NULL DEFAULT 1,
82    big_endian      BOOLEAN  NOT NULL DEFAULT false,
83    source_guid     BYTEA    NOT NULL DEFAULT '\\x00000000000000000000000000000000',
84    source_sequence BIGINT   NOT NULL DEFAULT -1,
85    PRIMARY KEY (topic, instance, sequence, created_nanos)
86);
87CREATE INDEX IF NOT EXISTS idx_samples_topic ON samples(topic, instance, sequence);
88CREATE EXTENSION IF NOT EXISTS timescaledb;
89SELECT create_hypertable('samples', 'created_nanos',
90    chunk_time_interval => 86400000000000, if_not_exists => TRUE, migrate_data => TRUE);
91CREATE TABLE IF NOT EXISTS unregistered (
92    topic     TEXT   NOT NULL,
93    instance  BYTEA  NOT NULL,
94    at_nanos  BIGINT NOT NULL,
95    PRIMARY KEY (topic, instance)
96);
97";
98
99/// PostgreSQL-backed durability store.
100pub struct PostgresStore {
101    client: Mutex<Client>,
102    contracts: Mutex<BTreeMap<String, Contract>>,
103    default_contract: Contract,
104}
105
106/// A fixed advisory-lock key that serializes `CREATE TABLE IF NOT EXISTS`
107/// across connections. `IF NOT EXISTS` is not safe against a concurrent create
108/// (two sessions both pass the existence check, then collide on the system
109/// catalog — `pg_type_typname_nsp_index`), which happens when several daemons
110/// (one per domain) share a database. Holding this lock around the DDL lets
111/// exactly one session create the schema; the rest then see it already exists.
112const SCHEMA_LOCK_KEY: i64 = 0x7A64_6473_6368_656D; // "zddsschem"
113
114fn backend(ctx: &str, e: postgres::Error) -> StoreError {
115    // `postgres::Error`'s own `Display` is terse ("db error"); the useful text
116    // (the server's message) lives in the attached `DbError`.
117    let detail = e
118        .as_db_error()
119        .map(|d| d.message().to_string())
120        .unwrap_or_else(|| e.to_string());
121    StoreError::Backend(format!("postgres store: {ctx}: {detail}"))
122}
123
124fn nanos_of(t: SystemTime) -> i64 {
125    t.duration_since(UNIX_EPOCH)
126        .map(|d| i64::try_from(d.as_nanos()).unwrap_or(i64::MAX))
127        .unwrap_or(0)
128}
129
130fn time_of(nanos: i64) -> SystemTime {
131    UNIX_EPOCH + Duration::from_nanos(nanos.max(0) as u64)
132}
133
134impl PostgresStore {
135    /// Connects to PostgreSQL (libpq keyword/URI connection string, e.g.
136    /// `postgres://user@host/db` or `host=… user=… dbname=…`) and ensures the
137    /// schema. The samples survive a full process/system restart in the
138    /// database; contracts live in memory and are re-registered on startup.
139    ///
140    /// # Errors
141    /// Connection, schema, or (with `timescaledb`) hypertable-setup failure.
142    pub fn connect(conn_str: &str, default_contract: Contract) -> Result<Self> {
143        let client = Client::connect(conn_str, NoTls).map_err(|e| backend("connect", e))?;
144        Self::init(client, default_contract)
145    }
146
147    fn init(mut client: Client, default_contract: Contract) -> Result<Self> {
148        // Serialize schema creation across connections (see SCHEMA_LOCK_KEY):
149        // the advisory lock is held only for the DDL and released right after,
150        // so concurrent daemons initialise safely without a catalog race.
151        client
152            .execute("SELECT pg_advisory_lock($1)", &[&SCHEMA_LOCK_KEY])
153            .map_err(|e| backend("schema lock", e))?;
154        let schema_result = client.batch_execute(SCHEMA);
155        let unlock_result = client.execute("SELECT pg_advisory_unlock($1)", &[&SCHEMA_LOCK_KEY]);
156        schema_result.map_err(|e| backend("schema", e))?;
157        unlock_result.map_err(|e| backend("schema unlock", e))?;
158        Ok(Self {
159            client: Mutex::new(client),
160            contracts: Mutex::new(BTreeMap::new()),
161            default_contract,
162        })
163    }
164
165    fn lock_client(&self) -> Result<std::sync::MutexGuard<'_, Client>> {
166        self.client
167            .lock()
168            .map_err(|_| StoreError::Poisoned("postgres client"))
169    }
170
171    fn contract_for(&self, topic: &str) -> Result<Contract> {
172        Ok(self
173            .contracts
174            .lock()
175            .map_err(|_| StoreError::Poisoned("postgres contracts"))?
176            .get(topic)
177            .copied()
178            .unwrap_or(self.default_contract))
179    }
180
181    fn count(client: &mut Client, sql: &str, p: &[&(dyn ToSql + Sync)]) -> Result<i64> {
182        let row = client.query_one(sql, p).map_err(|e| backend("count", e))?;
183        Ok(row.get::<_, i64>(0))
184    }
185}
186
187impl DurabilityStore for PostgresStore {
188    fn set_contract(&self, topic: &str, contract: Contract) -> Result<()> {
189        self.contracts
190            .lock()
191            .map_err(|_| StoreError::Poisoned("postgres contracts"))?
192            .insert(topic.to_string(), contract);
193        Ok(())
194    }
195
196    fn store(&self, sample: DurabilitySample) -> Result<()> {
197        let contract = self.contract_for(&sample.topic)?;
198        let mut client = self.lock_client()?;
199        let inst = &sample.instance_key[..];
200        let seq = sample.sequence as i64;
201
202        // A re-send of an already-stored (topic, instance, sequence) is
203        // idempotent (reliable retransmit) — it grows nothing, so a KEEP_ALL
204        // cap must not reject it.
205        let is_resend = Self::count(
206            &mut client,
207            "SELECT COUNT(*) FROM samples WHERE topic=$1 AND instance=$2 AND sequence=$3",
208            &[&sample.topic, &inst, &seq],
209        )? > 0;
210
211        // Contract caps (identical to the sqlite adapter).
212        if !is_resend
213            && contract.samples_bounded()
214            && matches!(contract.history_kind, HistoryKind::KeepAll)
215        {
216            let n = Self::count(
217                &mut client,
218                "SELECT COUNT(*) FROM samples WHERE topic=$1",
219                &[&sample.topic],
220            )?;
221            if n >= i64::from(contract.max_samples) {
222                return Err(StoreError::OutOfResources("max_samples"));
223            }
224        }
225        if contract.instances_bounded() {
226            let exists = Self::count(
227                &mut client,
228                "SELECT COUNT(*) FROM samples WHERE topic=$1 AND instance=$2",
229                &[&sample.topic, &inst],
230            )? > 0;
231            if !exists {
232                let insts = Self::count(
233                    &mut client,
234                    "SELECT COUNT(DISTINCT instance) FROM samples WHERE topic=$1",
235                    &[&sample.topic],
236                )?;
237                if insts >= i64::from(contract.max_instances) {
238                    return Err(StoreError::OutOfResources("max_instances"));
239                }
240            }
241        }
242        if !is_resend
243            && contract.per_instance_bounded()
244            && matches!(contract.history_kind, HistoryKind::KeepAll)
245        {
246            let n = Self::count(
247                &mut client,
248                "SELECT COUNT(*) FROM samples WHERE topic=$1 AND instance=$2",
249                &[&sample.topic, &inst],
250            )?;
251            if n >= i64::from(contract.max_samples_per_instance) {
252                return Err(StoreError::OutOfResources("max_samples_per_instance"));
253            }
254        }
255
256        // Delete-then-insert on the (topic, instance, sequence) identity: an
257        // explicit upsert that holds whether or not `created_nanos` is part of
258        // the primary key (it is under the `timescaledb` feature).
259        let created = nanos_of(sample.created_at);
260        let rep = i16::from(sample.representation);
261        let src_guid = sample.source_guid.to_vec();
262        client
263            .execute(
264                "DELETE FROM samples WHERE topic=$1 AND instance=$2 AND sequence=$3",
265                &[&sample.topic, &inst, &seq],
266            )
267            .map_err(|e| backend("replace delete", e))?;
268        client
269            .execute(
270                "INSERT INTO samples(topic,instance,sequence,created_nanos,payload,representation,big_endian,source_guid,source_sequence) \
271                 VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)",
272                &[
273                    &sample.topic,
274                    &inst,
275                    &seq,
276                    &created,
277                    &sample.payload,
278                    &rep,
279                    &sample.big_endian,
280                    &src_guid,
281                    &sample.source_sequence,
282                ],
283            )
284            .map_err(|e| backend("insert", e))?;
285
286        // KEEP_LAST: trim to the newest `depth` per instance.
287        if matches!(contract.history_kind, HistoryKind::KeepLast) {
288            let depth = contract.effective_depth() as i64;
289            client
290                .execute(
291                    "DELETE FROM samples WHERE topic=$1 AND instance=$2 AND sequence NOT IN \
292                     (SELECT sequence FROM samples WHERE topic=$1 AND instance=$2 \
293                      ORDER BY sequence DESC LIMIT $3)",
294                    &[&sample.topic, &inst, &depth],
295                )
296                .map_err(|e| backend("keep_last trim", e))?;
297        }
298        Ok(())
299    }
300
301    fn query(&self, topic: &str, selector: &Selector) -> Result<Page> {
302        let mut client = self.lock_client()?;
303        let limit = selector.limit.unwrap_or(DEFAULT_PAGE);
304        let mut sql = String::from(
305            "SELECT instance,sequence,created_nanos,payload,representation,big_endian,source_guid,source_sequence \
306             FROM samples WHERE topic=$1",
307        );
308        let topic_owned = topic.to_string();
309        let mut binds: Vec<Box<dyn ToSql + Sync>> = vec![Box::new(topic_owned.clone())];
310        if let Some(k) = selector.instance_key {
311            binds.push(Box::new(k.to_vec()));
312            sql.push_str(&format!(" AND instance=${}", binds.len()));
313        }
314        if let Some(lo) = selector.seq_from {
315            binds.push(Box::new(lo as i64));
316            sql.push_str(&format!(" AND sequence>=${}", binds.len()));
317        }
318        if let Some(hi) = selector.seq_to {
319            binds.push(Box::new(hi as i64));
320            sql.push_str(&format!(" AND sequence<=${}", binds.len()));
321        }
322        if let Some(t0) = selector.time_from {
323            binds.push(Box::new(nanos_of(t0)));
324            sql.push_str(&format!(" AND created_nanos>=${}", binds.len()));
325        }
326        if let Some(t1) = selector.time_to {
327            binds.push(Box::new(nanos_of(t1)));
328            sql.push_str(&format!(" AND created_nanos<=${}", binds.len()));
329        }
330        if let Some((ck, cs)) = selector.after {
331            binds.push(Box::new(ck.to_vec()));
332            let i1 = binds.len();
333            binds.push(Box::new(ck.to_vec()));
334            let i2 = binds.len();
335            binds.push(Box::new(cs as i64));
336            let i3 = binds.len();
337            sql.push_str(&format!(
338                " AND (instance>${i1} OR (instance=${i2} AND sequence>${i3}))"
339            ));
340        }
341        // Fetch limit+1 to detect a following page.
342        binds.push(Box::new((limit + 1) as i64));
343        sql.push_str(&format!(
344            " ORDER BY instance,sequence LIMIT ${}",
345            binds.len()
346        ));
347
348        let params: Vec<&(dyn ToSql + Sync)> = binds.iter().map(|b| b.as_ref()).collect();
349        let rows = client
350            .query(&sql, params.as_slice())
351            .map_err(|e| backend("query", e))?;
352
353        let mut samples = Vec::with_capacity(rows.len());
354        for r in &rows {
355            let inst: Vec<u8> = r.get(0);
356            let mut key = [0u8; 16];
357            if inst.len() == 16 {
358                key.copy_from_slice(&inst);
359            }
360            let sg: Vec<u8> = r.get(6);
361            let mut source_guid = [0u8; 16];
362            if sg.len() == 16 {
363                source_guid.copy_from_slice(&sg);
364            }
365            samples.push(DurabilitySample {
366                topic: topic_owned.clone(),
367                instance_key: key,
368                sequence: r.get::<_, i64>(1) as u64,
369                created_at: time_of(r.get::<_, i64>(2)),
370                payload: r.get(3),
371                representation: r.get::<_, i16>(4) as u8,
372                big_endian: r.get::<_, bool>(5),
373                source_guid,
374                source_sequence: r.get::<_, i64>(7),
375            });
376        }
377        let exhausted = samples.len() <= limit;
378        samples.truncate(limit);
379        let next: Option<Cursor> = if exhausted {
380            None
381        } else {
382            samples.last().map(|s| (s.instance_key, s.sequence))
383        };
384        Ok(Page { samples, next })
385    }
386
387    fn unregister(&self, topic: &str, instance_key: &[u8; 16], now: SystemTime) -> Result<()> {
388        let mut client = self.lock_client()?;
389        let inst = &instance_key[..];
390        let at = nanos_of(now);
391        client
392            .execute(
393                "INSERT INTO unregistered(topic,instance,at_nanos) VALUES ($1,$2,$3) \
394                 ON CONFLICT (topic,instance) DO UPDATE SET at_nanos=EXCLUDED.at_nanos",
395                &[&topic, &inst, &at],
396            )
397            .map_err(|e| backend("unregister", e))?;
398        Ok(())
399    }
400
401    fn cleanup(&self, now: SystemTime) -> Result<usize> {
402        let mut client = self.lock_client()?;
403        let rows = client
404            .query("SELECT topic,instance,at_nanos FROM unregistered", &[])
405            .map_err(|e| backend("cleanup scan", e))?;
406        let due: Vec<(String, Vec<u8>)> = rows
407            .iter()
408            .filter_map(|r| {
409                let topic: String = r.get(0);
410                let inst: Vec<u8> = r.get(1);
411                let at: i64 = r.get(2);
412                let delay = self.contract_for(&topic).ok()?.cleanup_delay;
413                let deadline = time_of(at).checked_add(delay)?;
414                (now >= deadline).then_some((topic, inst))
415            })
416            .collect();
417        let mut removed = 0usize;
418        for (topic, inst) in due {
419            client
420                .execute(
421                    "DELETE FROM samples WHERE topic=$1 AND instance=$2",
422                    &[&topic, &inst],
423                )
424                .map_err(|e| backend("cleanup delete samples", e))?;
425            client
426                .execute(
427                    "DELETE FROM unregistered WHERE topic=$1 AND instance=$2",
428                    &[&topic, &inst],
429                )
430                .map_err(|e| backend("cleanup delete marker", e))?;
431            removed += 1;
432        }
433        Ok(removed)
434    }
435
436    fn stats(&self, topic: &str) -> Result<StoreStats> {
437        let mut client = self.lock_client()?;
438        let samples = Self::count(
439            &mut client,
440            "SELECT COUNT(*) FROM samples WHERE topic=$1",
441            &[&topic],
442        )? as usize;
443        let instances = Self::count(
444            &mut client,
445            "SELECT COUNT(DISTINCT instance) FROM samples WHERE topic=$1",
446            &[&topic],
447        )? as usize;
448        let bytes = Self::count(
449            &mut client,
450            "SELECT COALESCE(SUM(LENGTH(payload)),0)::BIGINT FROM samples WHERE topic=$1",
451            &[&topic],
452        )? as u64;
453        Ok(StoreStats {
454            samples,
455            instances,
456            bytes,
457        })
458    }
459}
zerodds_durability_store_postgres/lib.rs

zerodds_durability_store_postgres/
lib.rs
