mirror of
https://github.com/block/buzz.git
synced 2026-10-31 14:31:02 +00:00
refactor(desktop): drive secret txn lock through lock-owning seams
The interleave/txn-lock regressions drove `transaction_lock_at` directly in the test, not the production lock acquisition, so deleting the real lock in `save_managed_agents`/`save_agent_definitions` or the identity persist path left both tests green. The lock primitive was proven; the production wiring was not (Thufir r5 IMPORTANT #1). Extract AppHandle-free entry points that own lock -> mutation as one unit: `save_managed_agents_locked_at`/`save_agent_definitions_locked_at` in storage, and `persist_identity_locked` (in commands/identity, the command-orchestration layer) for the three identity-persist sites. All production wrappers route through them. Identity persist contends on the same store-directory inode every agent save takes via `secret_txn_lock_dir`, so identity/agent secrets on the shared blob mutually exclude. The interleave tests now drive those exact seams via a store-injected barrier that pauses mid-mutation inside the lock span; a non-blocking flock(LOCK_NB) probe proves EWOULDBLOCK exclusion. Deleting either storage acquisition or the identity acquisition now turns the matching regression red. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
@@ -405,7 +405,7 @@ const MIGRATION_MARKER_NAME: &str = "identity.migrated";
|
||||
/// The keyring operations the identity resolution flow needs. Abstracted so the
|
||||
/// corrupt-keyring recovery decision ([`recover_from_keyring`]) can be
|
||||
/// unit-tested against a fake without touching the live OS keyring.
|
||||
trait IdentityKeyStore {
|
||||
pub(crate) trait IdentityKeyStore {
|
||||
fn probe(&self, name: &str) -> crate::secret_store::KeyringProbe;
|
||||
fn load(&self, name: &str) -> Result<Option<String>, String>;
|
||||
fn store(&self, name: &str, value: &str) -> Result<(), String>;
|
||||
@@ -869,7 +869,7 @@ fn persist_identity_to_keyring(
|
||||
/// first via [`persist_identity_to_keyring`]; if the keyring is unavailable,
|
||||
/// falls back to the `0o600` identity.key file. Returns `Err` only when both
|
||||
/// the keyring write and the file fallback fail.
|
||||
fn persist_imported_identity_impl(
|
||||
pub(crate) fn persist_imported_identity_impl(
|
||||
store: &impl IdentityKeyStore,
|
||||
keys: &Keys,
|
||||
legacy_path: &std::path::Path,
|
||||
@@ -888,17 +888,6 @@ fn persist_imported_identity_impl(
|
||||
}
|
||||
}
|
||||
|
||||
/// Public entry point binding [`persist_imported_identity_impl`] to the shared
|
||||
/// [`crate::secret_store::SecretStore`]. See the impl for the persistence policy.
|
||||
pub(crate) fn persist_imported_identity(
|
||||
store: &crate::secret_store::SecretStore,
|
||||
keys: &Keys,
|
||||
legacy_path: &std::path::Path,
|
||||
data_dir: &std::path::Path,
|
||||
) -> Result<IdentityStorage, String> {
|
||||
persist_imported_identity_impl(store, keys, legacy_path, data_dir)
|
||||
}
|
||||
|
||||
/// Path of the migration-completed marker within `data_dir`.
|
||||
fn migration_marker_path(data_dir: &std::path::Path) -> std::path::PathBuf {
|
||||
data_dir.join(keyring_config::migration_marker_name(
|
||||
|
||||
@@ -362,20 +362,21 @@ pub async fn import_identity(
|
||||
let key_path = data_dir.join("identity.key");
|
||||
|
||||
let (pubkey, storage) = commit_imported_identity(&state, &data_dir, keys, |keys| {
|
||||
// Serialize the shared-keyring blob write against a concurrent
|
||||
// agent save/GC, which hold the same transaction lock across their
|
||||
// full span: identity and agent secrets share one SecretStore, so
|
||||
// an unserialized identity persist could interleave with a
|
||||
// projection transaction on the same blob. Held across the persist
|
||||
// span only — the in-memory key swap that follows this closure
|
||||
// touches no keyring blob.
|
||||
let _txn = crate::managed_agents::storage::acquire_secret_txn_lock(&app_handle)?;
|
||||
// Serialize the shared-keyring blob write against a concurrent agent
|
||||
// save/GC on the same SecretStore through the lock-owning
|
||||
// `persist_identity_locked` seam: it acquires the cross-process
|
||||
// secret transaction lock (the same store-directory inode every
|
||||
// agent save takes) and persists under it, so identity and agent
|
||||
// secrets can never interleave on the shared blob. Held across the
|
||||
// persist span only — the in-memory key swap that follows this
|
||||
// closure touches no keyring blob.
|
||||
let lock_dir = crate::managed_agents::storage::secret_txn_lock_dir(&app_handle)?;
|
||||
// Persist into the OS keyring first (store → read-back verify →
|
||||
// marker → delete file). Falls back to the 0o600 file when the
|
||||
// keyring is unavailable; returns Err only when both backends fail.
|
||||
let store =
|
||||
crate::secret_store::SecretStore::shared(crate::app_state::keyring_service());
|
||||
crate::app_state::persist_imported_identity(store, keys, &key_path, &data_dir)
|
||||
persist_identity_locked(store, keys, &key_path, &data_dir, &lock_dir)
|
||||
})?;
|
||||
|
||||
let pubkey_hex = pubkey.to_hex();
|
||||
@@ -459,6 +460,33 @@ pub(crate) fn commit_imported_identity(
|
||||
Ok((pubkey, storage))
|
||||
}
|
||||
|
||||
/// Lock-owning identity-persist seam: acquire the cross-process secret
|
||||
/// transaction lock on `lock_dir`, then persist the identity under it. This is
|
||||
/// the single seam every identity-persist entry point (`import_identity`,
|
||||
/// `persist_current_identity`, and the pairing recover path) routes through —
|
||||
/// so the lock and the keyring blob write it protects are one unit.
|
||||
///
|
||||
/// Identity and agent secrets share one `SecretStore` blob, so an unserialized
|
||||
/// identity persist could interleave with a concurrent agent save/GC projection
|
||||
/// transaction on the same blob. The lock target is the store **directory
|
||||
/// inode** ([`crate::secret_store::store_txn_lock_dir`]) — the same one every
|
||||
/// agent save takes — so the two surfaces mutually exclude. Held across the
|
||||
/// persist span only; the in-memory key swap that follows touches no blob.
|
||||
///
|
||||
/// AppHandle-free (takes `store` and `lock_dir` directly) so the interleave
|
||||
/// regression can drive this exact seam: delete the acquisition here and the
|
||||
/// concurrent-save probe acquires instead of reporting exclusion.
|
||||
pub(crate) fn persist_identity_locked<S: crate::app_state::IdentityKeyStore>(
|
||||
store: &S,
|
||||
keys: &Keys,
|
||||
legacy_path: &std::path::Path,
|
||||
data_dir: &std::path::Path,
|
||||
lock_dir: &std::path::Path,
|
||||
) -> Result<crate::app_state::IdentityStorage, String> {
|
||||
let _txn = crate::secret_store::transaction_lock_at(lock_dir)?;
|
||||
crate::app_state::persist_imported_identity_impl(store, keys, legacy_path, data_dir)
|
||||
}
|
||||
|
||||
/// Make the current ephemeral identity durable by persisting it to the OS
|
||||
/// keyring (or falling back to identity.key). This is called when the user
|
||||
/// chooses to start a new identity instead of re-importing their previous one
|
||||
@@ -504,10 +532,11 @@ pub async fn persist_current_identity(
|
||||
let store = crate::secret_store::SecretStore::shared(crate::app_state::keyring_service());
|
||||
let storage = {
|
||||
// Serialize the shared-keyring blob write against a concurrent agent
|
||||
// save/GC on the same SecretStore. Scoped to the persist span only —
|
||||
// save/GC on the same SecretStore through the lock-owning
|
||||
// `persist_identity_locked` seam. Scoped to the persist span only —
|
||||
// the identity_lost clear that follows touches no keyring blob.
|
||||
let _txn = crate::managed_agents::storage::acquire_secret_txn_lock(&app_handle)?;
|
||||
crate::app_state::persist_imported_identity(store, &keys, &key_path, &data_dir)?
|
||||
let lock_dir = crate::managed_agents::storage::secret_txn_lock_dir(&app_handle)?;
|
||||
persist_identity_locked(store, &keys, &key_path, &data_dir, &lock_dir)?
|
||||
};
|
||||
|
||||
// Keys are already the live identity. Record where the durable write
|
||||
@@ -802,6 +831,6 @@ mod nostr_identity_binding_tests {
|
||||
#[path = "identity_key_backup_tests.rs"]
|
||||
mod identity_key_backup_tests;
|
||||
|
||||
#[cfg(test)]
|
||||
#[cfg(all(test, unix, feature = "system-keyring"))]
|
||||
#[path = "identity_txn_lock_tests.rs"]
|
||||
mod identity_txn_lock_tests;
|
||||
|
||||
@@ -1,80 +1,145 @@
|
||||
//! W6-A interleave coverage.
|
||||
//! W6-A interleave coverage — driven through the PRODUCTION lock-owning seam.
|
||||
//!
|
||||
//! The identity-persist entry points (`import_identity`,
|
||||
//! `persist_current_identity`, and the pairing recover path) now hold the same
|
||||
//! cross-process secret transaction lock that a concurrent agent save/GC holds,
|
||||
//! so an identity keyring write can no longer interleave with a projection
|
||||
//! transaction on the shared `SecretStore` blob.
|
||||
//! `persist_current_identity`, and the pairing recover path) all route through
|
||||
//! `crate::commands::identity::persist_identity_locked`, which acquires the same
|
||||
//! cross-process secret transaction lock a concurrent agent save/GC holds
|
||||
//! (`transaction_lock_at(store_txn_lock_dir(store_file))`) and then persists the
|
||||
//! identity under it. So an identity keyring write can no longer interleave with
|
||||
//! a projection transaction on the shared `SecretStore` blob.
|
||||
//!
|
||||
//! Both real callers reach the lock through
|
||||
//! `acquire_secret_txn_lock`, which resolves it as
|
||||
//! `transaction_lock_at(store_txn_lock_dir(store_file))`. Driving the Tauri
|
||||
//! commands themselves needs a live `AppHandle` (integration-only); the unit
|
||||
//! level below proves the property that matters — that the identity-persist
|
||||
//! span and an agent-save span derive one lock target from the shared store
|
||||
//! file and mutually exclude on it.
|
||||
//! This drives that seam directly instead of hand-building a lock span: the
|
||||
//! injected store pauses the persist mid-keyring-write, and while it is paused a
|
||||
//! concurrent agent-save projection span (an independent open file description,
|
||||
//! exactly what a second process or a parallel save has) must be excluded.
|
||||
//! Delete the acquisition from `persist_identity_locked` and the probe acquires
|
||||
//! the lock instead of reporting exclusion — this test goes red.
|
||||
|
||||
/// While the identity-persist span holds the transaction lock, a concurrent
|
||||
/// agent-save projection span (an independent open file description, exactly
|
||||
/// what a second process or a parallel save has) must be excluded, and may
|
||||
/// acquire only after the persist span releases.
|
||||
use crate::app_state::{IdentityKeyStore, IdentityStorage};
|
||||
use crate::secret_store::KeyringProbe;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::mpsc;
|
||||
|
||||
/// An `IdentityKeyStore` that pauses the persist exactly once, inside `store`,
|
||||
/// so an observer can prove `persist_identity_locked` holds the transaction lock
|
||||
/// across the keyring write. `store` runs after the seam acquires the lock and
|
||||
/// before it releases, so the pause is squarely inside the protected span.
|
||||
struct BlockingIdentityStore {
|
||||
slot: std::cell::RefCell<HashMap<String, String>>,
|
||||
entered: mpsc::Sender<()>,
|
||||
release: mpsc::Receiver<()>,
|
||||
tripped: std::cell::Cell<bool>,
|
||||
}
|
||||
|
||||
impl BlockingIdentityStore {
|
||||
fn new(entered: mpsc::Sender<()>, release: mpsc::Receiver<()>) -> Self {
|
||||
Self {
|
||||
slot: std::cell::RefCell::new(HashMap::new()),
|
||||
entered,
|
||||
release,
|
||||
tripped: std::cell::Cell::new(false),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl IdentityKeyStore for BlockingIdentityStore {
|
||||
fn probe(&self, _name: &str) -> KeyringProbe {
|
||||
KeyringProbe::ReachableButEmpty
|
||||
}
|
||||
fn load(&self, name: &str) -> Result<Option<String>, String> {
|
||||
Ok(self.slot.borrow().get(name).cloned())
|
||||
}
|
||||
fn store(&self, name: &str, value: &str) -> Result<(), String> {
|
||||
// Pause once, under the seam's held lock, so the observer's non-blocking
|
||||
// probe reports EWOULDBLOCK. Store afterward so the read-back verify and
|
||||
// the rest of the persist succeed (→ SystemKeyring).
|
||||
if !self.tripped.replace(true) {
|
||||
self.entered.send(()).expect("signal persist entered store");
|
||||
self.release.recv().expect("observer released the persist");
|
||||
}
|
||||
self.slot
|
||||
.borrow_mut()
|
||||
.insert(name.to_string(), value.to_string());
|
||||
Ok(())
|
||||
}
|
||||
fn delete(&self, name: &str) -> Result<(), String> {
|
||||
self.slot.borrow_mut().remove(name);
|
||||
Ok(())
|
||||
}
|
||||
fn verify_stored(&self, name: &str, expected: &str) -> Result<bool, String> {
|
||||
Ok(self.slot.borrow().get(name).map(String::as_str) == Some(expected))
|
||||
}
|
||||
}
|
||||
|
||||
/// While `persist_identity_locked` holds the transaction lock across its keyring
|
||||
/// write, a concurrent agent-save projection span (an independent open file
|
||||
/// description) must be excluded, and may acquire only after the persist span
|
||||
/// releases.
|
||||
///
|
||||
/// Deterministic by handshake: the persist span holds the lock until the save
|
||||
/// span has probed exclusion and explicitly hands back — no timing sleeps. The
|
||||
/// `released` flag is set before the guard drops, so whenever the blocking
|
||||
/// acquire returns it is guaranteed to observe it, proving the save span could
|
||||
/// only proceed after release.
|
||||
#[cfg(all(unix, feature = "system-keyring"))]
|
||||
/// Deterministic by handshake: the persist span holds the lock until the
|
||||
/// injected store's paused `store` call is explicitly released — no timing
|
||||
/// sleeps. The probe runs while the persist is paused, so exclusion is the
|
||||
/// seam's own acquisition.
|
||||
#[test]
|
||||
fn test_identity_persist_span_excludes_concurrent_agent_save_on_shared_txn_lock() {
|
||||
use crate::secret_store::{store_txn_lock_dir, transaction_lock_at};
|
||||
fn test_persist_identity_locked_excludes_concurrent_agent_save_on_shared_txn_lock() {
|
||||
use crate::secret_store::store_txn_lock_dir;
|
||||
use std::os::unix::io::AsRawFd;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::mpsc;
|
||||
use std::sync::Arc;
|
||||
|
||||
// A store directory shared by identity and agent secrets — the same file
|
||||
// `managed_agents_store_path` yields and `acquire_secret_txn_lock` resolves.
|
||||
// `managed_agents_store_path` yields and the seam resolves its lock from.
|
||||
let dir = std::env::temp_dir().join(format!("buzz-w6a-interleave-{}", std::process::id()));
|
||||
std::fs::create_dir_all(&dir).expect("create store dir");
|
||||
let store_file = dir.join("managed-agents.json");
|
||||
std::fs::write(&store_file, b"[]").expect("seed store file");
|
||||
let data_dir = dir.clone();
|
||||
let key_path = dir.join("identity.key");
|
||||
|
||||
// The crux of W6-A: both spans derive the lock target from the store FILE
|
||||
// via `store_txn_lock_dir`, so identity persist and agent save contend on
|
||||
// one directory inode instead of racing on the shared keyring blob.
|
||||
// The crux of W6-A: the persist seam derives its lock target from the store
|
||||
// FILE via `store_txn_lock_dir`, so identity persist and agent save contend
|
||||
// on one directory inode instead of racing on the shared keyring blob.
|
||||
let lock_dir = store_txn_lock_dir(&store_file);
|
||||
|
||||
let released = Arc::new(AtomicBool::new(false));
|
||||
let (held_tx, held_rx) = mpsc::channel();
|
||||
let (go_tx, go_rx) = mpsc::channel();
|
||||
let keys = nostr::Keys::generate();
|
||||
let (entered_tx, entered_rx) = mpsc::channel::<()>();
|
||||
let (release_tx, release_rx) = mpsc::channel::<()>();
|
||||
|
||||
// Identity-persist span: hold the txn lock across its keyring write, then
|
||||
// release only once the save span has verified exclusion.
|
||||
let persist_dir = lock_dir.clone();
|
||||
let persist_released = Arc::clone(&released);
|
||||
// Identity-persist span: run the PRODUCTION seam on a worker; its injected
|
||||
// store pauses inside the keyring write, holding the lock until released.
|
||||
let persist_lock_dir = lock_dir.clone();
|
||||
let persist = std::thread::spawn(move || {
|
||||
let guard = transaction_lock_at(&persist_dir).expect("identity persist acquires txn lock");
|
||||
held_tx.send(()).expect("signal lock held");
|
||||
go_rx.recv().expect("save span verified exclusion");
|
||||
// Set BEFORE releasing: the waiter can only unblock once this drops, so
|
||||
// it is guaranteed to observe `true`.
|
||||
persist_released.store(true, Ordering::SeqCst);
|
||||
drop(guard);
|
||||
let store = BlockingIdentityStore::new(entered_tx, release_rx);
|
||||
let storage = crate::commands::identity::persist_identity_locked(
|
||||
&store,
|
||||
&keys,
|
||||
&key_path,
|
||||
&data_dir,
|
||||
&persist_lock_dir,
|
||||
)
|
||||
.expect("identity persist under the txn lock");
|
||||
assert_eq!(
|
||||
storage,
|
||||
IdentityStorage::SystemKeyring,
|
||||
"the blocking store persists successfully to the keyring"
|
||||
);
|
||||
});
|
||||
|
||||
held_rx
|
||||
entered_rx
|
||||
.recv()
|
||||
.expect("identity persist reported the lock held");
|
||||
.expect("persist reached its keyring write under the lock");
|
||||
|
||||
// Agent-save projection span: a non-blocking acquire on an independent OFD
|
||||
// must fail while identity persist holds the lock.
|
||||
// must fail while the persist seam holds the lock.
|
||||
let probe = std::fs::File::open(&lock_dir).expect("agent save opens store dir");
|
||||
let rc = unsafe { libc::flock(probe.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
|
||||
assert_eq!(
|
||||
rc, -1,
|
||||
"an agent save must NOT acquire the txn lock while identity persist holds it"
|
||||
);
|
||||
if rc == 0 {
|
||||
// Acquired: the seam is NOT holding the lock (regression). Release the
|
||||
// persist so the thread can exit, then fail with a clear message.
|
||||
unsafe { libc::flock(probe.as_raw_fd(), libc::LOCK_UN) };
|
||||
release_tx.send(()).expect("release persist span");
|
||||
persist.join().expect("identity persist thread");
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
panic!("an agent save acquired the txn lock while identity persist should have held it");
|
||||
}
|
||||
assert_eq!(
|
||||
std::io::Error::last_os_error().raw_os_error(),
|
||||
Some(libc::EWOULDBLOCK),
|
||||
@@ -83,16 +148,13 @@ fn test_identity_persist_span_excludes_concurrent_agent_save_on_shared_txn_lock(
|
||||
drop(probe);
|
||||
|
||||
// Hand back: persist releases, and the save span acquires through the real
|
||||
// blocking API. It returns only after release — proven by `released`.
|
||||
go_tx.send(()).expect("release identity persist span");
|
||||
let save_guard =
|
||||
transaction_lock_at(&lock_dir).expect("agent save acquires txn lock after release");
|
||||
assert!(
|
||||
released.load(Ordering::SeqCst),
|
||||
"agent save acquired the lock only after the identity persist span released it"
|
||||
);
|
||||
// blocking API only after release.
|
||||
release_tx.send(()).expect("release identity persist span");
|
||||
persist.join().expect("identity persist thread");
|
||||
|
||||
let save_guard = crate::secret_store::transaction_lock_at(&lock_dir)
|
||||
.expect("agent save acquires txn lock after release");
|
||||
drop(save_guard);
|
||||
|
||||
persist.join().expect("identity persist thread");
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
}
|
||||
|
||||
@@ -483,14 +483,17 @@ async fn import_recovered_identity(
|
||||
let key_path = data_dir.join("identity.key");
|
||||
crate::commands::identity::commit_imported_identity(&state, &data_dir, keys, |keys| {
|
||||
// Serialize the shared-keyring blob write against a concurrent
|
||||
// agent save/GC on the same SecretStore, held across the persist
|
||||
// span only. Lock order on this path is identity_mutation →
|
||||
// pairing generation_fence → txn; no save/GC path touches either
|
||||
// of the first two, so there is no inversion.
|
||||
let _txn = crate::managed_agents::storage::acquire_secret_txn_lock(&app)?;
|
||||
// agent save/GC on the same SecretStore through the lock-owning
|
||||
// `persist_identity_locked` seam, held across the persist span
|
||||
// only. Lock order on this path is identity_mutation → pairing
|
||||
// generation_fence → txn; no save/GC path touches either of the
|
||||
// first two, so there is no inversion.
|
||||
let lock_dir = crate::managed_agents::storage::secret_txn_lock_dir(&app)?;
|
||||
let store =
|
||||
crate::secret_store::SecretStore::shared(crate::app_state::keyring_service());
|
||||
crate::app_state::persist_imported_identity(store, keys, &key_path, &data_dir)
|
||||
crate::commands::identity::persist_identity_locked(
|
||||
store, keys, &key_path, &data_dir, &lock_dir,
|
||||
)
|
||||
})?;
|
||||
Ok(())
|
||||
})
|
||||
|
||||
@@ -336,16 +336,36 @@ fn hydrate_keys_with(store: &impl KeyStore, records: &mut [ManagedAgentRecord])
|
||||
/// before the wholesale rewrite so a definition is never dropped by an
|
||||
/// instance-side save (and vice versa via [`save_agent_definitions`]).
|
||||
pub fn save_managed_agents(app: &AppHandle, records: &[ManagedAgentRecord]) -> Result<(), String> {
|
||||
// Lock FIRST — the definition half re-read below must be under the lock
|
||||
// (Race 1); see `acquire_secret_txn_lock` for span + residual.
|
||||
let _txn = acquire_secret_txn_lock(app)?;
|
||||
save_managed_agents_at(
|
||||
save_managed_agents_locked_at(
|
||||
&managed_agents_store_path(app)?,
|
||||
agent_secret_store(),
|
||||
records,
|
||||
)
|
||||
}
|
||||
|
||||
/// Lock-owning path-based entry point for the instance-side save: acquires the
|
||||
/// cross-process secret transaction lock on the store directory, then runs the
|
||||
/// definition-preserving [`save_managed_agents_at`] under it. The lock and the
|
||||
/// mutation it protects live in ONE seam — the same one the interleave test
|
||||
/// drives — so the wiring is provable: delete the acquisition here and the
|
||||
/// concurrent-save regression stops observing exclusion. See
|
||||
/// [`acquire_secret_txn_lock`] for the lock span + residual.
|
||||
fn save_managed_agents_locked_at<S>(
|
||||
store_path: &Path,
|
||||
store: Option<&S>,
|
||||
records: &[ManagedAgentRecord],
|
||||
) -> Result<(), String>
|
||||
where
|
||||
S: KeyStore + crate::managed_agents::secret_projection::ProjectionStore,
|
||||
{
|
||||
// Lock FIRST — the definition half re-read inside `save_managed_agents_at`
|
||||
// must be under the lock (Race 1).
|
||||
let _txn = crate::secret_store::transaction_lock_at(&crate::secret_store::store_txn_lock_dir(
|
||||
store_path,
|
||||
))?;
|
||||
save_managed_agents_at(store_path, store, records)
|
||||
}
|
||||
|
||||
/// Path-based core of [`save_managed_agents`]: merges the caller's instance
|
||||
/// records with the definition half re-read from disk and commits the unified
|
||||
/// store. Split out so the definition-preservation contract can be exercised
|
||||
@@ -399,16 +419,34 @@ pub(crate) fn save_agent_definitions(
|
||||
app: &AppHandle,
|
||||
definitions: &[ManagedAgentRecord],
|
||||
) -> Result<(), String> {
|
||||
// Lock FIRST — the instance half re-read below must be under the lock
|
||||
// (Race 1); mirror of `save_managed_agents`. Leaf-level.
|
||||
let _txn = acquire_secret_txn_lock(app)?;
|
||||
save_agent_definitions_at(
|
||||
save_agent_definitions_locked_at(
|
||||
&managed_agents_store_path(app)?,
|
||||
agent_secret_store(),
|
||||
definitions,
|
||||
)
|
||||
}
|
||||
|
||||
/// Lock-owning path-based entry point for the definition-side save — the mirror
|
||||
/// of [`save_managed_agents_locked_at`]. Acquires the cross-process secret
|
||||
/// transaction lock on the store directory, then runs [`save_agent_definitions_at`]
|
||||
/// under it, so the lock and the instance-half re-read it protects are one seam
|
||||
/// the interleave test drives directly.
|
||||
fn save_agent_definitions_locked_at<S>(
|
||||
store_path: &Path,
|
||||
store: Option<&S>,
|
||||
definitions: &[ManagedAgentRecord],
|
||||
) -> Result<(), String>
|
||||
where
|
||||
S: crate::managed_agents::secret_projection::ProjectionStore,
|
||||
{
|
||||
// Lock FIRST — the instance half re-read inside `save_agent_definitions_at`
|
||||
// must be under the lock (Race 1); mirror of `save_managed_agents`.
|
||||
let _txn = crate::secret_store::transaction_lock_at(&crate::secret_store::store_txn_lock_dir(
|
||||
store_path,
|
||||
))?;
|
||||
save_agent_definitions_at(store_path, store, definitions)
|
||||
}
|
||||
|
||||
/// Path-based core of [`save_agent_definitions`]: merges the caller's definition
|
||||
/// records with the instance half re-read from disk and commits the unified
|
||||
/// store — the definition-side mirror of [`save_managed_agents_at`]. Split out
|
||||
@@ -680,6 +718,17 @@ pub(crate) fn acquire_secret_txn_lock(
|
||||
crate::secret_store::transaction_lock_at(&dir).map(Some)
|
||||
}
|
||||
|
||||
/// Resolve the store-directory lock target the secret transaction lock uses,
|
||||
/// from the same `managed-agents.json` path [`acquire_secret_txn_lock`] resolves.
|
||||
/// Exposed so the identity-persist seam ([`crate::commands::identity::persist_identity_locked`])
|
||||
/// contends on the exact directory inode every agent save takes — keeping the
|
||||
/// lock-target resolution in one place.
|
||||
pub(crate) fn secret_txn_lock_dir(app: &AppHandle) -> Result<PathBuf, String> {
|
||||
Ok(crate::secret_store::store_txn_lock_dir(
|
||||
&managed_agents_store_path(app)?,
|
||||
))
|
||||
}
|
||||
|
||||
/// Atomic, symlink-preserving JSON write.
|
||||
/// Resolves symlinks so the tmp+rename happens at the real target path,
|
||||
/// preserving any symlink at `path`.
|
||||
|
||||
@@ -4,36 +4,143 @@
|
||||
//! so it reuses that module's `FakeCombinedStore` and `record_with_pubkey_and_key`
|
||||
//! helpers without duplication, while keeping `storage_tests.rs` under the
|
||||
//! desktop file-size gate.
|
||||
//!
|
||||
//! These tests drive the PRODUCTION lock-owning entry points
|
||||
//! (`save_managed_agents_locked_at`, `save_agent_definitions_locked_at`) — the
|
||||
//! seams that own `lock -> raw read -> project -> atomic commit` — rather than
|
||||
//! hand-building a lock span around the lock-free `*_at` cores. That is the
|
||||
//! point: the exclusion the test observes is the production seam's own
|
||||
//! acquisition, so deleting `transaction_lock_at` from either wrapper turns the
|
||||
//! matching assertion red.
|
||||
|
||||
use super::super::{save_agent_definitions_at, save_managed_agents_at};
|
||||
use super::{record_with_pubkey_and_key, FakeCombinedStore, ManagedAgentRecord};
|
||||
use super::super::{save_agent_definitions_locked_at, save_managed_agents_locked_at};
|
||||
use super::{
|
||||
record_with_pubkey_and_key, FakeCombinedStore, KeyStore, KeyringProbe, ManagedAgentRecord,
|
||||
};
|
||||
use crate::managed_agents::secret_projection::ProjectionStore;
|
||||
use std::collections::HashMap;
|
||||
|
||||
/// A `ProjectionStore` + `KeyStore` that pauses the production save exactly
|
||||
/// once, mid-mutation, so an observer can prove the save holds the transaction
|
||||
/// lock at that instant.
|
||||
///
|
||||
/// The pause fires inside `store_batch_verified` — the single blob write every
|
||||
/// save routes its secret projection through — which the production seam
|
||||
/// reaches only AFTER acquiring the lock and BEFORE the JSON commit. On the
|
||||
/// first call it signals `entered` and blocks on `release`; every other
|
||||
/// operation delegates to an inner `FakeCombinedStore` so the save still reads
|
||||
/// back and commits normally once released.
|
||||
struct BarrierStore {
|
||||
inner: FakeCombinedStore,
|
||||
entered: std::sync::mpsc::Sender<()>,
|
||||
release: std::sync::mpsc::Receiver<()>,
|
||||
tripped: std::cell::Cell<bool>,
|
||||
}
|
||||
|
||||
impl BarrierStore {
|
||||
fn new(entered: std::sync::mpsc::Sender<()>, release: std::sync::mpsc::Receiver<()>) -> Self {
|
||||
Self {
|
||||
inner: FakeCombinedStore::new(),
|
||||
entered,
|
||||
release,
|
||||
tripped: std::cell::Cell::new(false),
|
||||
}
|
||||
}
|
||||
|
||||
/// Signal + block on the first secret write, once. Runs under the txn lock
|
||||
/// the production seam acquired, before the JSON commit.
|
||||
fn trip_once(&self) {
|
||||
if !self.tripped.replace(true) {
|
||||
self.entered.send(()).expect("signal save entered mutation");
|
||||
self.release.recv().expect("observer released the save");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl KeyStore for BarrierStore {
|
||||
fn probe(&self, name: &str) -> KeyringProbe {
|
||||
self.inner.probe(name)
|
||||
}
|
||||
fn load(&self, name: &str) -> Result<Option<String>, String> {
|
||||
self.inner.load(name)
|
||||
}
|
||||
fn load_all_readonly(&self) -> Result<Option<HashMap<String, String>>, String> {
|
||||
self.inner.load_all_readonly()
|
||||
}
|
||||
fn write_and_verify(&self, name: &str, value: &str) -> Result<(), String> {
|
||||
KeyStore::write_and_verify(&self.inner, name, value)
|
||||
}
|
||||
fn store_all(&self, entries: &HashMap<String, String>) -> Result<(), String> {
|
||||
self.inner.store_all(entries)
|
||||
}
|
||||
}
|
||||
|
||||
impl ProjectionStore for BarrierStore {
|
||||
fn write_and_verify(&self, key: &str, value: &str) -> Result<(), String> {
|
||||
ProjectionStore::write_and_verify(&self.inner, key, value)
|
||||
}
|
||||
fn load_key(&self, key: &str) -> Result<Option<String>, String> {
|
||||
self.inner.load_key(key)
|
||||
}
|
||||
fn load_all(&self) -> Result<Option<HashMap<String, String>>, String> {
|
||||
self.inner.load_all()
|
||||
}
|
||||
fn store_batch(&self, entries: &HashMap<String, String>) -> Result<(), String> {
|
||||
self.inner.store_batch(entries)
|
||||
}
|
||||
fn remove_batch(&self, keys: &[&str]) -> Result<(), String> {
|
||||
self.inner.remove_batch(keys)
|
||||
}
|
||||
fn store_batch_verified(&self, entries: &HashMap<String, String>) -> Result<(), String> {
|
||||
// Pause here, under the production seam's held lock, so the observer's
|
||||
// non-blocking probe reports EWOULDBLOCK. Delegate afterward so the save
|
||||
// reads its projection back and commits.
|
||||
self.trip_once();
|
||||
self.inner.store_batch_verified(entries)
|
||||
}
|
||||
}
|
||||
|
||||
/// A non-blocking acquire of the store-dir txn lock on an independent open file
|
||||
/// description must fail while another holder has it. `true` = excluded
|
||||
/// (`EWOULDBLOCK`), `false` = acquired (lock NOT held).
|
||||
fn probe_excluded(lock_dir: &std::path::Path) -> bool {
|
||||
use std::os::unix::io::AsRawFd;
|
||||
let probe = std::fs::File::open(lock_dir).expect("open store dir for probe");
|
||||
let rc = unsafe { libc::flock(probe.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
|
||||
if rc == 0 {
|
||||
// Acquired: release immediately so we leave no lock behind.
|
||||
unsafe { libc::flock(probe.as_raw_fd(), libc::LOCK_UN) };
|
||||
return false;
|
||||
}
|
||||
assert_eq!(
|
||||
std::io::Error::last_os_error().raw_os_error(),
|
||||
Some(libc::EWOULDBLOCK),
|
||||
"exclusion must report EWOULDBLOCK"
|
||||
);
|
||||
true
|
||||
}
|
||||
|
||||
/// Concurrent instance-save and definition-save against ONE store dir must lose
|
||||
/// neither half and re-inline neither half's secrets. Each save re-reads the
|
||||
/// OTHER half from disk under the shared secret transaction lock, so whichever
|
||||
/// commits second observes the first's write and carries it forward — instead
|
||||
/// of the last-writer-wins clobber a lock-free read-modify-write interleave
|
||||
/// would cause.
|
||||
/// neither half and re-inline neither half's secrets, and each save must hold
|
||||
/// the cross-process transaction lock across its own read-modify-write cycle.
|
||||
///
|
||||
/// Deterministic by handshake (identity_txn_lock_tests style): the instance
|
||||
/// save commits while holding the lock; a non-blocking probe proves the
|
||||
/// definition save is excluded mid-cycle; only after release does the
|
||||
/// definition save acquire and commit, so its instance re-read is guaranteed to
|
||||
/// observe the committed instance rather than the empty seed. The two saves use
|
||||
/// independent stores because `RefCell` is not `Sync` and the assertions read
|
||||
/// the committed JSON, not the keyring — each save only needs its own store to
|
||||
/// project into.
|
||||
/// Driven through the production lock-owning seams. Each save is paused
|
||||
/// mid-mutation by an injected `BarrierStore`; while paused, a non-blocking
|
||||
/// probe on an independent open file description proves the seam holds the lock
|
||||
/// at that instant (delete the acquisition and the probe acquires instead —
|
||||
/// red). After the instance save releases and commits, the definition save runs
|
||||
/// the same way and its raw instance re-read observes the committed instance
|
||||
/// rather than the empty seed, so neither half is clobbered or re-inlined.
|
||||
#[test]
|
||||
fn concurrent_instance_and_definition_saves_preserve_both_halves_without_reinlining() {
|
||||
use crate::secret_store::{store_txn_lock_dir, transaction_lock_at};
|
||||
use std::os::unix::io::AsRawFd;
|
||||
use std::sync::mpsc;
|
||||
fn production_saves_hold_txn_lock_and_preserve_both_halves_without_reinlining() {
|
||||
use crate::secret_store::store_txn_lock_dir;
|
||||
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let store_path = dir.path().join("managed-agents.json");
|
||||
std::fs::write(&store_path, b"[]").expect("seed empty store");
|
||||
let lock_dir = store_txn_lock_dir(&store_path);
|
||||
|
||||
// ── Instance save: holds the lock across its mutation ──────────────────
|
||||
// Instance half: a keyed record with an inline key AND inline env, both of
|
||||
// which the save must project into the keyring (never leave in JSON).
|
||||
let mut instance = record_with_pubkey_and_key("instance-pub", "nsec1instkey");
|
||||
@@ -41,54 +148,30 @@ fn concurrent_instance_and_definition_saves_preserve_both_halves_without_reinlin
|
||||
.into_iter()
|
||||
.collect();
|
||||
|
||||
let (committed_tx, committed_rx) = mpsc::channel();
|
||||
let (go_tx, go_rx) = mpsc::channel();
|
||||
|
||||
// Instance-save span: hold the txn lock across its full read-modify-write
|
||||
// cycle, then release only after the definition save has probed exclusion.
|
||||
let instance_store_path = store_path.clone();
|
||||
let instance_lock_dir = lock_dir.clone();
|
||||
let (inst_entered_tx, inst_entered_rx) = std::sync::mpsc::channel::<()>();
|
||||
let (inst_release_tx, inst_release_rx) = std::sync::mpsc::channel::<()>();
|
||||
let inst_store_path = store_path.clone();
|
||||
let instance_save = std::thread::spawn(move || {
|
||||
let store = FakeCombinedStore::new();
|
||||
let guard =
|
||||
transaction_lock_at(&instance_lock_dir).expect("instance save acquires txn lock");
|
||||
save_managed_agents_at(
|
||||
&instance_store_path,
|
||||
let store = BarrierStore::new(inst_entered_tx, inst_release_rx);
|
||||
save_managed_agents_locked_at(
|
||||
&inst_store_path,
|
||||
Some(&store),
|
||||
std::slice::from_ref(&instance),
|
||||
)
|
||||
.expect("instance save commits");
|
||||
committed_tx.send(()).expect("signal instance committed");
|
||||
go_rx.recv().expect("definition save probed exclusion");
|
||||
drop(guard);
|
||||
});
|
||||
|
||||
committed_rx
|
||||
inst_entered_rx
|
||||
.recv()
|
||||
.expect("instance save committed while holding the lock");
|
||||
|
||||
// Mid-cycle the definition save must be excluded: a non-blocking acquire on
|
||||
// an independent OFD fails while the instance save holds the lock. This is
|
||||
// what makes the ordering the lock's doing, not the scheduler's.
|
||||
let probe = std::fs::File::open(&lock_dir).expect("open store dir");
|
||||
let rc = unsafe { libc::flock(probe.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
|
||||
assert_eq!(
|
||||
rc, -1,
|
||||
"a definition save must NOT acquire the txn lock while an instance save holds it"
|
||||
.expect("instance save reached its mutation under the lock");
|
||||
assert!(
|
||||
probe_excluded(&lock_dir),
|
||||
"the instance save must hold the txn lock across its mutation"
|
||||
);
|
||||
assert_eq!(
|
||||
std::io::Error::last_os_error().raw_os_error(),
|
||||
Some(libc::EWOULDBLOCK),
|
||||
"exclusion must report EWOULDBLOCK"
|
||||
);
|
||||
drop(probe);
|
||||
|
||||
// Hand back: the instance save releases and exits, then the definition save
|
||||
// acquires through the real blocking lock. Its instance re-read now sees
|
||||
// the committed instance.
|
||||
go_tx.send(()).expect("release instance save");
|
||||
inst_release_tx.send(()).expect("release instance save");
|
||||
instance_save.join().expect("instance save thread");
|
||||
|
||||
// ── Definition save: holds the lock, and preserves the committed instance ─
|
||||
let mut definition: ManagedAgentRecord = serde_json::from_str(
|
||||
r#"{
|
||||
"pubkey": "",
|
||||
@@ -109,19 +192,30 @@ fn concurrent_instance_and_definition_saves_preserve_both_halves_without_reinlin
|
||||
.into_iter()
|
||||
.collect();
|
||||
|
||||
let def_store = FakeCombinedStore::new();
|
||||
let guard = transaction_lock_at(&lock_dir).expect("definition save acquires after release");
|
||||
save_agent_definitions_at(
|
||||
&store_path,
|
||||
Some(&def_store),
|
||||
std::slice::from_ref(&definition),
|
||||
)
|
||||
.expect("definition save commits");
|
||||
drop(guard);
|
||||
let (def_entered_tx, def_entered_rx) = std::sync::mpsc::channel::<()>();
|
||||
let (def_release_tx, def_release_rx) = std::sync::mpsc::channel::<()>();
|
||||
let def_store_path = store_path.clone();
|
||||
let definition_save = std::thread::spawn(move || {
|
||||
let store = BarrierStore::new(def_entered_tx, def_release_rx);
|
||||
save_agent_definitions_locked_at(
|
||||
&def_store_path,
|
||||
Some(&store),
|
||||
std::slice::from_ref(&definition),
|
||||
)
|
||||
.expect("definition save commits");
|
||||
});
|
||||
|
||||
// Neither half's secrets may appear inline in the committed JSON: the
|
||||
// instance the definition save re-read stays projected, and the definition
|
||||
// it wrote is projected too.
|
||||
def_entered_rx
|
||||
.recv()
|
||||
.expect("definition save reached its mutation under the lock");
|
||||
assert!(
|
||||
probe_excluded(&lock_dir),
|
||||
"the definition save must hold the txn lock across its mutation"
|
||||
);
|
||||
def_release_tx.send(()).expect("release definition save");
|
||||
definition_save.join().expect("definition save thread");
|
||||
|
||||
// ── Neither half lost, neither half re-inlined ─────────────────────────
|
||||
let committed = std::fs::read_to_string(&store_path).expect("read committed");
|
||||
assert!(
|
||||
!committed.contains("inst-plaintext"),
|
||||
|
||||
Reference in New Issue
Block a user