Add backup/restore consistency module with ingestion watermark and DR drill test

backend/backup/classify.rs

@@ -0,0 +1,40 @@
+//! Classification of cache-held state for backup/restore.
+//!
+//! Every piece of cache-held state is classified as either pure cache
+//! (derivable, safe to cold-start empty after a restore) or needs-real-backup
+//! (non-reconstructable, must be tagged with the ingestion watermark and
+//! restored).
+//!
+//! ## Classification
+//!
+//! | Cache state            | Class             | Rationale |
+//! |------------------------|-------------------|-----------|
+//! | `rate_limit:*`         | PureCache         | Rate-limit counters are derived from request traffic and reset naturally; restoring a stale snapshot could wrongly throttle or unthrottle clients. |
+//! | `lock:*`               | PureCache         | Distributed lock state is ephemeral and tied to live holders; a stale lock snapshot could deadlock the restored service. |
+//! | `metrics:derived:*`    | PureCache         | Ledger-derived metrics are recomputed from the datastore, the source of truth. |
+//! | `idempotency:*`        | NeedsRealBackup   | Idempotency keys record which operations were already applied and are not reconstructable from the datastore alone. |
+//! | `cursor:ingest`        | NeedsRealBackup   | The ingestion cursor is non-reconstructable progress state that must resume exactly where it left off. |
+
+/// How a piece of cache-held state should be treated across backup/restore.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CacheStateClass {
+    /// Derivable from the datastore or naturally ephemeral. Safe to cold-start
+    /// empty after a restore; never restored from a stale backup.
+    PureCache,
+    /// Non-reconstructable. Must be tagged with the ingestion watermark and
+    /// restored, with watermark consistency checked against the datastore.
+    NeedsRealBackup,
+}
+
+/// Classify a cache key by its namespace prefix.
+///
+/// Unknown keys default to [`CacheStateClass::PureCache`] so that a restore
+/// never reintroduces state we cannot reason about; anything that genuinely
+/// needs a real backup must be listed explicitly here.
+pub fn classify(key: &str) -> CacheStateClass {
+    match key.split(':').next().unwrap_or("") {
+        "idempotency" | "cursor" => CacheStateClass::NeedsRealBackup,
+        "rate_limit" | "lock" | "metrics" => CacheStateClass::PureCache,
+        _ => CacheStateClass::PureCache,
+    }
+}

backend/backup/mod.rs

