diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 2696b7c6..8dd61e85 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -122,7 +122,7 @@ jobs: - name: Start regtest services run: ./regtest.sh start - name: Run lib_sdk tests - run: cargo test --features "uniffi,test-utils,vls" --test lib_sdk -- --test-threads=1 + run: SKIP_INIT=1 cargo test --features "uniffi,test-utils,vls" --test lib_sdk -- --test-threads=1 --nocapture - name: Stop regtest services if: always() run: ./regtest.sh stop @@ -139,20 +139,82 @@ jobs: rustflags: "" - name: Build with vss feature run: cargo build --features vss + - name: Run deterministic VSS durability regressions + run: cargo test --features vss "test::vss_durability_gaps" -- --test-threads=1 - name: Start regtest and VSS services run: VSS=1 ./regtest.sh start - name: Run VSS tests (e2e against regtest + VSS server) - run: cargo test --features vss "test::vss::tests" -- --test-threads=1 + run: SKIP_INIT=1 cargo test --features vss "test::vss::tests" -- --test-threads=1 - name: Run VSS unreachable openchannel e2e test - run: cargo test --features vss "test::vss_unreachable_openchannel" -- --test-threads=1 + run: SKIP_INIT=1 cargo test --features vss "test::vss_unreachable_openchannel" -- --test-threads=1 + - name: Run local SQLite process-kill matrix + run: cargo test --features "uniffi,test-utils,vls,vss" --lib kv_store_os_kill_matrix -- --test-threads=1 + - name: Run VSS process-kill matrix + run: cargo test --features "uniffi,test-utils,vls,vss" --bin rgb-lightning-node synced_kv_os_kill_matrix -- --ignored --test-threads=1 - name: Run VSS full-restore e2e test (wipe node dir, restore from seed) - run: cargo test --features "uniffi,test-utils,vls,vss" --test lib_sdk vss_restore -- --test-threads=1 + run: SKIP_INIT=1 cargo test --features "uniffi,test-utils,vls,vss" --test lib_sdk vss_restore -- --test-threads=1 - name: Run consignment re-import e2e test (restore without RGB backup) run: cargo test --features "uniffi,test-utils,vls,vss" --test lib_sdk vss_consignment_reimport -- --test-threads=1 - name: Stop regtest and VSS services if: always() run: docker compose --profile vss down -v --remove-orphans + persistence-sanitizers: + runs-on: ubuntu-latest + timeout-minutes: 120 + strategy: + fail-fast: false + matrix: + sanitizer: [address, thread] + env: + CARGO_BUILD_JOBS: 1 + RUSTFLAGS: >- + -Zsanitizer=${{ matrix.sanitizer }} -Cdebuginfo=0 -Copt-level=0 + -Ccodegen-units=256 -Clinker=clang -Clink-arg=-fuse-ld=lld + RUSTDOCFLAGS: -Zsanitizer=${{ matrix.sanitizer }} + ASAN_OPTIONS: detect_leaks=1:halt_on_error=1 + TSAN_OPTIONS: halt_on_error=1 + steps: + - uses: actions/checkout@v6 + with: + submodules: true + - uses: actions-rust-lang/setup-rust-toolchain@v1 + with: + toolchain: nightly + components: rust-src + - name: Install sanitizer build dependencies + run: | + sudo apt-get update + sudo apt-get install -y clang lld cmake pkg-config libssl-dev libsqlite3-dev protobuf-compiler libudev-dev + - name: Run SQLite persistence kill matrix under AddressSanitizer + if: matrix.sanitizer == 'address' + run: >- + cargo +nightly test -Zbuild-std --target x86_64-unknown-linux-gnu + --features "uniffi,test-utils,vls,vss" --lib + kv_store::persistence_tests::kv_store_os_kill_matrix -- --exact --test-threads=1 + - name: Start VSS services for ThreadSanitizer + if: matrix.sanitizer == 'thread' + run: | + docker compose --profile vss up -d --build vss-postgres vss-server + for attempt in $(seq 1 60); do + if curl --silent --show-error --max-time 2 --output /dev/null http://127.0.0.1:8081/vss/list-key-versions; then + exit 0 + fi + sleep 2 + done + docker compose --profile vss logs vss-postgres vss-server + exit 1 + - name: Run VSS ordering regression under ThreadSanitizer + if: matrix.sanitizer == 'thread' + run: >- + cargo +nightly test -Zbuild-std --target x86_64-unknown-linux-gnu + --all-features --bin rgb-lightning-node + test::vss::tests::synced_kv_store_drain_never_regresses_newer_write + -- --exact --test-threads=1 + - name: Stop sanitizer VSS services + if: always() && matrix.sanitizer == 'thread' + run: docker compose --profile vss down -v --remove-orphans + gossip-tests: runs-on: warp-ubuntu-latest-x64-16x timeout-minutes: 60 @@ -168,7 +230,7 @@ jobs: - name: Start regtest services run: ./regtest.sh start - name: Run gossip tests (e2e against regtest + RGS server) - run: cargo test "test::gossip_" -- --test-threads=1 + run: SKIP_INIT=1 cargo test "test::gossip_" -- --test-threads=1 - name: Dump gossip service logs (on failure) if: failure() run: | diff --git a/compose.yaml b/compose.yaml index 9ce5d39a..406f69ff 100644 --- a/compose.yaml +++ b/compose.yaml @@ -2,13 +2,13 @@ services: bitcoind: platform: linux/amd64 image: registry.gitlab.com/hashbeam/docker/bitcoind:30.2 - command: ["-fallbackfee=0.0002", "-rest"] + command: ["-fallbackfee=0.0002", "-rest", "-rpcallowip=0.0.0.0/0"] environment: MYUID: 1000 MYGID: 1000 RPCAUTH: "user:84c66d54d736d8b02aaa5b02e07e759b$$cc56c229b2a49ae2bfd5932cc8a6135d435bb9a7ac037ddd351d65936082c03d" ports: - - 18443:18443 + - 127.0.0.1:18443:18443 volumes: - ./datacore:/srv/app/.bitcoin electrs: diff --git a/regtest.sh b/regtest.sh index 1dc061b6..9c82bf0a 100755 --- a/regtest.sh +++ b/regtest.sh @@ -79,12 +79,28 @@ _wait_for_proxy() { done } +_wait_for_esplora() { + start_time=$(date +%s) + until curl --fail --silent "http://127.0.0.1:3002/block-height/0" >/dev/null; do + current_time=$(date +%s) + if [ $((current_time - start_time)) -gt $TIMEOUT ]; then + echo "Timeout waiting for Esplora to start" + $COMPOSE --profile esplora logs esplora + exit 1 + fi + sleep 1 + done +} + _start_services() { _stop_services mkdir -p data{core,index,ldk0,ldk1,ldk2,rgs} # see compose.yaml for the exposed ports EXPOSED_PORTS=(3000 50001) + if [ "${ESPLORA:-}" = "1" ]; then + EXPOSED_PORTS+=(3002) + fi for port in "${EXPOSED_PORTS[@]}"; do if _is_port_bound "$port"; then _die "port $port is already bound, services can't be started" @@ -102,6 +118,13 @@ _start_services() { echo "waiting for proxy to be ready" _wait_for_proxy + if [ "${ESPLORA:-}" = "1" ]; then + echo "starting Esplora server..." + $COMPOSE --profile esplora up -d esplora + _wait_for_esplora + echo "Esplora server available at http://localhost:3002" + fi + # optionally start VSS server if [ "${VSS:-}" = "1" ]; then echo "starting VSS server..." @@ -111,7 +134,7 @@ _start_services() { } _stop_services() { - $COMPOSE --profile gossip down -v --remove-orphans + $COMPOSE --profile gossip --profile vss --profile esplora down -v --remove-orphans rm -rf data{core,index,ldk0,ldk1,ldk2,rgs} 2>/dev/null || sudo rm -rf data{core,index,ldk0,ldk1,ldk2,rgs} } diff --git a/src/kv_store.rs b/src/kv_store.rs index d478285a..6849a900 100644 --- a/src/kv_store.rs +++ b/src/kv_store.rs @@ -6,13 +6,98 @@ use sea_orm::sea_query::OnConflict; use crate::database::entities::{KvStoreActMod, KvStoreColumn, KvStoreEntity}; use crate::runtime::block_on; +#[cfg(feature = "vss")] +use sea_orm::TransactionTrait; use sea_orm::{ActiveValue, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter}; +#[cfg(test)] +fn kv_persistence_checkpoint(name: &str) { + use std::fs::OpenOptions; + use std::io::Write; + + if let Ok(path) = std::env::var("RLN_KV_PERSISTENCE_TRACE_PATH") { + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(path) + .expect("open KV persistence trace"); + writeln!(file, "{name}").expect("write KV persistence trace"); + file.sync_all().expect("sync KV persistence trace"); + } + + if std::env::var("RLN_KV_KILL_AT").as_deref() == Ok(name) { + let path = std::env::var("RLN_KV_KILL_READY_PATH").expect("KV persistence kill-ready path"); + let mut file = std::fs::File::create(path).expect("create KV persistence kill-ready file"); + writeln!(file, "{name}").expect("write KV persistence kill-ready file"); + file.sync_all() + .expect("sync KV persistence kill-ready file"); + loop { + std::thread::park(); + } + } +} + +#[cfg(not(test))] +#[inline] +fn kv_persistence_checkpoint(_name: &str) {} + /// sea-orm based KVStore implementation for LDK persistence. pub struct SeaOrmKvStore { connection: Arc, } +#[cfg(feature = "vss")] +pub(crate) struct KvStoreKey<'a> { + primary_namespace: &'a str, + secondary_namespace: &'a str, + key: &'a str, +} + +#[cfg(feature = "vss")] +impl<'a> KvStoreKey<'a> { + pub(crate) fn new( + primary_namespace: &'a str, + secondary_namespace: &'a str, + key: &'a str, + ) -> Self { + Self { + primary_namespace, + secondary_namespace, + key, + } + } +} + +#[cfg(feature = "vss")] +pub(crate) struct KvStoreEntry<'a> { + key: KvStoreKey<'a>, + value: Vec, +} + +#[cfg(feature = "vss")] +impl<'a> KvStoreEntry<'a> { + pub(crate) fn new( + primary_namespace: &'a str, + secondary_namespace: &'a str, + key: &'a str, + value: Vec, + ) -> Self { + Self { + key: KvStoreKey::new(primary_namespace, secondary_namespace, key), + value, + } + } + + fn into_active_model(self) -> KvStoreActMod { + KvStoreActMod { + primary_namespace: ActiveValue::Set(self.key.primary_namespace.to_string()), + secondary_namespace: ActiveValue::Set(self.key.secondary_namespace.to_string()), + key: ActiveValue::Set(self.key.key.to_string()), + value: ActiveValue::Set(self.value), + } + } +} + impl SeaOrmKvStore { /// create a SeaOrmKvStore from an existing shared connection. /// does not run migrations (assumes they were already run). @@ -23,6 +108,130 @@ impl SeaOrmKvStore { fn get_connection(&self) -> &DatabaseConnection { &self.connection } + + /// Atomically writes a local value and its durable VSS replication intent. + /// + /// A process may terminate immediately after this transaction commits. Keeping both rows in + /// the same SQLite transaction guarantees that startup either observes the old local value + /// with the old intent, or the new local value with the new intent; it can never observe a + /// local mutation that has no remote-replay evidence. + #[cfg(feature = "vss")] + pub(crate) fn write_with_replication_intent( + &self, + value: KvStoreEntry<'_>, + intent: KvStoreEntry<'_>, + ) -> Result<(), io::Error> { + let primary_namespace = value.key.primary_namespace; + let secondary_namespace = value.key.secondary_namespace; + let key = value.key.key; + let value_model = value.into_active_model(); + let intent_model = intent.into_active_model(); + + block_on(async { + let transaction = self.get_connection().begin().await?; + kv_persistence_checkpoint("atomic-write-before-value"); + KvStoreEntity::insert(value_model) + .on_conflict( + OnConflict::columns([ + KvStoreColumn::PrimaryNamespace, + KvStoreColumn::SecondaryNamespace, + KvStoreColumn::Key, + ]) + .update_column(KvStoreColumn::Value) + .to_owned(), + ) + .exec(&transaction) + .await?; + kv_persistence_checkpoint("atomic-write-after-value"); + kv_persistence_checkpoint("atomic-write-before-intent"); + KvStoreEntity::insert(intent_model) + .on_conflict( + OnConflict::columns([ + KvStoreColumn::PrimaryNamespace, + KvStoreColumn::SecondaryNamespace, + KvStoreColumn::Key, + ]) + .update_column(KvStoreColumn::Value) + .to_owned(), + ) + .exec(&transaction) + .await?; + kv_persistence_checkpoint("atomic-write-after-intent"); + kv_persistence_checkpoint("atomic-write-before-commit"); + transaction.commit().await?; + kv_persistence_checkpoint("atomic-write-after-commit"); + Ok::<(), sea_orm::DbErr>(()) + }) + .map_err(|error| { + tracing::error!( + primary_namespace, + secondary_namespace, + key, + error = %error, + "Atomic KVStore write and replication-intent persistence failed" + ); + io::Error::new( + io::ErrorKind::Other, + format!("Atomic database write and replication intent failed: {error}"), + ) + }) + } + + /// Atomically removes a local value and records the corresponding VSS tombstone intent. + #[cfg(feature = "vss")] + pub(crate) fn remove_with_replication_intent( + &self, + key: KvStoreKey<'_>, + intent: KvStoreEntry<'_>, + ) -> Result<(), io::Error> { + let primary_namespace = key.primary_namespace; + let secondary_namespace = key.secondary_namespace; + let key = key.key; + let intent_model = intent.into_active_model(); + + block_on(async { + let transaction = self.get_connection().begin().await?; + kv_persistence_checkpoint("atomic-remove-before-value"); + KvStoreEntity::delete_many() + .filter(KvStoreColumn::PrimaryNamespace.eq(primary_namespace)) + .filter(KvStoreColumn::SecondaryNamespace.eq(secondary_namespace)) + .filter(KvStoreColumn::Key.eq(key)) + .exec(&transaction) + .await?; + kv_persistence_checkpoint("atomic-remove-after-value"); + kv_persistence_checkpoint("atomic-remove-before-intent"); + KvStoreEntity::insert(intent_model) + .on_conflict( + OnConflict::columns([ + KvStoreColumn::PrimaryNamespace, + KvStoreColumn::SecondaryNamespace, + KvStoreColumn::Key, + ]) + .update_column(KvStoreColumn::Value) + .to_owned(), + ) + .exec(&transaction) + .await?; + kv_persistence_checkpoint("atomic-remove-after-intent"); + kv_persistence_checkpoint("atomic-remove-before-commit"); + transaction.commit().await?; + kv_persistence_checkpoint("atomic-remove-after-commit"); + Ok::<(), sea_orm::DbErr>(()) + }) + .map_err(|error| { + tracing::error!( + primary_namespace, + secondary_namespace, + key, + error = %error, + "Atomic KVStore removal and replication-intent persistence failed" + ); + io::Error::new( + io::ErrorKind::Other, + format!("Atomic database removal and replication intent failed: {error}"), + ) + }) + } } impl KVStoreSync for SeaOrmKvStore { @@ -88,6 +297,7 @@ impl KVStoreSync for SeaOrmKvStore { value: ActiveValue::Set(buf), }; + kv_persistence_checkpoint("plain-write-before-value"); block_on( KvStoreEntity::insert(model) .on_conflict( @@ -111,6 +321,7 @@ impl KVStoreSync for SeaOrmKvStore { ); io::Error::new(io::ErrorKind::Other, format!("Database write failed: {e}")) })?; + kv_persistence_checkpoint("plain-write-after-value"); Ok(()) } @@ -130,6 +341,7 @@ impl KVStoreSync for SeaOrmKvStore { "KVStore remove" ); + kv_persistence_checkpoint("plain-remove-before-value"); block_on( KvStoreEntity::delete_many() .filter(KvStoreColumn::PrimaryNamespace.eq(primary_namespace)) @@ -147,6 +359,7 @@ impl KVStoreSync for SeaOrmKvStore { ); io::Error::new(io::ErrorKind::Other, format!("Database delete failed: {e}")) })?; + kv_persistence_checkpoint("plain-remove-after-value"); Ok(()) } @@ -261,3 +474,213 @@ pub(crate) mod test_support { } } } + +#[cfg(all(test, feature = "vss"))] +mod persistence_tests { + use super::*; + use rln_migration::{Migrator, MigratorTrait}; + use sea_orm::{ConnectOptions, Database}; + use std::path::{Path, PathBuf}; + use std::process::{Command, Stdio}; + use std::time::{Duration, Instant}; + + const TARGET_NS: &str = "crash-test"; + const TARGET_KEY: &str = "target"; + const INTENT_NS: &str = "crash-test-intent"; + const INTENT_KEY: &str = "target-intent"; + const OLD_VALUE: &[u8] = b"old-value"; + const NEW_VALUE: &[u8] = b"new-value"; + const WRITE_INTENT: &[u8] = b"write-intent"; + const REMOVE_INTENT: &[u8] = b"remove-intent"; + + fn open_store(path: &Path) -> SeaOrmKvStore { + let connection_string = format!("sqlite:{}?mode=rwc", path.display()); + let mut options = ConnectOptions::new(connection_string); + options.max_connections(1).min_connections(1); + let connection = + crate::runtime::block_on(Database::connect(options)).expect("open test DB"); + crate::runtime::block_on(Migrator::up(&connection, None)).expect("migrate test DB"); + SeaOrmKvStore::from_connection(Arc::new(connection)) + } + + fn prepare_fixture(mode: &str, path: &Path) { + let store = open_store(path); + if matches!( + mode, + "atomic-write" | "atomic-remove" | "plain-write" | "plain-remove" + ) { + store + .write(TARGET_NS, "", TARGET_KEY, OLD_VALUE.to_vec()) + .expect("seed target value"); + } + } + + fn child_command(mode: &str, path: &Path) -> Command { + let mut command = Command::new(std::env::current_exe().expect("current test executable")); + command + .args([ + "--exact", + "kv_store::persistence_tests::kv_store_os_kill_child", + "--ignored", + "--nocapture", + ]) + .env("RLN_KV_CHILD_MODE", mode) + .env("RLN_KV_CHILD_DB", path) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + command + } + + fn trace_checkpoints(mode: &str) -> Vec { + let directory = tempfile::tempdir().expect("trace tempdir"); + let db_path = directory.path().join("store.sqlite"); + let trace_path = directory.path().join("trace.txt"); + prepare_fixture(mode, &db_path); + let status = child_command(mode, &db_path) + .env("RLN_KV_PERSISTENCE_TRACE_PATH", &trace_path) + .status() + .expect("run trace child"); + assert!(status.success(), "KV trace child failed for {mode}"); + let prefix = match mode { + "atomic-write" => "atomic-write-", + "atomic-remove" => "atomic-remove-", + "plain-write" => "plain-write-", + "plain-remove" => "plain-remove-", + _ => unreachable!(), + }; + let checkpoints: Vec<_> = std::fs::read_to_string(trace_path) + .expect("read KV trace") + .lines() + .filter(|line| line.starts_with(prefix)) + .map(str::to_owned) + .collect(); + assert!( + !checkpoints.is_empty(), + "no persistence checkpoints for {mode}" + ); + let mut unique = checkpoints.clone(); + unique.sort(); + unique.dedup(); + assert_eq!( + checkpoints.len(), + unique.len(), + "duplicate checkpoints for {mode}" + ); + checkpoints + } + + fn kill_at_checkpoint(mode: &str, checkpoint: &str) -> PathBuf { + let directory = tempfile::tempdir().expect("kill tempdir"); + let root = directory.keep(); + let db_path = root.join("store.sqlite"); + let ready_path = root.join("ready"); + prepare_fixture(mode, &db_path); + let mut child = child_command(mode, &db_path) + .env("RLN_KV_KILL_AT", checkpoint) + .env("RLN_KV_KILL_READY_PATH", &ready_path) + .spawn() + .expect("spawn kill child"); + + let deadline = Instant::now() + Duration::from_secs(15); + loop { + if ready_path.exists() { + break; + } + if let Some(status) = child.try_wait().expect("poll kill child") { + panic!("KV child exited before {checkpoint}: {status}"); + } + if Instant::now() >= deadline { + let _ = child.kill(); + let _ = child.wait(); + panic!("KV child did not reach {checkpoint}"); + } + std::thread::sleep(Duration::from_millis(10)); + } + child.kill().expect("kill persistence child"); + let status = child.wait().expect("wait for persistence child"); + assert!(!status.success(), "killed child unexpectedly succeeded"); + db_path + } + + fn optional_read(store: &SeaOrmKvStore, namespace: &str, key: &str) -> Option> { + match store.read(namespace, "", key) { + Ok(value) => Some(value), + Err(error) if error.kind() == io::ErrorKind::NotFound => None, + Err(error) => panic!("read failed: {error}"), + } + } + + fn assert_recovered_state(mode: &str, db_path: &Path) { + let store = open_store(db_path); + let target = optional_read(&store, TARGET_NS, TARGET_KEY); + let intent = optional_read(&store, INTENT_NS, INTENT_KEY); + match mode { + "atomic-write" => assert!( + (target.as_deref() == Some(OLD_VALUE) && intent.is_none()) + || (target.as_deref() == Some(NEW_VALUE) + && intent.as_deref() == Some(WRITE_INTENT)), + "atomic write recovered a split state: target={target:?}, intent={intent:?}" + ), + "atomic-remove" => assert!( + (target.as_deref() == Some(OLD_VALUE) && intent.is_none()) + || (target.is_none() && intent.as_deref() == Some(REMOVE_INTENT)), + "atomic remove recovered a split state: target={target:?}, intent={intent:?}" + ), + "plain-write" => assert!( + matches!(target.as_deref(), Some(OLD_VALUE) | Some(NEW_VALUE)), + "plain write produced an invalid value: {target:?}" + ), + "plain-remove" => assert!( + target.is_none() || target.as_deref() == Some(OLD_VALUE), + "plain remove produced an invalid value: {target:?}" + ), + _ => unreachable!(), + } + } + + #[test] + #[ignore = "subprocess used by kv_store_os_kill_matrix"] + fn kv_store_os_kill_child() { + let mode = std::env::var("RLN_KV_CHILD_MODE").expect("child mode"); + let db_path = PathBuf::from(std::env::var("RLN_KV_CHILD_DB").expect("child DB")); + let store = open_store(&db_path); + match mode.as_str() { + "atomic-write" => store + .write_with_replication_intent( + KvStoreEntry::new(TARGET_NS, "", TARGET_KEY, NEW_VALUE.to_vec()), + KvStoreEntry::new(INTENT_NS, "", INTENT_KEY, WRITE_INTENT.to_vec()), + ) + .expect("atomic write"), + "atomic-remove" => store + .remove_with_replication_intent( + KvStoreKey::new(TARGET_NS, "", TARGET_KEY), + KvStoreEntry::new(INTENT_NS, "", INTENT_KEY, REMOVE_INTENT.to_vec()), + ) + .expect("atomic remove"), + "plain-write" => store + .write(TARGET_NS, "", TARGET_KEY, NEW_VALUE.to_vec()) + .expect("plain write"), + "plain-remove" => store + .remove(TARGET_NS, "", TARGET_KEY, false) + .expect("plain remove"), + _ => panic!("unknown child mode {mode}"), + } + } + + #[test] + fn kv_store_os_kill_matrix() { + for mode in [ + "atomic-write", + "atomic-remove", + "plain-write", + "plain-remove", + ] { + for checkpoint in trace_checkpoints(mode) { + let db_path = kill_at_checkpoint(mode, &checkpoint); + assert_recovered_state(mode, &db_path); + std::fs::remove_dir_all(db_path.parent().expect("DB parent")) + .expect("remove kill fixture"); + } + } + } +} diff --git a/src/synced_kv_store.rs b/src/synced_kv_store.rs index e0c17a2e..86d3f623 100644 --- a/src/synced_kv_store.rs +++ b/src/synced_kv_store.rs @@ -4,13 +4,46 @@ use bitcoin::io; use lightning::util::persist::KVStoreSync; use crate::kv_store::SeaOrmKvStore; +#[cfg(feature = "vss")] +use crate::kv_store::{KvStoreEntry, KvStoreKey}; + +#[cfg(all(test, feature = "vss"))] +fn synced_persistence_checkpoint(name: &str) { + use std::fs::OpenOptions; + use std::io::Write; + + if let Ok(path) = std::env::var("RLN_SYNCED_KV_PERSISTENCE_TRACE_PATH") { + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(path) + .expect("open synchronized KV persistence trace"); + writeln!(file, "{name}").expect("write synchronized KV persistence trace"); + file.sync_all() + .expect("sync synchronized KV persistence trace"); + } + + if std::env::var("RLN_SYNCED_KV_KILL_AT").as_deref() == Ok(name) { + let path = std::env::var("RLN_SYNCED_KV_KILL_READY_PATH") + .expect("synchronized KV persistence kill-ready path"); + let mut file = std::fs::File::create(path).expect("create synchronized KV kill-ready file"); + writeln!(file, "{name}").expect("write synchronized KV kill-ready file"); + file.sync_all() + .expect("sync synchronized KV kill-ready file"); + loop { + std::thread::park(); + } + } +} + +#[cfg(all(not(test), feature = "vss"))] +#[inline] +fn synced_persistence_checkpoint(_name: &str) {} -/// Maximum number of writes the pending-retry queue is allowed to hold. +/// Maximum number of distinct keys the pending-retry queue is allowed to hold. /// -/// When this cap is reached we drop the oldest entry per key (newest write -/// wins for any given key — channel-monitor persistence is last-write-wins -/// anyway). Bounds memory under prolonged VSS outage while still preserving -/// the latest state for each touched key. +/// Capacity is enforced before the local mutation. Entries are never evicted: losing a durable +/// replication intent would make a later VSS restore silently stale. #[cfg(feature = "vss")] const PENDING_QUEUE_CAP: usize = 1000; @@ -18,10 +51,19 @@ const PENDING_QUEUE_CAP: usize = 1000; #[cfg(feature = "vss")] const PENDING_DRAIN_BATCH: usize = 16; -/// Local-only namespace persisting the pending queue across restarts. Rows are -/// written straight to the local store, so they never replicate to VSS. +// Protocol records that must be remotely acknowledged before channel funding may advance. +// These names are persisted storage contracts and intentionally live with the durability policy. #[cfg(feature = "vss")] -const PENDING_NS: &str = "vss_pending"; +pub(crate) const RGB_SENDER_FUNDING_NAMESPACE: &str = "rgb_sender_funding"; +#[cfg(feature = "vss")] +pub(crate) const RGB_PRIMARY_NAMESPACE: &str = "rgb"; +#[cfg(feature = "vss")] +pub(crate) const RGB_FUNDING_ACCEPTANCE_NAMESPACE: &str = "funding_acceptance"; + +/// Local-only namespace persisting the pending queue across restarts. Target mutations and rows in +/// this namespace are committed in one SQLite transaction. +#[cfg(feature = "vss")] +pub(crate) const PENDING_NS: &str = "vss_pending"; /// KVStore wrapper that writes to the local SeaORM store and (optionally) /// replicates to a remote VSS server. Reads always go to the local store for @@ -31,10 +73,8 @@ pub struct SyncedKvStore { local: Arc, #[cfg(feature = "vss")] remote: Option>, - /// VSS-replication failures park here so a transient outage doesn't lose - /// state from the remote backup. Each successful VSS write drains up to - /// [`PENDING_DRAIN_BATCH`] entries; the map is capped at - /// [`PENDING_QUEUE_CAP`]. + /// Every local mutation is represented here before VSS is contacted. Each successful VSS + /// write drains up to [`PENDING_DRAIN_BATCH`] entries. /// /// Key: encoded VSS key string (same format as `vss_key()`). /// Value: bytes that should be re-written to VSS. `None` represents a @@ -42,16 +82,22 @@ pub struct SyncedKvStore { /// removal correctly). #[cfg(feature = "vss")] pending: Arc>>>>, + #[cfg(feature = "vss")] + pending_capacity: usize, /// Serializes VSS puts per key across writes and drains so a drained stale /// value can never land after a newer one. #[cfg(feature = "vss")] key_locks: std::sync::Mutex>>>, - /// Set at teardown; [`Self::stop`] then waits out an in-flight drain via - /// `drain_gate` so no queued put can land after the fence is released. + /// Set at teardown; [`Self::stop`] then waits out every in-flight remote mutation via + /// `drain_gate` so no put or removal can land after the fence is released. #[cfg(feature = "vss")] stopped: std::sync::atomic::AtomicBool, #[cfg(feature = "vss")] drain_gate: std::sync::Mutex<()>, + #[cfg(all(test, feature = "vss"))] + before_drain_gate_hook: std::sync::Mutex>>, + #[cfg(all(test, feature = "vss"))] + before_stop_gate_hook: std::sync::Mutex>>, } impl SyncedKvStore { @@ -64,11 +110,17 @@ impl SyncedKvStore { #[cfg(feature = "vss")] pending: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())), #[cfg(feature = "vss")] + pending_capacity: PENDING_QUEUE_CAP, + #[cfg(feature = "vss")] key_locks: std::sync::Mutex::new(std::collections::HashMap::new()), #[cfg(feature = "vss")] stopped: std::sync::atomic::AtomicBool::new(false), #[cfg(feature = "vss")] drain_gate: std::sync::Mutex::new(()), + #[cfg(all(test, feature = "vss"))] + before_drain_gate_hook: std::sync::Mutex::new(None), + #[cfg(all(test, feature = "vss"))] + before_stop_gate_hook: std::sync::Mutex::new(None), } } @@ -79,6 +131,28 @@ impl SyncedKvStore { local: Arc, remote: Arc, ) -> Self { + Self::with_vss_capacity_inner(local, remote, PENDING_QUEUE_CAP) + } + + #[cfg(all(feature = "vss", test))] + pub(crate) fn with_vss_capacity( + local: Arc, + remote: Arc, + pending_capacity: usize, + ) -> Self { + Self::with_vss_capacity_inner(local, remote, pending_capacity) + } + + #[cfg(feature = "vss")] + fn with_vss_capacity_inner( + local: Arc, + remote: Arc, + pending_capacity: usize, + ) -> Self { + assert!( + pending_capacity > 0, + "pending VSS capacity must be non-zero" + ); let mut pending = std::collections::HashMap::new(); if let Ok(keys) = local.list(PENDING_NS, "") { for key in keys { @@ -90,7 +164,16 @@ impl SyncedKvStore { Some((0, _)) => { pending.insert(key, None); } - _ => tracing::warn!(key, "dropping malformed pending VSS row"), + _ => { + tracing::error!(key, "removing malformed pending VSS row"); + if let Err(error) = local.remove(PENDING_NS, "", &key, false) { + tracing::error!( + key, + error = %error, + "failed to remove malformed pending VSS row" + ); + } + } }, Err(e) => tracing::warn!(key, error = %e, "failed to load pending VSS row"), } @@ -106,9 +189,40 @@ impl SyncedKvStore { local, remote: Some(remote), pending: Arc::new(std::sync::Mutex::new(pending)), + pending_capacity, key_locks: std::sync::Mutex::new(std::collections::HashMap::new()), stopped: std::sync::atomic::AtomicBool::new(false), drain_gate: std::sync::Mutex::new(()), + #[cfg(test)] + before_drain_gate_hook: std::sync::Mutex::new(None), + #[cfg(test)] + before_stop_gate_hook: std::sync::Mutex::new(None), + } + } + + #[cfg(all(test, feature = "vss"))] + pub(crate) fn set_before_drain_gate_hook(&self, hook: Arc) { + *self.before_drain_gate_hook.lock().unwrap() = Some(hook); + } + + #[cfg(all(test, feature = "vss"))] + fn run_before_drain_gate_hook(&self) { + let hook = self.before_drain_gate_hook.lock().unwrap().clone(); + if let Some(hook) = hook { + hook(); + } + } + + #[cfg(all(test, feature = "vss"))] + pub(crate) fn set_before_stop_gate_hook(&self, hook: Arc) { + *self.before_stop_gate_hook.lock().unwrap() = Some(hook); + } + + #[cfg(all(test, feature = "vss"))] + fn run_before_stop_gate_hook(&self) { + let hook = self.before_stop_gate_hook.lock().unwrap().clone(); + if let Some(hook) = hook { + hook(); } } @@ -122,18 +236,19 @@ impl SyncedKvStore { .clone() } - /// Stops future drains and waits for an in-flight one to finish. Call at + /// Stops future replication and waits for every in-flight remote mutation to finish. Call at /// teardown before releasing the VSS fence. #[cfg(feature = "vss")] pub(crate) fn stop(&self) { self.stopped .store(true, std::sync::atomic::Ordering::Release); + #[cfg(test)] + self.run_before_stop_gate_hook(); drop(self.drain_gate.lock().unwrap()); } - /// Persist a queue entry so it survives restarts. #[cfg(feature = "vss")] - fn persist_pending_row(&self, vss_key: &str, value: &Option>) { + fn encode_pending_row(value: &Option>) -> Vec { let mut row = Vec::with_capacity(1 + value.as_ref().map_or(0, |v| v.len())); match value { Some(buf) => { @@ -142,20 +257,68 @@ impl SyncedKvStore { } None => row.push(0), } - if let Err(e) = self.local.write(PENDING_NS, "", vss_key, row) { - tracing::warn!(vss_key, error = %e, "failed to persist pending VSS row"); + row + } + + #[cfg(feature = "vss")] + fn reserve_pending( + &self, + vss_key: &str, + value: Option>, + ) -> Result>>, io::Error> { + let mut pending = self.pending.lock().unwrap(); + if pending.len() >= self.pending_capacity && !pending.contains_key(vss_key) { + tracing::error!( + vss_key, + pending = pending.len(), + cap = self.pending_capacity, + "VSS replication backlog is full; rejecting local mutation" + ); + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + format!( + "VSS replication backlog is full ({} distinct keys)", + self.pending_capacity + ), + )); } + Ok(pending.insert(vss_key.to_owned(), value)) } #[cfg(feature = "vss")] - fn clear_pending_row(&self, vss_key: &str) { - if let Err(e) = self.local.remove(PENDING_NS, "", vss_key, false) { - if e.kind() != io::ErrorKind::NotFound { - tracing::warn!(vss_key, error = %e, "failed to clear pending VSS row"); + fn restore_pending_reservation(&self, vss_key: &str, previous: Option>>) { + let mut pending = self.pending.lock().unwrap(); + match previous { + Some(value) => { + pending.insert(vss_key.to_owned(), value); + } + None => { + pending.remove(vss_key); } } } + #[cfg(feature = "vss")] + fn clear_pending_row(&self, vss_key: &str) -> Result<(), io::Error> { + self.local.remove(PENDING_NS, "", vss_key, false)?; + self.pending.lock().unwrap().remove(vss_key); + Ok(()) + } + + /// These records define channel-funding recovery boundaries. With VSS configured, callers may + /// not advance the protocol after only a local acknowledgement. Other keys remain locally + /// authoritative and use the durable retry queue for eventual VSS convergence. + /// + /// `psbt` and `pending_funding` stay best-effort for now: their current writers unwrap the + /// result, so failing closed would panic the event handler during a VSS outage. They join + /// this set together with the funding state machine that handles the errors. + #[cfg(feature = "vss")] + fn requires_remote_durability(primary_namespace: &str, secondary_namespace: &str) -> bool { + (primary_namespace == RGB_SENDER_FUNDING_NAMESPACE && secondary_namespace.is_empty()) + || (primary_namespace == RGB_PRIMARY_NAMESPACE + && secondary_namespace == RGB_FUNDING_ACCEPTANCE_NAMESPACE) + } + /// Releases the VSS single-writer fence if this instance owns it. No-op /// without a remote store. #[cfg(feature = "vss")] @@ -269,30 +432,6 @@ impl SyncedKvStore { self.pending.lock().unwrap().len() } - /// Enqueue a failed VSS replication for later retry, persisted so it - /// survives restarts. - #[cfg(feature = "vss")] - fn enqueue_pending(&self, vss_key: String, value: Option>) { - self.persist_pending_row(&vss_key, &value); - let mut pending = self.pending.lock().unwrap(); - if pending.len() >= PENDING_QUEUE_CAP && !pending.contains_key(&vss_key) { - // Evict an arbitrary other key to bound the queue; logged so the - // operator can alert on it. - if let Some(evict) = pending.keys().next().cloned() { - pending.remove(&evict); - drop(pending); - self.clear_pending_row(&evict); - pending = self.pending.lock().unwrap(); - tracing::warn!( - evicted_key = evict, - cap = PENDING_QUEUE_CAP, - "VSS pending-writes queue at cap; evicted an entry to make room" - ); - } - } - pending.insert(vss_key, value); - } - /// Attempt to drain up to [`PENDING_DRAIN_BATCH`] entries from the /// pending queue. Called after each successful VSS write and periodically /// from a background task. Entries that re-fail are re-queued. @@ -304,7 +443,15 @@ impl SyncedKvStore { if self.stopped.load(std::sync::atomic::Ordering::Acquire) { return; } + #[cfg(test)] + self.run_before_drain_gate_hook(); let _gate = self.drain_gate.lock().unwrap(); + // A drain may have observed the store as running and then waited behind + // stop(). Re-check under the gate so no remote mutation can begin after + // shutdown has released the VSS fence. + if self.stopped.load(std::sync::atomic::Ordering::Acquire) { + return; + } // Snapshot without removing: entries leave the queue only once VSS // confirms them, so nothing is ever in flight outside the map. let snapshot: Vec<(String, Option>)> = { @@ -330,8 +477,13 @@ impl SyncedKvStore { let parsed = crate::vss_kv_store::parse_vss_key(&vss_key); let Some((primary, secondary, key)) = parsed else { tracing::warn!(vss_key, "Dropping unparseable key from pending queue"); - self.pending.lock().unwrap().remove(&vss_key); - self.clear_pending_row(&vss_key); + if let Err(error) = self.clear_pending_row(&vss_key) { + tracing::error!( + vss_key, + error = %error, + "failed to clear unparseable pending VSS key" + ); + } continue; }; let result = match &value { @@ -339,11 +491,17 @@ impl SyncedKvStore { None => remote.remove(&primary, &secondary, &key, false), }; match result { - Ok(()) => { - self.pending.lock().unwrap().remove(&vss_key); - self.clear_pending_row(&vss_key); - drained += 1; - } + Ok(()) => match self.clear_pending_row(&vss_key) { + Ok(()) => drained += 1, + Err(error) => { + tracing::error!( + vss_key, + error = %error, + "VSS write succeeded but its local retry intent could not be cleared" + ); + break; + } + }, Err(e) => { // VSS is likely still down; the entry stays queued. tracing::debug!( @@ -373,64 +531,110 @@ impl SyncedKvStore { } } } -} - -impl KVStoreSync for SyncedKvStore { - fn read( - &self, - primary_namespace: &str, - secondary_namespace: &str, - key: &str, - ) -> Result, io::Error> { - // Always read from local store - self.local.read(primary_namespace, secondary_namespace, key) - } - fn write( + fn write_with_durability( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, ) -> Result<(), io::Error> { - // Write to local first (must succeed) - self.local - .write(primary_namespace, secondary_namespace, key, buf.clone())?; - - // Replicate to VSS (best-effort, queue for retry on failure). #[cfg(feature = "vss")] if let Some(ref remote) = self.remote { + let drain_gate = self.drain_gate.lock().unwrap(); + if self.stopped.load(std::sync::atomic::Ordering::Acquire) { + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "VSS-synchronized store is stopped", + )); + } let vss_key = crate::vss_kv_store::vss_key(primary_namespace, secondary_namespace, key); - let replicated = { + let remote_required = + Self::requires_remote_durability(primary_namespace, secondary_namespace); + let (replicated, remote_error) = { let lock = self.key_lock(&vss_key); let _guard = lock.lock().unwrap(); + + let pending_value = Some(buf.clone()); + let previous = self.reserve_pending(&vss_key, pending_value.clone())?; + let pending_row = Self::encode_pending_row(&pending_value); + if let Err(error) = self.local.write_with_replication_intent( + KvStoreEntry::new(primary_namespace, secondary_namespace, key, buf.clone()), + KvStoreEntry::new(PENDING_NS, "", &vss_key, pending_row), + ) { + self.restore_pending_reservation(&vss_key, previous); + return Err(error); + } + synced_persistence_checkpoint("synced-write-after-local-commit"); + + synced_persistence_checkpoint("synced-write-before-remote"); match remote.write(primary_namespace, secondary_namespace, key, buf.clone()) { Ok(()) => { - // Drop any stale queued value so a drain can't regress the remote. - if self.pending.lock().unwrap().remove(&vss_key).is_some() { - self.clear_pending_row(&vss_key); + synced_persistence_checkpoint("synced-write-after-remote"); + synced_persistence_checkpoint("synced-write-before-pending-clear"); + if let Err(error) = self.clear_pending_row(&vss_key) { + tracing::error!( + primary_namespace, + secondary_namespace, + key, + error = %error, + "VSS write succeeded but its local retry intent could not be cleared" + ); + (false, None) + } else { + synced_persistence_checkpoint("synced-write-after-pending-clear"); + (true, None) } - true } - Err(e) => { + Err(error) => { tracing::warn!( primary_namespace, secondary_namespace, key, - error = %e, - "VSS replication write failed; queued for retry" + error = %error, + remote_required, + "VSS replication write failed; durable retry intent retained" ); - self.enqueue_pending(vss_key, Some(buf)); - false + (false, Some(error)) } } }; + drop(drain_gate); if replicated { self.drain_pending(); } + if remote_required { + if let Some(error) = remote_error { + return Err(error); + } + } + return Ok(()); } - Ok(()) + self.local + .write(primary_namespace, secondary_namespace, key, buf) + } +} + +impl KVStoreSync for SyncedKvStore { + fn read( + &self, + primary_namespace: &str, + secondary_namespace: &str, + key: &str, + ) -> Result, io::Error> { + // Always read from local store + self.local.read(primary_namespace, secondary_namespace, key) + } + + fn write( + &self, + primary_namespace: &str, + secondary_namespace: &str, + key: &str, + buf: Vec, + ) -> Result<(), io::Error> { + self.write_with_durability(primary_namespace, secondary_namespace, key, buf) } fn remove( @@ -440,24 +644,52 @@ impl KVStoreSync for SyncedKvStore { key: &str, lazy: bool, ) -> Result<(), io::Error> { - // Remove from local first (must succeed) - self.local - .remove(primary_namespace, secondary_namespace, key, lazy)?; - - // Replicate removal to VSS (best-effort, queue for retry on failure). #[cfg(feature = "vss")] if let Some(ref remote) = self.remote { + let drain_gate = self.drain_gate.lock().unwrap(); + if self.stopped.load(std::sync::atomic::Ordering::Acquire) { + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "VSS-synchronized store is stopped", + )); + } let vss_key = crate::vss_kv_store::vss_key(primary_namespace, secondary_namespace, key); - let replicated = { + let remote_required = + Self::requires_remote_durability(primary_namespace, secondary_namespace); + let (replicated, remote_error) = { let lock = self.key_lock(&vss_key); let _guard = lock.lock().unwrap(); + + let pending_value = None; + let previous = self.reserve_pending(&vss_key, pending_value.clone())?; + let pending_row = Self::encode_pending_row(&pending_value); + if let Err(error) = self.local.remove_with_replication_intent( + KvStoreKey::new(primary_namespace, secondary_namespace, key), + KvStoreEntry::new(PENDING_NS, "", &vss_key, pending_row), + ) { + self.restore_pending_reservation(&vss_key, previous); + return Err(error); + } + synced_persistence_checkpoint("synced-remove-after-local-commit"); + + synced_persistence_checkpoint("synced-remove-before-remote"); match remote.remove(primary_namespace, secondary_namespace, key, lazy) { Ok(()) => { - // Same as in `write`. - if self.pending.lock().unwrap().remove(&vss_key).is_some() { - self.clear_pending_row(&vss_key); + synced_persistence_checkpoint("synced-remove-after-remote"); + synced_persistence_checkpoint("synced-remove-before-pending-clear"); + if let Err(error) = self.clear_pending_row(&vss_key) { + tracing::error!( + primary_namespace, + secondary_namespace, + key, + error = %error, + "VSS removal succeeded but its local retry intent could not be cleared" + ); + (false, None) + } else { + synced_persistence_checkpoint("synced-remove-after-pending-clear"); + (true, None) } - true } Err(e) => { tracing::warn!( @@ -465,19 +697,27 @@ impl KVStoreSync for SyncedKvStore { secondary_namespace, key, error = %e, - "VSS replication remove failed; queued for retry" + remote_required, + "VSS replication remove failed; durable retry intent retained" ); - self.enqueue_pending(vss_key, None); - false + (false, Some(e)) } } }; + drop(drain_gate); if replicated { self.drain_pending(); } + if remote_required { + if let Some(error) = remote_error { + return Err(error); + } + } + return Ok(()); } - Ok(()) + self.local + .remove(primary_namespace, secondary_namespace, key, lazy) } fn list( diff --git a/src/test/lib_sdk/vss_manager_lag.rs b/src/test/lib_sdk/vss_manager_lag.rs index b7f68178..b2af0ab8 100644 --- a/src/test/lib_sdk/vss_manager_lag.rs +++ b/src/test/lib_sdk/vss_manager_lag.rs @@ -162,7 +162,6 @@ fn handle_conn( upstream.write_all(new_head.as_bytes())?; upstream.write_all(&body)?; std::io::copy(&mut upstream, &mut client)?; - return Ok(()); } Ok(()) } @@ -178,10 +177,10 @@ fn node_ldk_log_contains(dir: &std::path::Path, needle: &str) -> bool { if node_ldk_log_contains(&path, needle) { return true; } - } else if path.file_name().is_some_and(|n| n == "logs.txt") { - if fs::read_to_string(&path).is_ok_and(|c| c.contains(needle)) { - return true; - } + } else if path.file_name().is_some_and(|n| n == "logs.txt") + && fs::read_to_string(&path).is_ok_and(|c| c.contains(needle)) + { + return true; } } false @@ -267,7 +266,7 @@ fn setup_with_open_channel(test_name: &str) -> LagSetup { fee_base_msat: None, fee_proportional_millionths: None, temporary_channel_id: None, - asset_id: Some(asset_id.clone()), + asset_id: Some(asset_id), asset_amount: Some(600), push_asset_amount: None, virtual_open_mode: None, diff --git a/src/test/mod.rs b/src/test/mod.rs index 84388ce1..e9ae664c 100644 --- a/src/test/mod.rs +++ b/src/test/mod.rs @@ -510,17 +510,36 @@ async fn clear_vss_fence(node_address: SocketAddr, password: &str) { let payload = VssClearFenceRequest { password: password.to_string(), }; - let res = reqwest::Client::new() - .post(format!("http://{node_address}/vssclearfence")) - .json(&payload) - .send() - .await - .unwrap(); - check_response_is_ok(res) - .await - .json::() - .await - .unwrap(); + let client = reqwest::Client::new(); + let mut last_error = String::new(); + + for attempt in 1..=20 { + match client + .post(format!("http://{node_address}/vssclearfence")) + .json(&payload) + .send() + .await + { + Ok(response) if response.status().is_success() => { + response.json::().await.unwrap(); + return; + } + Ok(response) => { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + last_error = format!("HTTP {status}: {body}"); + } + Err(error) => { + last_error = error.to_string(); + } + } + + if attempt < 20 { + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + } + } + + panic!("VSS fence clear did not become ready: {last_error}"); } /// Gracefully stop a node and wipe its entire storage dir (DB, channel state, @@ -3125,6 +3144,8 @@ mod virtual_channels; #[cfg(feature = "vss")] mod vss; #[cfg(feature = "vss")] +mod vss_durability_gaps; +#[cfg(feature = "vss")] mod vss_offline_force_close; #[cfg(feature = "vss")] mod vss_unreachable_openchannel; diff --git a/src/test/vss.rs b/src/test/vss.rs index ab7e5f4b..88330312 100644 --- a/src/test/vss.rs +++ b/src/test/vss.rs @@ -5,15 +5,17 @@ #[cfg(feature = "vss")] mod tests { + use std::path::{Path, PathBuf}; + use std::process::{Command, Stdio}; use std::sync::Arc; - use std::time::Duration; + use std::time::{Duration, Instant}; use bitcoin::secp256k1::{rand::rngs::OsRng, Secp256k1, SecretKey}; use hex::DisplayHex; use lightning::util::persist::KVStoreSync; - use sea_orm::{ConnectOptions, Database}; + use sea_orm::{ConnectOptions, ConnectionTrait, Database, DatabaseBackend, Statement}; - use crate::kv_store::SeaOrmKvStore; + use crate::kv_store::{KvStoreEntry, KvStoreKey, SeaOrmKvStore}; use crate::synced_kv_store::SyncedKvStore; use crate::vss_kv_store::{parse_vss_key, vss_key, VssKvStore}; @@ -49,6 +51,143 @@ mod tests { Arc::new(db) } + fn open_test_sqlite(path: &Path) -> Arc { + use rln_migration::MigratorTrait; + + let conn_str = format!("sqlite:{}?mode=rwc", path.display()); + let mut opt = ConnectOptions::new(conn_str); + opt.max_connections(1) + .connect_timeout(Duration::from_secs(5)); + let db = crate::runtime::block_on(Database::connect(opt)).expect("persistent test DB"); + crate::runtime::block_on(rln_migration::Migrator::up(&db, None)).expect("migration"); + Arc::new(db) + } + + fn synced_kv_child_command( + mode: &str, + db_path: &Path, + store_id: &str, + signing_key: &SecretKey, + ) -> Command { + let mut command = Command::new(std::env::current_exe().expect("current test executable")); + command + .args([ + "--exact", + "test::vss::tests::synced_kv_os_kill_child", + "--ignored", + "--nocapture", + ]) + .env("RLN_SYNCED_KV_CHILD_MODE", mode) + .env("RLN_SYNCED_KV_CHILD_DB", db_path) + .env("RLN_SYNCED_KV_CHILD_STORE_ID", store_id) + .env( + "RLN_SYNCED_KV_CHILD_SECRET", + signing_key.secret_bytes().as_hex().to_string(), + ) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + command + } + + fn prepare_synced_kv_fixture(db_path: &Path, store_id: &str, signing_key: SecretKey) { + let local = SeaOrmKvStore::from_connection(open_test_sqlite(db_path)); + local + .write( + crate::synced_kv_store::RGB_SENDER_FUNDING_NAMESPACE, + "", + "funding", + b"old".to_vec(), + ) + .expect("seed local value"); + let remote = VssKvStore::new(VSS_URL.to_owned(), store_id.to_owned(), signing_key) + .expect("seed VSS store"); + remote + .write( + crate::synced_kv_store::RGB_SENDER_FUNDING_NAMESPACE, + "", + "funding", + b"old".to_vec(), + ) + .expect("seed remote value"); + } + + fn trace_synced_kv_checkpoints(mode: &str) -> Vec { + let directory = tempfile::tempdir().expect("trace tempdir"); + let db_path = directory.path().join("store.sqlite"); + let trace_path = directory.path().join("trace.txt"); + let (signing_key, store_id) = generate_test_keys(); + prepare_synced_kv_fixture(&db_path, &store_id, signing_key); + let status = synced_kv_child_command(mode, &db_path, &store_id, &signing_key) + .env("RLN_SYNCED_KV_PERSISTENCE_TRACE_PATH", &trace_path) + .status() + .expect("run synchronized KV trace child"); + assert!( + status.success(), + "synchronized KV trace child failed for {mode}" + ); + let prefix = if mode == "write" { + "synced-write-" + } else { + "synced-remove-" + }; + let checkpoints: Vec<_> = std::fs::read_to_string(trace_path) + .expect("read synchronized KV trace") + .lines() + .filter(|line| line.starts_with(prefix)) + .map(str::to_owned) + .collect(); + assert!( + !checkpoints.is_empty(), + "no synchronized KV checkpoints for {mode}" + ); + let mut unique = checkpoints.clone(); + unique.sort(); + unique.dedup(); + assert_eq!( + checkpoints.len(), + unique.len(), + "duplicate checkpoints for {mode}" + ); + checkpoints + } + + fn kill_synced_kv_at_checkpoint(mode: &str, checkpoint: &str) -> (PathBuf, String, SecretKey) { + let directory = tempfile::tempdir().expect("kill tempdir"); + let root = directory.keep(); + let db_path = root.join("store.sqlite"); + let ready_path = root.join("ready"); + let (signing_key, store_id) = generate_test_keys(); + prepare_synced_kv_fixture(&db_path, &store_id, signing_key); + let mut child = synced_kv_child_command(mode, &db_path, &store_id, &signing_key) + .env("RLN_SYNCED_KV_KILL_AT", checkpoint) + .env("RLN_SYNCED_KV_KILL_READY_PATH", &ready_path) + .spawn() + .expect("spawn synchronized KV kill child"); + + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if ready_path.exists() { + break; + } + if let Some(status) = child.try_wait().expect("poll synchronized KV child") { + panic!("synchronized KV child exited before {checkpoint}: {status}"); + } + if Instant::now() >= deadline { + let _ = child.kill(); + let _ = child.wait(); + panic!("synchronized KV child did not reach {checkpoint}"); + } + std::thread::sleep(Duration::from_millis(10)); + } + child.kill().expect("kill synchronized KV child"); + let status = child.wait().expect("wait for synchronized KV child"); + assert!( + !status.success(), + "killed synchronized KV child unexpectedly succeeded" + ); + (db_path, store_id, signing_key) + } + // --- Unit tests --- #[test] @@ -634,6 +773,29 @@ mod tests { /// A failed unlock must roll back what it acquired: the VSS fence is /// released and the changing-state flag is cleared, so a retry (failed or /// successful) is never wedged behind a stranded fence. + fn select_electrum_backend(payload: &mut crate::routes::UnlockRequest) { + payload.bitcoind_rpc_username = None; + payload.bitcoind_rpc_password = None; + payload.bitcoind_rpc_host = None; + payload.bitcoind_rpc_port = None; + } + + async fn unlock_with_electrum_backend(node_address: std::net::SocketAddr, password: &str) { + let mut payload = crate::test::unlock_req(password); + select_electrum_backend(&mut payload); + let res = reqwest::Client::new() + .post(format!("http://{node_address}/unlock")) + .json(&payload) + .send() + .await + .unwrap(); + crate::test::check_response_is_ok(res) + .await + .json::() + .await + .unwrap(); + } + #[serial_test::serial] #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn vss_failed_unlock_releases_fence_and_allows_retry() { @@ -666,6 +828,7 @@ mod tests { // Unlock against an unreachable indexer: fails after the fence acquire. let mut payload = crate::test::unlock_req(password); + select_electrum_backend(&mut payload); payload.indexer_url = Some("127.0.0.1:1".to_string()); let client = reqwest::Client::new(); for attempt in 1..=2 { @@ -693,7 +856,7 @@ mod tests { } // With the fence rolled back a retry with good parameters succeeds. - crate::test::unlock(node_address, password).await; + unlock_with_electrum_backend(node_address, password).await; crate::test::shutdown(&[node_address]).await; } @@ -730,9 +893,9 @@ mod tests { let password = "vss_lock_release_fence"; crate::test::init(node_address, password, None).await; - crate::test::unlock(node_address, password).await; + unlock_with_electrum_backend(node_address, password).await; crate::test::lock(node_address).await; - crate::test::unlock(node_address, password).await; + unlock_with_electrum_backend(node_address, password).await; crate::test::shutdown(&[node_address]).await; } @@ -1233,4 +1396,242 @@ mod tests { let reopened = SyncedKvStore::with_vss(Arc::clone(&local), Arc::clone(&remote)); assert_eq!(reopened.pending_remote_writes(), 0); } + + #[test] + fn local_mutation_and_replication_intent_are_one_sqlite_transaction() { + let connection = create_test_sqlite(); + let local = SeaOrmKvStore::from_connection(Arc::clone(&connection)); + local + .write("target", "", "write-key", b"old".to_vec()) + .expect("seed write target"); + local + .write("target", "", "remove-key", b"retained".to_vec()) + .expect("seed remove target"); + + crate::runtime::block_on(connection.execute(Statement::from_string( + DatabaseBackend::Sqlite, + format!( + "CREATE TRIGGER reject_vss_intent BEFORE INSERT ON kv_store \ + WHEN NEW.primary_namespace = '{}' \ + BEGIN SELECT RAISE(ABORT, 'injected intent failure'); END", + crate::synced_kv_store::PENDING_NS + ), + ))) + .expect("install failure trigger"); + + assert!(local + .write_with_replication_intent( + KvStoreEntry::new("target", "", "write-key", b"new".to_vec()), + KvStoreEntry::new( + crate::synced_kv_store::PENDING_NS, + "", + "target//write-key", + vec![1, 0x01], + ), + ) + .is_err()); + assert_eq!( + local + .read("target", "", "write-key") + .expect("old value survives"), + b"old" + ); + assert!( + local + .read(crate::synced_kv_store::PENDING_NS, "", "target//write-key") + .is_err(), + "failed intent insertion must roll back the target write" + ); + + assert!(local + .remove_with_replication_intent( + KvStoreKey::new("target", "", "remove-key"), + KvStoreEntry::new( + crate::synced_kv_store::PENDING_NS, + "", + "target//remove-key", + vec![0], + ), + ) + .is_err()); + assert_eq!( + local + .read("target", "", "remove-key") + .expect("removed value survives rollback"), + b"retained" + ); + } + + #[test] + fn full_pending_queue_rejects_before_local_mutation_without_eviction() { + let (signing_key, store_id) = generate_test_keys(); + let connection = create_test_sqlite(); + let local = Arc::new(SeaOrmKvStore::from_connection(connection)); + let existing_vss_key = vss_key("existing", "", "key"); + let mut existing_row = vec![1]; + existing_row.extend_from_slice(b"pending"); + local + .write( + crate::synced_kv_store::PENDING_NS, + "", + &existing_vss_key, + existing_row, + ) + .expect("seed pending intent"); + let unreachable = Arc::new( + VssKvStore::new("http://127.0.0.1:5/vss".to_string(), store_id, signing_key) + .expect("vss store"), + ); + let synced = SyncedKvStore::with_vss_capacity(Arc::clone(&local), unreachable, 1); + + let error = synced + .write("new", "", "key", b"must-not-land".to_vec()) + .expect_err("a full backlog must apply backpressure"); + assert_eq!(error.kind(), bitcoin::io::ErrorKind::WouldBlock); + assert!(local.read("new", "", "key").is_err()); + assert_eq!(synced.pending_remote_writes(), 1); + assert_eq!( + local + .read(crate::synced_kv_store::PENDING_NS, "", &existing_vss_key) + .expect("existing intent must not be evicted"), + [vec![1], b"pending".to_vec()].concat() + ); + } + + #[test] + fn funding_journal_write_fails_closed_and_remains_replayable() { + let (signing_key, store_id) = generate_test_keys(); + let connection = create_test_sqlite(); + let local = Arc::new(SeaOrmKvStore::from_connection(Arc::clone(&connection))); + let unreachable = Arc::new( + VssKvStore::new("http://127.0.0.1:5/vss".to_string(), store_id, signing_key) + .expect("vss store"), + ); + let synced = SyncedKvStore::with_vss(Arc::clone(&local), Arc::clone(&unreachable)); + let journal = b"funding-journal".to_vec(); + + synced + .write( + crate::synced_kv_store::RGB_PRIMARY_NAMESPACE, + crate::synced_kv_store::RGB_FUNDING_ACCEPTANCE_NAMESPACE, + "temporary-channel", + journal.clone(), + ) + .expect_err("critical funding state must not acknowledge a local-only write"); + assert_eq!( + synced + .read( + crate::synced_kv_store::RGB_PRIMARY_NAMESPACE, + crate::synced_kv_store::RGB_FUNDING_ACCEPTANCE_NAMESPACE, + "temporary-channel" + ) + .expect("local recovery state"), + journal + ); + assert_eq!(synced.pending_remote_writes(), 1); + drop(synced); + + let reopened = SyncedKvStore::with_vss( + Arc::new(SeaOrmKvStore::from_connection(connection)), + unreachable, + ); + assert_eq!( + reopened.pending_remote_writes(), + 1, + "restart must recover the failed critical replication" + ); + } + + #[test] + #[ignore = "subprocess used by synced_kv_os_kill_matrix"] + fn synced_kv_os_kill_child() { + let mode = std::env::var("RLN_SYNCED_KV_CHILD_MODE").expect("child mode"); + let db_path = + PathBuf::from(std::env::var("RLN_SYNCED_KV_CHILD_DB").expect("child database path")); + let store_id = std::env::var("RLN_SYNCED_KV_CHILD_STORE_ID").expect("child VSS store ID"); + let secret = std::env::var("RLN_SYNCED_KV_CHILD_SECRET").expect("child VSS secret"); + let signing_key = secret.parse::().expect("parse child VSS secret"); + let local = Arc::new(SeaOrmKvStore::from_connection(open_test_sqlite(&db_path))); + let remote = Arc::new( + VssKvStore::new(VSS_URL.to_owned(), store_id, signing_key).expect("child VSS store"), + ); + let synced = SyncedKvStore::with_vss(local, remote); + match mode.as_str() { + "write" => synced + .write( + crate::synced_kv_store::RGB_SENDER_FUNDING_NAMESPACE, + "", + "funding", + b"new".to_vec(), + ) + .expect("synchronized write"), + "remove" => synced + .remove( + crate::synced_kv_store::RGB_SENDER_FUNDING_NAMESPACE, + "", + "funding", + false, + ) + .expect("synchronized remove"), + _ => panic!("unknown child mode {mode}"), + } + } + + #[test] + #[ignore = "requires the VSS integration service"] + fn synced_kv_os_kill_matrix() { + assert!( + vss_server_available(), + "VSS server is required at {VSS_URL}" + ); + for mode in ["write", "remove"] { + for checkpoint in trace_synced_kv_checkpoints(mode) { + let (db_path, store_id, signing_key) = + kill_synced_kv_at_checkpoint(mode, &checkpoint); + let local = Arc::new(SeaOrmKvStore::from_connection(open_test_sqlite(&db_path))); + let remote = Arc::new( + VssKvStore::new(VSS_URL.to_owned(), store_id, signing_key) + .expect("reopen VSS store"), + ); + let synced = SyncedKvStore::with_vss(Arc::clone(&local), Arc::clone(&remote)); + let deadline = Instant::now() + Duration::from_secs(20); + while synced.pending_remote_writes() != 0 && Instant::now() < deadline { + synced.drain_pending(); + std::thread::sleep(Duration::from_millis(20)); + } + assert_eq!( + synced.pending_remote_writes(), + 0, + "pending replication did not drain after {checkpoint}" + ); + + let local_value = local.read( + crate::synced_kv_store::RGB_SENDER_FUNDING_NAMESPACE, + "", + "funding", + ); + let remote_value = remote.read( + crate::synced_kv_store::RGB_SENDER_FUNDING_NAMESPACE, + "", + "funding", + ); + if mode == "write" { + assert_eq!(local_value.expect("local write"), b"new".to_vec()); + assert_eq!(remote_value.expect("remote write"), b"new".to_vec()); + } else { + assert_eq!( + local_value.expect_err("local removal").kind(), + bitcoin::io::ErrorKind::NotFound + ); + assert_eq!( + remote_value.expect_err("remote removal").kind(), + bitcoin::io::ErrorKind::NotFound + ); + } + drop(synced); + std::fs::remove_dir_all(db_path.parent().expect("database parent")) + .expect("remove synchronized KV fixture"); + } + } + } } diff --git a/src/test/vss_durability_gaps.rs b/src/test/vss_durability_gaps.rs new file mode 100644 index 00000000..ef2be1aa --- /dev/null +++ b/src/test/vss_durability_gaps.rs @@ -0,0 +1,383 @@ +//! Deterministic tests for the VSS durability contract of the best-effort +//! replication path (`SyncedKvStore`). +//! +//! Each test encodes an invariant a crash/device-loss recovery relies on. +//! They are expected to fail on current `dev` and pass once the persistence +//! layer is made crash-consistent. + +#[cfg(feature = "vss")] +mod tests { + use std::io::Read; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{mpsc, Arc}; + use std::time::Duration; + + use bitcoin::secp256k1::{rand::rngs::OsRng, Secp256k1, SecretKey}; + use hex::DisplayHex; + use lightning::util::persist::KVStoreSync; + use sea_orm::{ConnectOptions, Database}; + + use crate::kv_store::SeaOrmKvStore; + use crate::synced_kv_store::SyncedKvStore; + use crate::vss_kv_store::{vss_key, VssKvStore}; + + /// Same persisted contract as `SyncedKvStore`'s pending-retry namespace. + const PENDING_NS: &str = "vss_pending"; + const PENDING_QUEUE_CAP: usize = 1000; + + fn generate_test_keys() -> (SecretKey, String) { + let secp = Secp256k1::new(); + let (secret_key, public_key) = secp.generate_keypair(&mut OsRng); + let store_id = format!("rln_test_{}", public_key.serialize()[0..8].as_hex()); + (secret_key, store_id) + } + + fn open_sqlite(dir: &std::path::Path) -> Arc { + use rln_migration::MigratorTrait; + + let db_path = dir.join("test_rln_db"); + let conn_str = format!("sqlite:{}?mode=rwc", db_path.display()); + let mut opt = ConnectOptions::new(conn_str); + opt.max_connections(1) + .connect_timeout(Duration::from_secs(5)); + let db = crate::runtime::block_on(Database::connect(opt)).expect("test db"); + crate::runtime::block_on(rln_migration::Migrator::up(&db, None)).expect("migration"); + Arc::new(db) + } + + fn unreachable_vss() -> Arc { + let (signing_key, store_id) = generate_test_keys(); + Arc::new( + VssKvStore::new("http://127.0.0.1:5/vss".to_string(), store_id, signing_key) + .expect("vss store"), + ) + } + + /// TCP server that accepts VSS requests and holds them open (no response) + /// until `cut()`, which drops every connection and starts refusing new + /// ones. Models a remote mutation that is in flight at crash/shutdown + /// time: sent, not acknowledged. + struct StallingVss { + port: u16, + arrived: mpsc::Receiver<()>, + cut: Arc, + } + + impl StallingVss { + fn start() -> Self { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let (tx, arrived) = mpsc::channel(); + let cut = Arc::new(AtomicBool::new(false)); + let cut_bg = Arc::clone(&cut); + std::thread::spawn(move || { + let mut held: Vec = Vec::new(); + listener + .set_nonblocking(true) + .expect("nonblocking listener"); + loop { + if cut_bg.load(Ordering::Acquire) { + return; // drops listener and held streams + } + match listener.accept() { + Ok((mut stream, _)) => { + stream + .set_read_timeout(Some(Duration::from_millis(500))) + .ok(); + let mut buf = [0u8; 512]; + if stream.read(&mut buf).is_ok() { + let _ = tx.send(()); + } + held.push(stream); + } + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(20)); + } + Err(_) => return, + } + } + }); + Self { port, arrived, cut } + } + + fn url(&self) -> String { + format!("http://127.0.0.1:{}/vss", self.port) + } + + fn wait_for_request(&self) { + self.arrived + .recv_timeout(Duration::from_secs(20)) + .expect("VSS request must arrive"); + } + + fn try_request(&self) -> Result { + match self.arrived.try_recv() { + Ok(()) => Ok(true), + Err(mpsc::TryRecvError::Empty) => Ok(false), + Err(error) => Err(error), + } + } + + fn cut(&self) { + self.cut.store(true, Ordering::Release); + } + } + + /// Copy the SQLite files (db + wal + shm) while the node is "running". + /// The copy is the exact on-disk image an OS kill at that instant leaves + /// behind. + fn snapshot_sqlite(dir: &std::path::Path, dest: &std::path::Path) { + std::fs::create_dir_all(dest).expect("snapshot dir"); + for suffix in ["", "-wal", "-shm"] { + let src = dir.join(format!("test_rln_db{suffix}")); + if src.exists() { + std::fs::copy(&src, dest.join(format!("test_rln_db{suffix}"))).expect("copy"); + } + } + } + + /// A killed process must observe either the complete old state or the + /// complete new local value plus its durable retry intent. On `dev` the + /// local value commits first and the retry intent is only persisted after + /// the remote attempt fails, so a kill while the VSS request is in flight + /// leaves a crash image whose value will never be replicated: a later + /// device-loss restore is silently stale. + #[test] + fn crash_image_must_retain_replication_intent() { + let dir = tempfile::tempdir().expect("tempdir").keep(); + let local = Arc::new(SeaOrmKvStore::from_connection(open_sqlite(&dir))); + let stall = StallingVss::start(); + let (signing_key, store_id) = generate_test_keys(); + let remote = + Arc::new(VssKvStore::new(stall.url(), store_id, signing_key).expect("vss store")); + let synced = Arc::new(SyncedKvStore::with_vss(Arc::clone(&local), remote)); + + let writer = { + let synced = Arc::clone(&synced); + std::thread::spawn(move || synced.write("", "", "aux_state", b"v1".to_vec())) + }; + + // The remote put is in flight: VSS has acknowledged nothing. Snapshot + // the exact on-disk image an OS kill leaves now. + stall.wait_for_request(); + let snapshot_dir = dir.join("crash_image"); + snapshot_sqlite(&dir, &snapshot_dir); + + stall.cut(); + let _ = writer.join().expect("writer thread"); + + // "Restart" from the crash image. Complete old state is fine; the new + // local value without a surviving retry intent is not. + let restarted_local = Arc::new(SeaOrmKvStore::from_connection(open_sqlite(&snapshot_dir))); + let restarted = SyncedKvStore::with_vss(Arc::clone(&restarted_local), unreachable_vss()); + + if let Ok(value) = restarted_local.read("", "", "aux_state") { + assert_eq!(value, b"v1".to_vec()); + assert_eq!( + restarted.pending_remote_writes(), + 1, + "crash image holds a local value VSS never acknowledged, but no \ + retry intent survived: the value will never replicate and a \ + device-loss restore is silently stale" + ); + } + } + + /// A full retry backlog must never discard recovery evidence: every local + /// mutation VSS has not acknowledged needs a durable retry intent. On + /// `dev` a new distinct mutation at cap evicts an arbitrary queued entry, + /// so that entry's key silently stops replicating. + #[test] + fn pending_queue_cap_must_not_discard_recovery_evidence() { + let dir = tempfile::tempdir().expect("tempdir").keep(); + let local = Arc::new(SeaOrmKvStore::from_connection(open_sqlite(&dir))); + + // Backlog accumulated by a previous run during a VSS outage, in the + // persisted pending-row format (0x01 prefix = queued write). + for i in 0..PENDING_QUEUE_CAP { + let key = vss_key("", "", &format!("backlog_{i}")); + local + .write("", "", &format!("backlog_{i}"), vec![0xAA]) + .expect("value"); + local + .write(PENDING_NS, "", &key, vec![1, 0xAA]) + .expect("pending row"); + } + + let synced = SyncedKvStore::with_vss(Arc::clone(&local), unreachable_vss()); + assert_eq!(synced.pending_remote_writes(), PENDING_QUEUE_CAP); + + let result = synced.write("", "", "new_mutation", b"fresh".to_vec()); + + let durable_intents = local.list(PENDING_NS, "").expect("list pending").len(); + match result { + Ok(()) => { + assert_eq!( + durable_intents, + PENDING_QUEUE_CAP + 1, + "the mutation was accepted at cap, so recovery evidence \ + was evicted: one key will never replicate to VSS" + ); + } + Err(_) => { + // Fail-closed alternative: the mutation is rejected whole. + assert!( + local.read("", "", "new_mutation").is_err(), + "a rejected mutation must not commit locally" + ); + assert_eq!( + durable_intents, PENDING_QUEUE_CAP, + "backlog must stay intact" + ); + } + } + } + + /// When `stop()` returns the caller releases the VSS single-writer fence, + /// so no remote mutation may still be in flight. Event-ordered: a correct + /// `stop()` blocks on the drain gate the in-flight write holds and can + /// only return after the connection is cut; a `stop()` that ignores the + /// in-flight put acquires the free gate and returns inside the + /// observation window. + #[test] + fn stop_must_wait_for_inflight_remote_mutation() { + let dir = tempfile::tempdir().expect("tempdir").keep(); + let local = Arc::new(SeaOrmKvStore::from_connection(open_sqlite(&dir))); + let stall = StallingVss::start(); + let (signing_key, store_id) = generate_test_keys(); + let remote = + Arc::new(VssKvStore::new(stall.url(), store_id, signing_key).expect("vss store")); + let synced = Arc::new(SyncedKvStore::with_vss(local, remote)); + + let writer = { + let synced = Arc::clone(&synced); + std::thread::spawn(move || synced.write("", "", "inflight", b"v1".to_vec())) + }; + stall.wait_for_request(); + + let (at_gate_tx, at_gate_rx) = mpsc::channel(); + synced.set_before_stop_gate_hook(Arc::new(move || { + let _ = at_gate_tx.send(()); + })); + let (stop_done_tx, stop_done_rx) = mpsc::channel(); + let stopper = { + let synced = Arc::clone(&synced); + std::thread::spawn(move || { + synced.stop(); + let _ = stop_done_tx.send(()); + }) + }; + + at_gate_rx + .recv_timeout(Duration::from_secs(10)) + .expect("stop must reach the gate"); + // A stop() blocked on the held gate cannot complete until the cut + // below, so this window can only ever observe a stop() that skipped + // the in-flight put; the wait it bounds is an uncontended lock plus a + // channel send. + let stopped_early = stop_done_rx.recv_timeout(Duration::from_secs(10)).is_ok(); + + stall.cut(); + stopper.join().expect("stopper thread"); + let _ = writer.join().expect("writer thread"); + + assert!( + !stopped_early, + "stop() returned while a remote mutation was in flight: the VSS \ + fence can be released (and re-acquired by another instance) \ + before this put lands, breaking single-writer" + ); + } + + /// A retry drain that passed its initial admission check before shutdown + /// must not start a remote mutation after `stop()` has returned. + #[test] + fn queued_drain_must_not_run_after_stop() { + let dir = tempfile::tempdir().expect("tempdir").keep(); + let local = Arc::new(SeaOrmKvStore::from_connection(open_sqlite(&dir))); + let pending_key = vss_key("", "", "queued"); + local + .write("", "", "queued", b"v1".to_vec()) + .expect("value"); + local + .write(PENDING_NS, "", &pending_key, vec![1, b'v', b'1']) + .expect("pending row"); + + let stall = StallingVss::start(); + let (signing_key, store_id) = generate_test_keys(); + let remote = + Arc::new(VssKvStore::new(stall.url(), store_id, signing_key).expect("vss store")); + let synced = Arc::new(SyncedKvStore::with_vss(local, remote)); + + let (at_gate_tx, at_gate_rx) = mpsc::channel(); + let (resume_tx, resume_rx) = mpsc::channel(); + let resume_rx = std::sync::Mutex::new(resume_rx); + synced.set_before_drain_gate_hook(Arc::new(move || { + at_gate_tx.send(()).expect("signal drain admission"); + resume_rx.lock().unwrap().recv().expect("resume drain"); + })); + + let (drain_done_tx, drain_done_rx) = mpsc::channel(); + let drainer = { + let synced = Arc::clone(&synced); + std::thread::spawn(move || { + synced.drain_pending(); + drain_done_tx.send(()).expect("signal drain completion"); + }) + }; + at_gate_rx + .recv_timeout(Duration::from_secs(5)) + .expect("drain must reach the shutdown gate"); + + synced.stop(); + resume_tx.send(()).expect("resume queued drain"); + + let deadline = std::time::Instant::now() + Duration::from_secs(20); + let attempted_remote_write = loop { + if stall + .try_request() + .expect("VSS listener must stay available") + { + break true; + } + match drain_done_rx.try_recv() { + Ok(()) => break false, + Err(mpsc::TryRecvError::Empty) => {} + Err(mpsc::TryRecvError::Disconnected) => { + panic!("drain thread exited without reporting completion") + } + } + assert!( + std::time::Instant::now() < deadline, + "queued drain produced neither completion nor remote I/O" + ); + std::thread::sleep(Duration::from_millis(10)); + }; + stall.cut(); + drainer.join().expect("drainer thread"); + + assert!( + !attempted_remote_write, + "a queued retry drain started a remote mutation after stop() returned" + ); + assert_eq!(synced.pending_remote_writes(), 1); + } + + /// `psbt` and `pending_funding` writers currently unwrap the write result, + /// so these namespaces must keep acking during a VSS outage until the + /// funding state machine handles the errors. + #[test] + fn psbt_and_pending_funding_stay_best_effort_during_outage() { + let dir = tempfile::tempdir().expect("tempdir").keep(); + let local = Arc::new(SeaOrmKvStore::from_connection(open_sqlite(&dir))); + let synced = SyncedKvStore::with_vss(local, unreachable_vss()); + + synced + .write("psbt", "", "funding_txid", b"psbt".to_vec()) + .expect("psbt write must ack during an outage"); + synced + .write("pending_funding", "", "channel_id", b"txid".to_vec()) + .expect("pending_funding write must ack during an outage"); + assert_eq!(synced.pending_remote_writes(), 2); + } +} diff --git a/src/test/vss_offline_force_close.rs b/src/test/vss_offline_force_close.rs index b9ea1b72..28ff19c1 100644 --- a/src/test/vss_offline_force_close.rs +++ b/src/test/vss_offline_force_close.rs @@ -1,25 +1,41 @@ use super::*; +use axum::response::IntoResponse; const TEST_DIR_BASE: &str = "tmp/vss_offline_force_close/"; -const VSS_SERVER_ADDR: &str = "127.0.0.1:8081"; +const VSS_SERVER_URL: &str = "http://127.0.0.1:8081"; -/// TCP proxy in front of the VSS server that can go offline and back online, -/// simulating a VSS outage for a single node without touching the shared -/// service. Offline: established connections are cut and new ones are closed -/// on accept. +#[derive(Clone)] +struct VssProxyState { + online: Arc, + client: reqwest::Client, +} + +/// Reverse proxy in front of the VSS server that can go offline and back +/// online for one node without changing the shared test service. pub(super) struct VssProxy { port: u16, - online: tokio::sync::watch::Sender, + online: Arc, + shutdown: Option>, } impl VssProxy { - // The proxy gets a dedicated thread + runtime: the shared test runtime can - // stall for seconds on synchronous KVStore calls and would starve it. pub(super) fn start() -> Self { let std_listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); std_listener.set_nonblocking(true).unwrap(); let port = std_listener.local_addr().unwrap().port(); - let (online, rx) = tokio::sync::watch::channel(true); + let online = Arc::new(std::sync::atomic::AtomicBool::new(true)); + let state = VssProxyState { + online: Arc::clone(&online), + client: reqwest::Client::builder() + .no_proxy() + .build() + .expect("VSS fault proxy client must be constructible"), + }; + let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel(1); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel(); + + // Synchronous KVStore calls can stall the shared test runtime, so the + // fault proxy owns a dedicated runtime. std::thread::spawn(move || { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() @@ -27,27 +43,27 @@ impl VssProxy { .unwrap(); rt.block_on(async move { let listener = TcpListener::from_std(std_listener).unwrap(); - loop { - let Ok((mut inbound, _)) = listener.accept().await else { - break; - }; - if !*rx.borrow() { - continue; // offline: drop the connection immediately - } - let mut rx_conn = rx.clone(); - tokio::spawn(async move { - let Ok(mut outbound) = TcpStream::connect(VSS_SERVER_ADDR).await else { - return; - }; - tokio::select! { - _ = tokio::io::copy_bidirectional(&mut inbound, &mut outbound) => {} - _ = rx_conn.wait_for(|online| !online) => {} - } - }); - } + let app = axum::Router::new() + .fallback(axum::routing::any(forward_vss_request)) + .with_state(state); + ready_tx.send(()).unwrap(); + axum::serve(listener, app) + .with_graceful_shutdown(async { + let _ = shutdown_rx.await; + }) + .await + .unwrap(); }); }); - Self { port, online } + ready_rx + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("VSS fault proxy did not become ready"); + + Self { + port, + online, + shutdown: Some(shutdown_tx), + } } pub(super) fn url(&self) -> String { @@ -55,11 +71,81 @@ impl VssProxy { } pub(super) fn go_offline(&self) { - self.online.send(false).unwrap(); + self.online + .store(false, std::sync::atomic::Ordering::Release); } pub(super) fn go_online(&self) { - self.online.send(true).unwrap(); + self.online + .store(true, std::sync::atomic::Ordering::Release); + } +} + +impl Drop for VssProxy { + fn drop(&mut self) { + if let Some(shutdown) = self.shutdown.take() { + let _ = shutdown.send(()); + } + } +} + +async fn forward_vss_request( + axum::extract::State(state): axum::extract::State, + method: axum::http::Method, + uri: axum::http::Uri, + mut headers: axum::http::HeaderMap, + body: axum::body::Bytes, +) -> axum::response::Response { + if !state.online.load(std::sync::atomic::Ordering::Acquire) { + return ( + axum::http::StatusCode::SERVICE_UNAVAILABLE, + "VSS fault proxy is offline", + ) + .into_response(); + } + + headers.remove(axum::http::header::HOST); + headers.remove(axum::http::header::CONTENT_LENGTH); + headers.remove(axum::http::header::CONNECTION); + let path_and_query = uri + .path_and_query() + .map(|value| value.as_str()) + .unwrap_or("/"); + let upstream = state + .client + .request(method, format!("{VSS_SERVER_URL}{path_and_query}")) + .headers(headers) + .body(body) + .send() + .await; + + let response = match upstream { + Ok(response) => response, + Err(error) => { + eprintln!("VSS fault proxy upstream request failed: {error}"); + return ( + axum::http::StatusCode::BAD_GATEWAY, + format!("VSS upstream request failed: {error}"), + ) + .into_response(); + } + }; + let status = response.status(); + let mut response_headers = response.headers().clone(); + response_headers.remove(axum::http::header::CONTENT_LENGTH); + response_headers.remove(axum::http::header::CONNECTION); + response_headers.remove(axum::http::header::TRANSFER_ENCODING); + match response.bytes().await { + Ok(body) => { + let mut downstream = (status, body).into_response(); + downstream.headers_mut().extend(response_headers); + downstream + } + Err(error) => ( + axum::http::StatusCode::BAD_GATEWAY, + format!("VSS upstream response failed: {error}"), + ) + .into_response(), } } diff --git a/src/test/vss_unreachable_openchannel.rs b/src/test/vss_unreachable_openchannel.rs index c7213118..85e159f9 100644 --- a/src/test/vss_unreachable_openchannel.rs +++ b/src/test/vss_unreachable_openchannel.rs @@ -2,6 +2,65 @@ use super::*; const TEST_DIR_BASE: &str = "tmp/vss_unreachable_openchannel/"; +async fn unlock_electrum_only(node_address: SocketAddr, password: &str) { + let payload = UnlockRequest { + password: password.to_string(), + bitcoind_rpc_username: None, + bitcoind_rpc_password: None, + bitcoind_rpc_host: None, + bitcoind_rpc_port: None, + indexer_url: Some(ELECTRUM_URL_REGTEST.to_string()), + proxy_endpoint: Some(PROXY_ENDPOINT_LOCAL.to_string()), + announce_addresses: vec![], + announce_alias: Some(s!("RLN_alias")), + gossip_source: None, + }; + let res = reqwest::Client::new() + .post(format!("http://{node_address}/unlock")) + .json(&payload) + .send() + .await + .unwrap(); + check_response_is_ok(res) + .await + .json::() + .await + .unwrap(); +} + +async fn start_node_electrum_only( + node_test_dir: &str, + node_peer_port: u16, +) -> (SocketAddr, String) { + let node_address = start_daemon(node_test_dir, node_peer_port, None, false).await; + let password = format!("{node_test_dir}.{node_peer_port}"); + init(node_address, &password, None).await; + unlock_electrum_only(node_address, &password).await; + wait_for_peer_port_ready(node_peer_port).await; + (node_address, password) +} + +async fn start_vss_node_electrum_only( + node_test_dir: &str, + node_peer_port: u16, + vss_url: &str, +) -> (SocketAddr, String) { + let node_address = start_daemon_with_vss( + node_test_dir, + node_peer_port, + false, + Some(vss_url.into()), + false, + ) + .await; + let password = format!("{node_test_dir}.{node_peer_port}"); + init(node_address, &password, None).await; + clear_vss_fence(node_address, &password).await; + unlock_electrum_only(node_address, &password).await; + wait_for_peer_port_ready(node_peer_port).await; + (node_address, password) +} + /// A node whose configured VSS server is unreachable must refuse a channel /// open with a clear error instead of accepting it and leaving the channel /// silently stuck in `Opening`; once VSS is reachable again the same open @@ -25,16 +84,9 @@ async fn openchannel_refused_while_vss_unreachable_inner() { let test_dir_node2 = format!("{TEST_DIR_BASE}node2"); let proxy = super::vss_offline_force_close::VssProxy::start(); - let (node1_addr, _, _) = start_node_with_vss( - &test_dir_node1, - NODE1_PEER_PORT, - false, - &proxy.url(), - None, - false, - ) - .await; - let (node2_addr, _) = start_node(&test_dir_node2, NODE2_PEER_PORT, false).await; + let (node1_addr, _) = + start_vss_node_electrum_only(&test_dir_node1, NODE1_PEER_PORT, &proxy.url()).await; + let (node2_addr, _) = start_node_electrum_only(&test_dir_node2, NODE2_PEER_PORT).await; fund_and_create_utxos(node1_addr, None).await; fund_and_create_utxos(node2_addr, None).await; diff --git a/src/vss_kv_store.rs b/src/vss_kv_store.rs index b956bd1a..108667d7 100644 --- a/src/vss_kv_store.rs +++ b/src/vss_kv_store.rs @@ -1,5 +1,5 @@ use std::collections::HashMap; -use std::sync::Arc; +use std::sync::{Arc, LazyLock, Mutex}; use std::time::Duration; use bitcoin::io; @@ -21,14 +21,18 @@ use vss_client::util::retry::{ type VssRetryPolicy = MaxTotalDelayRetryPolicy>>; -/// Dedicated runtime for driving synchronous VSS calls (see [`VssKvStore::block_on`]). -static VSS_RUNTIME: std::sync::LazyLock = std::sync::LazyLock::new(|| { - tokio::runtime::Builder::new_multi_thread() - .worker_threads(2) - .thread_name("vss-runtime") - .enable_all() - .build() - .expect("Failed to create VSS tokio runtime") +/// Serialized runtime for driving synchronous VSS calls (see [`VssKvStore::block_on`]). +/// +/// A current-thread runtime is intentional here. Entering a multi-thread Tokio runtime from +/// short-lived handoff threads races its reactor registration under ThreadSanitizer, while VSS +/// persistence requires ordered synchronous semantics rather than parallel call execution. +static VSS_RUNTIME: LazyLock> = LazyLock::new(|| { + Mutex::new( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("Failed to create VSS tokio runtime"), + ) }); /// Result of a single VSS get-object during a paged `download_all`. `Ok(None)` @@ -167,12 +171,20 @@ impl VssKvStore { { if tokio::runtime::Handle::try_current().is_ok() { std::thread::scope(|s| { - s.spawn(|| VSS_RUNTIME.block_on(future)) - .join() - .expect("VSS runtime thread panicked") + s.spawn(|| { + VSS_RUNTIME + .lock() + .expect("VSS runtime lock poisoned") + .block_on(future) + }) + .join() + .expect("VSS runtime thread panicked") }) } else { - VSS_RUNTIME.block_on(future) + VSS_RUNTIME + .lock() + .expect("VSS runtime lock poisoned") + .block_on(future) } }