lean_ctx/core/context_store/
task_signals.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Per-task runtime signals that say whether a compressed read served the
3//! model: a full re-read of the same file (bounce), an edit that failed after
4//! a compressed read, or an expansion of a file handle that was read
5//! compressed.
6//!
7//! Each event is attributed to the task whose compressed read caused it — not
8//! to the task that happens to be running when it is noticed. Nothing missing
9//! may read as clean: an event whose causing read is unknown (another process,
10//! before a restart), an event that could not be recorded, and a task served
11//! by more than one process (whose bounces no single process can see) all
12//! leave that task's signals unknown. Written once per process and task and
13//! when an event happens, never per read; counts only, no paths.
14
15use std::path::{Path, PathBuf};
16
17use serde::{Deserialize, Serialize};
18
19const MAX_FILE_BYTES: u64 = 4 * 1024;
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub enum Signal {
23    Bounce,
24    EditFailure,
25    Expand,
26    Unattributed,
27    /// This process serves reads for the task.
28    Observed,
29}
30
31/// Counts of one task's events.
32#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(deny_unknown_fields)]
34pub(crate) struct TaskSignals {
35    pub(crate) schema_version: u32,
36    pub bounce: u32,
37    pub edit_failure: u32,
38    pub expand: u32,
39    pub unattributed: u32,
40    /// The process that served the task's reads, while only one did.
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub(crate) observer: Option<String>,
43}
44
45/// Identifies this process among the ones that serve a task's reads.
46fn process_nonce() -> &'static str {
47    static NONCE: std::sync::OnceLock<String> = std::sync::OnceLock::new();
48    NONCE.get_or_init(|| {
49        let started = std::time::SystemTime::now()
50            .duration_since(std::time::UNIX_EPOCH)
51            .map_or(0, |elapsed| elapsed.as_nanos());
52        format!("{}-{started}", std::process::id())
53    })
54}
55
56/// The tenant/project scope and task id a read ran under.
57pub(crate) type Origin = (String, String);
58
59/// The scope and task of the code running now, if it runs inside a task.
60pub(crate) fn current_origin() -> Option<Origin> {
61    let envelope = crate::core::task_spine::TaskSpine::current()?;
62    Some((
63        super::task_scope(envelope.tenant_id.as_ref(), &envelope.project_id),
64        envelope.task_id.as_str().to_owned(),
65    ))
66}
67
68fn gateway_dir() -> Option<PathBuf> {
69    crate::core::data_dir::lean_ctx_data_dir()
70        .ok()
71        .map(|dir| dir.join("gateway"))
72}
73
74fn signals_path(dir: &Path, (scope, task): &(String, String)) -> PathBuf {
75    let mut hasher = blake3::Hasher::new();
76    hasher.update(scope.as_bytes());
77    hasher.update(&[0]);
78    hasher.update(task.as_bytes());
79    let digest = hasher.finalize().to_hex();
80    dir.join("tasks")
81        .join(format!("{}.signals.json", &digest[..32]))
82}
83
84/// Marks a task whose event could not be recorded; written without the lock.
85fn incomplete_marker(path: &Path) -> PathBuf {
86    path.with_extension("incomplete")
87}
88
89fn read(path: &Path) -> Option<TaskSignals> {
90    if std::fs::metadata(path).ok()?.len() > MAX_FILE_BYTES {
91        return None;
92    }
93    serde_json::from_slice::<TaskSignals>(&std::fs::read(path).ok()?)
94        .ok()
95        .filter(|signals| signals.schema_version == 1)
96}
97
98pub(crate) fn record_in(dir: &Path, origin: &Origin, signal: Signal) -> Result<(), String> {
99    record_as(dir, origin, signal, process_nonce())
100}
101
102fn record_as(dir: &Path, origin: &Origin, signal: Signal, process: &str) -> Result<(), String> {
103    let path = signals_path(dir, origin);
104    let _lock = crate::core::context_admission::receipt_store::lock_task_files(dir)
105        .ok_or("task index lock unavailable")?;
106    // A damaged file is replaced by a state that says nothing is known.
107    let mut signals = match std::fs::metadata(&path) {
108        Ok(_) => read(&path).unwrap_or(TaskSignals {
109            unattributed: 1,
110            ..TaskSignals::default()
111        }),
112        Err(_) => TaskSignals::default(),
113    };
114    signals.schema_version = 1;
115    let counter = match signal {
116        Signal::Bounce => Some(&mut signals.bounce),
117        Signal::EditFailure => Some(&mut signals.edit_failure),
118        Signal::Expand => Some(&mut signals.expand),
119        Signal::Unattributed => Some(&mut signals.unattributed),
120        Signal::Observed => match signals.observer.as_deref() {
121            Some(observer) if observer == process => return Ok(()),
122            // A second process: neither sees the other's bounces.
123            Some(_) => Some(&mut signals.unattributed),
124            None => {
125                signals.observer = Some(process.to_owned());
126                None
127            }
128        },
129    };
130    if let Some(counter) = counter {
131        *counter = counter.saturating_add(1);
132    }
133    let bytes = serde_json::to_vec(&signals).map_err(|e| e.to_string())?;
134    // A file of the gateway evidence store: written under its bound (G09).
135    // Refused, the event is not recorded and the caller marks the task's
136    // signals unknown.
137    crate::core::context_admission::receipt_store::write_atomic(dir, &path, &bytes)
138}
139
140/// Record `signal` for the task whose read caused it. An event that cannot
141/// be recorded marks the task's signals unknown instead of leaving them clean.
142pub fn record(origin: &Origin, signal: Signal) {
143    let Some(dir) = gateway_dir() else {
144        return;
145    };
146    if let Err(error) = record_in(&dir, origin, signal) {
147        tracing::debug!("task signal not recorded: {error}");
148        // Zero bytes, set whatever the store's bound says: "unknown" must
149        // not be lost to it.
150        let marker = incomplete_marker(&signals_path(&dir, origin));
151        let _ = crate::core::context_admission::receipt_store::capacity::mark(&dir, &marker);
152    }
153}
154
155/// Register this process as serving reads for the task, once per process.
156pub(crate) fn observe(origin: &Origin) {
157    static OBSERVED: std::sync::OnceLock<std::sync::Mutex<std::collections::HashSet<Origin>>> =
158        std::sync::OnceLock::new();
159    let first = OBSERVED
160        .get_or_init(Default::default)
161        .lock()
162        .map_or(true, |mut observed| observed.insert(origin.clone()));
163    if first {
164        let origin = origin.clone();
165        std::thread::spawn(move || record(&origin, Signal::Observed));
166    }
167}
168
169/// An event caused by a compressed read nobody can name: the current task's
170/// signals become unknown.
171pub fn record_unattributed() {
172    if let Some(origin) = current_origin() {
173        record(&origin, Signal::Unattributed);
174    }
175}
176
177pub(crate) fn load_in(dir: &Path, origin: &Origin) -> Option<TaskSignals> {
178    let path = signals_path(dir, origin);
179    let unknown = TaskSignals {
180        unattributed: 1,
181        ..TaskSignals::default()
182    };
183    if incomplete_marker(&path).exists() {
184        return Some(read(&path).map_or(unknown, |signals| TaskSignals {
185            unattributed: signals.unattributed.max(1),
186            ..signals
187        }));
188    }
189    if !path.exists() {
190        return Some(TaskSignals::default());
191    }
192    read(&path).or(Some(unknown))
193}
194
195/// The task's signals; a task without a file had no event.
196pub(crate) fn load(origin: &Origin) -> Option<TaskSignals> {
197    load_in(&gateway_dir()?, origin)
198}
199
200#[cfg(test)]
201mod tests {
202    use super::*;
203
204    #[test]
205    fn events_count_per_task_and_scope_and_damage_means_unknown() {
206        let dir = tempfile::tempdir().expect("tempdir");
207        let task_a = ("[null,\"a\"]".to_owned(), "task-1".to_owned());
208        let task_b = ("[null,\"b\"]".to_owned(), "task-1".to_owned());
209        assert_eq!(load_in(dir.path(), &task_a), Some(TaskSignals::default()));
210        record_in(dir.path(), &task_a, Signal::Bounce).expect("record");
211        record_in(dir.path(), &task_a, Signal::Bounce).expect("record");
212        record_in(dir.path(), &task_a, Signal::EditFailure).expect("record");
213        let a = load_in(dir.path(), &task_a).expect("signals");
214        assert_eq!(
215            (a.bounce, a.edit_failure, a.expand, a.unattributed),
216            (2, 1, 0, 0)
217        );
218        assert_eq!(
219            load_in(dir.path(), &task_b),
220            Some(TaskSignals::default()),
221            "equal task ids in another project never share signals"
222        );
223
224        let path = signals_path(dir.path(), &task_b);
225        std::fs::create_dir_all(path.parent().expect("parent")).expect("dir");
226        std::fs::write(&path, "{broken").expect("damage");
227        assert_eq!(
228            load_in(dir.path(), &task_b).map(|s| s.unattributed),
229            Some(1)
230        );
231    }
232
233    #[test]
234    fn a_task_served_by_two_processes_or_with_a_lost_event_is_unknown() {
235        let dir = tempfile::tempdir().expect("tempdir");
236        let task = ("[null,\"p\"]".to_owned(), "task-1".to_owned());
237        record_as(dir.path(), &task, Signal::Observed, "proc-a").expect("observe");
238        record_as(dir.path(), &task, Signal::Observed, "proc-a").expect("observe again");
239        assert_eq!(
240            load_in(dir.path(), &task).map(|s| s.unattributed),
241            Some(0),
242            "one process sees every bounce of its task"
243        );
244        record_as(dir.path(), &task, Signal::Observed, "proc-b").expect("observe");
245        assert_eq!(
246            load_in(dir.path(), &task).map(|s| s.unattributed),
247            Some(1),
248            "a second process: bounces across them are invisible"
249        );
250
251        let lost = ("[null,\"p\"]".to_owned(), "task-2".to_owned());
252        std::fs::create_dir_all(dir.path().join("tasks")).expect("dir");
253        std::fs::write(incomplete_marker(&signals_path(dir.path(), &lost)), b"").expect("marker");
254        assert_eq!(
255            load_in(dir.path(), &lost).map(|s| s.unattributed),
256            Some(1),
257            "an event that could not be recorded is not a clean task"
258        );
259    }
260}
lean_ctx/core/context_store/task_signals.rs

lean_ctx/core/context_store/
task_signals.rs