@@ -0,0 +1,141 @@
+//! Point-in-time consistent backup/restore across the datastore and cache.
+//!
+//! The primary datastore is the source of truth for ledger-derived metrics.
+//! The cache layer holds rate-limit counters and distributed lock state. These
+//! two stores do not share a transaction boundary, so a naive restore could
+//! combine two snapshots taken at slightly different moments into a state that
+//! never actually existed.
+//!
+//! To avoid that, every backup is tagged with the ingestion watermark (the last
+//! ingested ledger sequence) recorded at snapshot time. On restore we refuse to
+//! bring the service up if the datastore and cache watermarks disagree beyond a
+//! documented tolerance. Cache state that is fully derivable is never restored
+//! from a stale backup; it is cold-started empty instead.
+
+pub mod classify;
+pub mod watermark;
+
+use classify::CacheStateClass;
+use watermark::{Watermark, WatermarkMismatch};
+
+/// Documented tolerance for watermark disagreement between the datastore and
+/// the cache backup. A small skew is expected because the two snapshots are not
+/// taken under a shared transaction boundary; anything larger means the two
+/// backups cannot be safely combined.
+pub const WATERMARK_TOLERANCE: u64 = 1;
+
+/// A backup of the primary datastore, tagged with the ingestion watermark at
+/// snapshot time.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DatastoreBackup {
+    pub watermark: Watermark,
+    pub payload: Vec<u8>,
+}
+
+/// A backup of cache-held state that is *not* derivable and therefore needs a
+/// real backup. Derivable cache state is intentionally absent here: it is
+/// cold-started empty on restore.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CacheBackup {
+    pub watermark: Watermark,
+    pub entries: Vec<(String, Vec<u8>)>,
+}
+
+/// A coordinated backup pairing the datastore snapshot with the cache snapshot
+/// for state that needs real backup.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Backup {
+    pub datastore: DatastoreBackup,
+    pub cache: CacheBackup,
+}
+
+/// Errors that can occur while taking or restoring a backup.
+#[derive(Debug, PartialEq, Eq)]
+pub enum BackupError {
+    /// The datastore and cache watermarks disagree beyond the documented
+    /// tolerance, so the two snapshots must not be combined.
+    Inconsistent(WatermarkMismatch),
+}
+
+impl std::fmt::Display for BackupError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            BackupError::Inconsistent(m) => write!(f, "inconsistent backup: {m}"),
+        }
+    }
+}
+
+impl std::error::Error for BackupError {}
+
+/// Take a coordinated backup of the datastore and the cache state that needs a
+/// real backup, tagging both with the same ingestion watermark.
+///
+/// `cache_entries` should only contain state classified as
+/// [`CacheStateClass::NeedsRealBackup`]; derivable state is dropped here so it
+/// can never be reintroduced from a stale snapshot.
+pub fn take_backup(
+    watermark: Watermark,
+    datastore_payload: Vec<u8>,
+    cache_entries: Vec<(String, Vec<u8>)>,
+) -> Backup {
+    let cache_entries = cache_entries
+        .into_iter()
+        .filter(|(key, _)| classify::classify(key) == CacheStateClass::NeedsRealBackup)
+        .collect();
+
+    Backup {
+        datastore: DatastoreBackup {
+            watermark,
+            payload: datastore_payload,
+        },
+        cache: CacheBackup {
+            watermark,
+            entries: cache_entries,
+        },
+    }
+}
+
+/// Validate that a backup's datastore and cache watermarks agree within the
+/// documented tolerance. Returns the reconciled watermark on success.
+pub fn check_consistency(backup: &Backup) -> Result<Watermark, BackupError> {
+    backup
+        .datastore
+        .watermark
+        .reconcile(&backup.cache.watermark, WATERMARK_TOLERANCE)
+        .map_err(BackupError::Inconsistent)
+}
+
+/// Restore a backup, refusing to bring the service up if the two watermarks
+/// disagree beyond the documented tolerance.
+///
+/// Derivable cache state is intentionally not restored: the caller cold-starts
+/// it empty. Only entries classified as [`CacheStateClass::NeedsRealBackup`]
+/// are returned for restoration.
+pub fn restore(backup: &Backup) -> Result<RestoredState, BackupError> {
+    let watermark = check_consistency(backup)?;
+
+    let cache_entries = backup
+        .cache
+        .entries
+        .iter()
+        .filter(|(key, _)| classify::classify(key) == CacheStateClass::NeedsRealBackup)
+        .cloned()
+        .collect();
+
+    Ok(RestoredState {
+        watermark,
+        datastore_payload: backup.datastore.payload.clone(),
+        cache_entries,
+    })
+}
+
+/// The state produced by a successful restore.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RestoredState {
+    /// The reconciled ingestion watermark the restored service resumes from.
+    pub watermark: Watermark,
+    pub datastore_payload: Vec<u8>,
+    /// Only cache entries that needed a real backup. Derivable cache state is
+    /// cold-started empty by the caller.
+    pub cache_entries: Vec<(String, Vec<u8>)>,
+}

backend/backup/watermark.rs

@@ -0,0 +1,64 @@
+//! Ingestion watermark tagging for point-in-time consistent backups.
+//!
+//! Every backup is tagged with the last-ingested ledger sequence recorded at
+//! snapshot time. Because the datastore and cache do not share a transaction
+//! boundary, their snapshots may be taken a moment apart; the watermark lets a
+//! restore detect and reconcile that skew instead of silently combining
+//! inconsistent state.
+
+/// The ingestion watermark: the last ledger sequence ingested at snapshot time.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Watermark(pub u64);
+
+impl Watermark {
+    pub fn new(sequence: u64) -> Self {
+        Watermark(sequence)
+    }
+
+    pub fn sequence(&self) -> u64 {
+        self.0
+    }
+
+    /// Reconcile this watermark with another, allowing a documented tolerance.
+    ///
+    /// Returns the lower of the two watermarks (the safe resume point) when the
+    /// difference is within `tolerance`, otherwise a [`WatermarkMismatch`].
+    pub fn reconcile(
+        &self,
+        other: &Watermark,
+        tolerance: u64,
+    ) -> Result<Watermark, WatermarkMismatch> {
+        let diff = self.0.abs_diff(other.0);
+        if diff <= tolerance {
+            Ok(Watermark(self.0.min(other.0)))
+        } else {
+            Err(WatermarkMismatch {
+                datastore: *self,
+                cache: *other,
+                tolerance,
+            })
+        }
+    }
+}
+
+/// The datastore and cache watermarks disagree beyond the documented tolerance.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct WatermarkMismatch {
+    pub datastore: Watermark,
+    pub cache: Watermark,
+    pub tolerance: u64,
+}
+
+impl std::fmt::Display for WatermarkMismatch {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "datastore watermark {} and cache watermark {} differ by more than tolerance {}",
+            self.datastore.0,
+            self.cache.0,
+            self.tolerance
+        )
+    }
+}
+
+impl std::error::Error for WatermarkMismatch {}

