libsql-server: add an operation-owned namespace fence with positive drain

README.md

@@ -33,8 +33,14 @@
 
 ---
 
-> [!NOTE]
-> This repository contains libSQL, a fork of SQLite developed by Turso. For the full SQLite rewriten in Rust (also by Turso), please visit [tursodatabase/turso](https://github.com/tursodatabase/turso).
+> [!IMPORTANT]
+> **Turso database and libSQL are two different projects from the same team.**
+>
+> **libSQL** (this repository) is an open-source fork of SQLite. It extends SQLite with features like embedded replicas and remote access, but inherits SQLite's fundamental limitations such as the single-writer model.
+>
+> **[Turso database](https://github.com/tursodatabase/turso)** is a SQLite-compatible database rewritten from scratch in Rust. It is **not** a fork of SQLite — it is a completely new implementation that goes beyond what any SQLite fork can offer, including concurrent writes and bi-directional sync with offline support. Turso is currently in beta.
+>
+> **If you're starting a new project, you probably want to look into [Turso](https://github.com/tursodatabase/turso).** libSQL is actively maintained, but new features are being developed in Turso.
 
 ## Documentation
 

libsql-replication/proto/metadata.proto

@@ -27,4 +27,17 @@ message DatabaseConfig {
     optional bool shared_schema = 11;
     optional string shared_schema_name = 12;
     optional DurabilityMode durability_mode = 13;
+    // The namespace fence as seen by the primary when it answered. Only ever filled by the
+    // primary's replication `Hello`, and only while a fence is active; it is never part of a
+    // stored configuration. Absent from older primaries; older replicas ignore it and still
+    // see the legacy `block_*` fields above.
+    optional ReplicatedFence fence = 14;
+}
+
+// The part of a namespace fence a replica needs to apply the primary's read admission.
+message ReplicatedFence {
+    // The fence state name, e.g. "SOURCE_READ_FENCED".
+    string state = 1;
+    // The revision of the fence record the state belongs to.
+    uint64 revision = 2;
 }

libsql-replication/proto/proxy.proto

@@ -43,6 +43,11 @@ message Error {
     ErrorCode code = 1;
     string message = 2;
     int32 extended_code = 3;
+    // Stable machine-readable outcome of the error, e.g. "MIGRATION_WRITE_FENCED" for a
+    // request refused by a namespace fence. Absent when the error has no typed outcome, and
+    // always absent from older servers: a receiver treats an absent field as "no typed
+    // outcome" and falls back to `code`.
+    optional string stable_code = 4;
 }
 
 message ResultRows {

libsql-replication/src/generated/metadata.rs

@@ -32,6 +32,23 @@ pub struct DatabaseConfig {
     pub shared_schema_name: ::core::option::Option<::prost::alloc::string::String>,
     #[prost(enumeration = "DurabilityMode", optional, tag = "13")]
     pub durability_mode: ::core::option::Option<i32>,
+    /// The namespace fence as seen by the primary when it answered. Only ever filled by the
+    /// primary's replication `Hello`, and only while a fence is active; it is never part of a
+    /// stored configuration. Absent from older primaries; older replicas ignore it and still
+    /// see the legacy `block_*` fields above.
+    #[prost(message, optional, tag = "14")]
+    pub fence: ::core::option::Option<ReplicatedFence>,
+}
+/// The part of a namespace fence a replica needs to apply the primary's read admission.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ReplicatedFence {
+    /// The fence state name, e.g. "SOURCE_READ_FENCED".
+    #[prost(string, tag = "1")]
+    pub state: ::prost::alloc::string::String,
+    /// The revision of the fence record the state belongs to.
+    #[prost(uint64, tag = "2")]
+    pub revision: u64,
 }
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
 #[repr(i32)]

libsql-replication/src/generated/proxy.rs

@@ -77,6 +77,12 @@ pub struct Error {
     pub message: ::prost::alloc::string::String,
     #[prost(int32, tag = "3")]
     pub extended_code: i32,
+    /// Stable machine-readable outcome of the error, e.g. "MIGRATION_WRITE_FENCED" for a
+    /// request refused by a namespace fence. Absent when the error has no typed outcome, and
+    /// always absent from older servers: a receiver treats an absent field as "no typed
+    /// outcome" and falls back to `code`.
+    #[prost(string, optional, tag = "4")]
+    pub stable_code: ::core::option::Option<::prost::alloc::string::String>,
 }
 /// Nested message and enum types in `Error`.
 pub mod error {

libsql-replication/src/rpc.rs

@@ -105,3 +105,118 @@ pub mod metadata {
     #![allow(clippy::all)]
     include!("generated/metadata.rs");
 }
+
+#[cfg(test)]
+mod test {
+    use prost::Message;
+
+    use super::metadata::{DatabaseConfig, ReplicatedFence};
+    use super::proxy::{error::ErrorCode, Error};
+
+    /// `proxy.Error` as a peer built before `stable_code` existed knows it.
+    #[derive(Clone, PartialEq, ::prost::Message)]
+    struct ErrorWithoutStableCode {
+        #[prost(enumeration = "ErrorCode", tag = "1")]
+        code: i32,
+        #[prost(string, tag = "2")]
+        message: String,
+        #[prost(int32, tag = "3")]
+        extended_code: i32,
+    }
+
+    /// The legacy part of `metadata.DatabaseConfig`, as a peer built before `fence` existed
+    /// knows it (the fields in between are skipped the same way as `fence` is).
+    #[derive(Clone, PartialEq, ::prost::Message)]
+    struct DatabaseConfigWithoutFence {
+        #[prost(bool, tag = "1")]
+        block_reads: bool,
+        #[prost(bool, tag = "2")]
+        block_writes: bool,
+        #[prost(string, optional, tag = "3")]
+        block_reason: Option<String>,
+        #[prost(uint64, tag = "4")]
+        max_db_pages: u64,
+    }
+
+    fn error(stable_code: Option<&str>) -> Error {
+        Error {
+            code: ErrorCode::SqlError as i32,
+            message: "writes are fenced".into(),
+            extended_code: 23,
+            stable_code: stable_code.map(Into::into),
+        }
+    }
+
+    #[test]
+    fn proxy_error_stable_code_is_additive() {
+        // A newer server's error, read by an older replica: the known fields are intact and
+        // the stable code is skipped.
+        let new = error(Some("MIGRATION_WRITE_FENCED"));
+        let old = ErrorWithoutStableCode::decode(&new.encode_to_vec()[..]).unwrap();
+        assert_eq!(
+            old,
+            ErrorWithoutStableCode {
+                code: ErrorCode::SqlError as i32,
+                message: "writes are fenced".into(),
+                extended_code: 23,
+            }
+        );
+
+        // An older server's error, read by a newer replica: no typed outcome.
+        let decoded = Error::decode(&old.encode_to_vec()[..]).unwrap();
+        assert_eq!(decoded, error(None));
+
+        // Without a stable code the encoding is exactly the older one, so an error that has no
+        // typed outcome is unchanged on the wire.
+        assert_eq!(error(None).encode_to_vec(), old.encode_to_vec());
+
+        // And the field round-trips between newer peers.
+        assert_eq!(Error::decode(&new.encode_to_vec()[..]).unwrap(), new);
+    }
+
+    fn config(fence: Option<ReplicatedFence>) -> DatabaseConfig {
+        DatabaseConfig {
+            block_reads: true,
+            block_writes: true,
+            block_reason: Some("namespace fence".into()),
+            max_db_pages: 1024,
+            fence,
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn replicated_fence_is_additive() {
+        let fence = ReplicatedFence {
+            state: "SOURCE_READ_FENCED".into(),
+            revision: 3,
+        };
+
+        // A newer primary's config, read by an older replica: the legacy block fields are
+        // intact, so the older replica still applies the legacy mirror of the fence.
+        let new = config(Some(fence.clone()));
+        let old = DatabaseConfigWithoutFence::decode(&new.encode_to_vec()[..]).unwrap();
+        assert_eq!(
+            old,
+            DatabaseConfigWithoutFence {
+                block_reads: true,
+                block_writes: true,
+                block_reason: Some("namespace fence".into()),
+                max_db_pages: 1024,
+            }
+        );
+
+        // An older primary's config, read by a newer replica: no fence.
+        let decoded = DatabaseConfig::decode(&old.encode_to_vec()[..]).unwrap();
+        assert_eq!(decoded.fence, None);
+        assert_eq!(decoded, config(None));
+
+        // Without a fence the encoding is exactly the older one: a stored configuration, which
+        // never carries a fence, is unchanged.
+        assert_eq!(config(None).encode_to_vec(), old.encode_to_vec());
+
+        // And the fence round-trips between newer peers.
+        let round_trip = DatabaseConfig::decode(&new.encode_to_vec()[..]).unwrap();
+        assert_eq!(round_trip.fence, Some(fence));
+    }
+}