zeph_durable/backend/execution_lock.rs
1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! Cross-process advisory lock enforcing INV-15's single-owner-process invariant, held for the
5//! lifetime of a [`crate::DurableContext`] opened via
6//! [`LocalBackend::open_execution_exclusive`](crate::LocalBackend::open_execution_exclusive).
7//!
8//! Two processes can independently derive the same [`ExecutionId`] — the P1 agent-turn adapter
9//! keys it on `(ConversationId, sqlite_path)`, so two CLI instances pointed at the same memory
10//! database and the same conversation always agree on the id (#6122). Without a lock, both
11//! processes race `LocalBackend::open_execution`'s plain SELECT-then-INSERT and both drive
12//! `next_step` from 0, corrupting the journal (`ReplayDivergence`/`ReplayIntegrity` on whichever
13//! process loses the race). [`ExecutionLock`] closes that race with a non-blocking, exclusive
14//! `flock(2)` on a lock file named after the execution's UUID.
15//!
16//! # Why not `zeph_common::pidfile::PidLockGuard`
17//!
18//! `PidLockGuard` (the primitive backing `zeph-core::daemon::PidGuard` and
19//! `zeph-scheduler::pidfile::PidFile`) unlinks its lock file *before* its file descriptor closes
20//! (see its `Drop` impl). That ordering is safe for a pid file acquired once at daemon startup and
21//! released once at shutdown — vanishingly unlikely to race a concurrent acquirer — but is a real
22//! `flock`+`unlink` TOCTOU hazard for a lock acquired and released once per conversation turn under
23//! real contention (e.g. many CI agents sharing a testing database): a second process racing the
24//! unlink window can `open(O_CREAT)` a *fresh* inode at the just-unlinked path and `flock` it
25//! immediately, believing it holds the lock, while the first process's descriptor — still open on
26//! the now-orphaned original inode — has not actually released yet. [`ExecutionLock`] instead
27//! mirrors `zeph-session::log::SessionEventLog`'s own `AdvisoryLock`: the lock file is **never**
28//! unlinked. It is a permanent sentinel, and the kernel releasing the `flock` when the holding
29//! process's descriptors close (including on `SIGKILL`) is the only correctness signal — no
30//! unlink/re-create race, no PID-liveness polling needed.
31
32use std::path::Path;
33
34use crate::error::DurableError;
35use crate::ids::ExecutionId;
36
37/// Holds the advisory lock for one [`ExecutionId`] while alive; releases it on drop.
38///
39/// Unix only. On non-Unix targets `ExecutionLock::acquire` always succeeds and returns a
40/// no-op guard — the workspace has no vetted cross-platform advisory-locking primitive, mirroring
41/// `SessionEventLog::open_exclusive`'s degrade.
42#[cfg(unix)]
43#[derive(Debug)]
44pub struct ExecutionLock(#[allow(dead_code)] rustix::fd::OwnedFd);
45
46#[cfg(unix)]
47impl ExecutionLock {
48 /// Acquire the exclusive lock for `id` under `lock_dir` (created on demand).
49 ///
50 /// Stamps the current process id into the lock file's content (best-effort, diagnostic only —
51 /// never load-bearing for correctness) so a contending process can report a useful
52 /// `holder_pid` in [`DurableError::ExecutionLocked`]; unlike `PidLockGuard`, the file is never
53 /// unlinked, so a stale/unreadable pid on an old sentinel just yields `holder_pid: 0`.
54 ///
55 /// # Errors
56 ///
57 /// Returns [`DurableError::ExecutionLocked`] if another process already holds the lock, or
58 /// [`DurableError::Storage`] for any other filesystem failure.
59 pub(crate) fn acquire(lock_dir: &Path, id: ExecutionId) -> Result<Self, DurableError> {
60 use rustix::fs::{FlockOperation, Mode, OFlags};
61
62 std::fs::create_dir_all(lock_dir)
63 .map_err(|e| DurableError::storage("open_execution_exclusive", e))?;
64 let lock_path = lock_dir.join(format!("{id}.lock"));
65
66 let fd = rustix::fs::open(
67 &lock_path,
68 OFlags::RDWR | OFlags::CREATE | OFlags::CLOEXEC,
69 Mode::from_raw_mode(0o600),
70 )
71 .map_err(|e| DurableError::storage("open_execution_exclusive", std::io::Error::from(e)))?;
72
73 rustix::fs::flock(&fd, FlockOperation::NonBlockingLockExclusive).map_err(|e| {
74 if e == rustix::io::Errno::WOULDBLOCK {
75 let holder_pid = zeph_common::pidfile::read_pid_lenient(&lock_path).unwrap_or(0);
76 DurableError::ExecutionLocked {
77 execution_id: id,
78 holder_pid,
79 }
80 } else {
81 DurableError::storage("open_execution_exclusive", std::io::Error::from(e))
82 }
83 })?;
84
85 // Best-effort PID stamp for the next contender's error message — never propagate a
86 // failure here, the lock itself is already held and correctness does not depend on it.
87 let _ = rustix::fs::ftruncate(&fd, 0);
88 let _ = rustix::io::write(&fd, std::process::id().to_string().as_bytes());
89
90 Ok(Self(fd))
91 }
92}
93
94/// No vetted cross-platform advisory-locking primitive exists in this workspace, so
95/// [`LocalBackend::open_execution_exclusive`](crate::LocalBackend::open_execution_exclusive) does
96/// not enforce INV-15 on non-Unix targets.
97#[cfg(not(unix))]
98#[derive(Debug)]
99pub struct ExecutionLock;
100
101#[cfg(not(unix))]
102impl ExecutionLock {
103 pub(crate) fn acquire(_lock_dir: &Path, _id: ExecutionId) -> Result<Self, DurableError> {
104 Ok(Self)
105 }
106}
107
108#[cfg(all(test, unix))]
109mod tests {
110 use super::*;
111
112 #[test]
113 fn second_acquire_for_same_execution_fails() {
114 let dir = tempfile::tempdir().unwrap();
115 let id = ExecutionId::new();
116
117 let _first = ExecutionLock::acquire(dir.path(), id).expect("first acquire succeeds");
118 let err = ExecutionLock::acquire(dir.path(), id).expect_err("second acquire must fail");
119 assert!(
120 matches!(err, DurableError::ExecutionLocked { execution_id, .. } if execution_id == id),
121 "expected ExecutionLocked for the same execution_id, got {err:?}"
122 );
123 }
124
125 #[test]
126 fn second_acquire_reports_holder_pid() {
127 let dir = tempfile::tempdir().unwrap();
128 let id = ExecutionId::new();
129
130 let _first = ExecutionLock::acquire(dir.path(), id).expect("first acquire succeeds");
131 let err = ExecutionLock::acquire(dir.path(), id).expect_err("second acquire must fail");
132 let DurableError::ExecutionLocked { holder_pid, .. } = err else {
133 panic!("expected ExecutionLocked, got {err:?}");
134 };
135 assert_eq!(
136 holder_pid,
137 std::process::id(),
138 "holder_pid should report this test process's own pid (the only holder)"
139 );
140 }
141
142 #[test]
143 fn distinct_executions_do_not_contend() {
144 let dir = tempfile::tempdir().unwrap();
145 let a = ExecutionId::new();
146 let b = ExecutionId::new();
147
148 let _lock_a = ExecutionLock::acquire(dir.path(), a).expect("lock a succeeds");
149 let _lock_b =
150 ExecutionLock::acquire(dir.path(), b).expect("distinct execution_id does not block");
151 }
152
153 #[test]
154 fn reacquire_after_drop_succeeds() {
155 let dir = tempfile::tempdir().unwrap();
156 let id = ExecutionId::new();
157
158 {
159 let _first = ExecutionLock::acquire(dir.path(), id).expect("first acquire succeeds");
160 }
161 let _second =
162 ExecutionLock::acquire(dir.path(), id).expect("lock is released when guard drops");
163 }
164
165 #[test]
166 fn lock_file_is_not_unlinked_on_drop() {
167 // Regression test for critic finding S2: unlike `PidLockGuard`, the sentinel file must
168 // survive release — only the flock itself signals ownership. Deleting it on drop reopens
169 // the flock+unlink TOCTOU race under the higher-contention per-execution locking pattern.
170 let dir = tempfile::tempdir().unwrap();
171 let id = ExecutionId::new();
172 let lock_path = dir.path().join(format!("{id}.lock"));
173
174 {
175 let _guard = ExecutionLock::acquire(dir.path(), id).expect("acquire succeeds");
176 assert!(lock_path.exists(), "lock file must exist while held");
177 }
178 assert!(
179 lock_path.exists(),
180 "lock file must remain on disk after the guard drops (permanent sentinel)"
181 );
182 }
183}