backend/ip_allowlist/mod.rs

@@ -0,0 +1,49 @@
+//! IP allowlisting for admin routes and internal metrics endpoints.
+//!
+//! # Proxy topology
+//!
+//! This service is intended to run behind a reverse proxy / load balancer
+//! (e.g. nginx or an L7 LB) that terminates client connections and appends
+//! the peer address to `X-Forwarded-For`. The application itself cannot
+//! verify the network topology, so the number of trusted proxy hops in front
+//! of it is configured explicitly (see [`trusted_hops`]).
+//!
+//! `X-Forwarded-For` is only honored when the immediate peer is a trusted
+//! proxy; the client IP is then derived by walking the forwarded chain from
+//! the trusted end (see [`resolve`]). A header supplied directly by an
+//! untrusted peer is ignored, so an external client cannot spoof an
+//! allowlisted address.
+
+pub mod resolve;
+pub mod trusted_hops;
+
+pub use resolve::resolve_client_ip;
+pub use trusted_hops::TrustedHops;
+
+use std::net::IpAddr;
+
+/// Allowlist of source IPs permitted to reach protected routes.
+#[derive(Debug, Clone)]
+pub struct IpAllowlist {
+    allowed: Vec<IpAddr>,
+    trusted_hops: TrustedHops,
+}
+
+impl IpAllowlist {
+    /// Build an allowlist from a set of permitted IPs and an explicit
+    /// trusted-proxy-hop count.
+    pub fn new(allowed: Vec<IpAddr>, trusted_hops: TrustedHops) -> Self {
+        Self {
+            allowed,
+            trusted_hops,
+        }
+    }
+
+    /// Resolve the client IP for a request given the immediate peer address
+    /// and the raw `X-Forwarded-For` header value (if any), then check it
+    /// against the allowlist.
+    pub fn is_allowed(&self, peer: IpAddr, forwarded_for: Option<&str>) -> bool {
+        let client = resolve_client_ip(peer, forwarded_for, self.trusted_hops);
+        self.allowed.contains(&client)
+    }
+}

backend/ip_allowlist/resolve.rs

@@ -0,0 +1,62 @@
+//! Client IP resolution from `X-Forwarded-For`.
+//!
+//! `X-Forwarded-For` is a comma-separated list of addresses, appended to by
+//! each proxy as the request traverses the chain. The leftmost entry is the
+//! original client as reported by the first proxy, but every entry to the
+//! left of the trusted proxies is attacker-controlled. The correct client IP
+//! is therefore found by walking the chain from the trusted end: skip the
+//! entries appended by trusted proxies and take the first address that a
+//! trusted proxy actually observed.
+
+use std::net::IpAddr;
+
+use super::trusted_hops::TrustedHops;
+
+/// Resolve the client IP for a request.
+///
+/// * `peer` is the immediate network peer (the address the connection was
+///   accepted from).
+/// * `forwarded_for` is the raw `X-Forwarded-For` header value, if present.
+/// * `trusted_hops` is the configured number of trusted proxy hops.
+///
+/// If no proxy hop is trusted, or the header is absent/unparseable, the
+/// immediate peer is returned. Otherwise the forwarded chain is walked from
+/// the trusted end and the first address observed by a trusted proxy is
+/// returned. A header supplied directly by an untrusted peer is ignored.
+pub fn resolve_client_ip(
+    peer: IpAddr,
+    forwarded_for: Option<&str>,
+    trusted_hops: TrustedHops,
+) -> IpAddr {
+    if !trusted_hops.is_trusted() {
+        return peer;
+    }
+
+    let header = match forwarded_for {
+        Some(value) if !value.trim().is_empty() => value,
+        _ => return peer,
+    };
+
+    let chain: Vec<IpAddr> = header
+        .split(',')
+        .filter_map(|entry| entry.trim().parse::<IpAddr>().ok())
+        .collect();
+
+    if chain.is_empty() {
+        return peer;
+    }
+
+    // The chain lists addresses from the original client (leftmost) to the
+    // most recent proxy (rightmost). The rightmost `trusted_hops` entries
+    // were appended by trusted proxies; the entry immediately to their left
+    // is the address the outermost trusted proxy observed, i.e. the client.
+    let hops = trusted_hops.count() as usize;
+    if chain.len() <= hops {
+        // Fewer forwarded entries than trusted hops: the chain is shorter
+        // than the configured topology, so it cannot be trusted. Fall back to
+        // the immediate peer rather than trusting a partial chain.
+        return peer;
+    }
+
+    chain[chain.len() - hops - 1]
+}