Skip to main content

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}