lean_ctx/core/
events.rs

1use serde::{Deserialize, Serialize};
2use std::collections::VecDeque;
3use std::io::Write;
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::sync::{Mutex, OnceLock};
6
7const RING_CAPACITY: usize = 1000;
8const JSONL_MAX_LINES: usize = 10_000;
9const EVENT_ID_BLOCK_SIZE: u64 = 1_024;
10
11#[derive(Clone, Debug, Serialize, Deserialize)]
12pub struct LeanCtxEvent {
13    pub id: u64,
14    pub timestamp: String,
15    pub kind: EventKind,
16}
17
18#[derive(Clone, Debug, Serialize, Deserialize)]
19#[serde(tag = "type")]
20pub enum EventKind {
21    ToolCall {
22        tool: String,
23        tokens_original: u64,
24        tokens_saved: u64,
25        mode: Option<String>,
26        duration_ms: u64,
27        path: Option<String>,
28    },
29    CacheHit {
30        path: String,
31        saved_tokens: u64,
32    },
33    Compression {
34        path: String,
35        before_lines: u32,
36        after_lines: u32,
37        strategy: String,
38        kept_line_count: u32,
39        removed_line_count: u32,
40    },
41    AgentAction {
42        agent_id: String,
43        action: String,
44        tool: Option<String>,
45    },
46    KnowledgeUpdate {
47        category: String,
48        key: String,
49        action: String,
50    },
51    ThresholdShift {
52        language: String,
53        old_entropy: f64,
54        new_entropy: f64,
55        old_jaccard: f64,
56        new_jaccard: f64,
57    },
58    BudgetWarning {
59        role: String,
60        dimension: String,
61        used: String,
62        limit: String,
63        percent: u8,
64    },
65    BudgetExhausted {
66        role: String,
67        dimension: String,
68        used: String,
69        limit: String,
70    },
71    PolicyViolation {
72        role: String,
73        tool: String,
74        reason: String,
75    },
76    RoleChanged {
77        from: String,
78        to: String,
79    },
80    ProfileChanged {
81        from: String,
82        to: String,
83    },
84    SloViolation {
85        slo_name: String,
86        metric: String,
87        threshold: f64,
88        actual: f64,
89        action: String,
90    },
91    Anomaly {
92        metric: String,
93        expected: f64,
94        actual: f64,
95        deviation_factor: f64,
96    },
97    VerificationWarning {
98        warning_kind: String,
99        detail: String,
100        severity: String,
101    },
102    ThresholdAdapted {
103        language: String,
104        arm: String,
105        old_threshold: f64,
106        new_threshold: f64,
107    },
108}
109
110struct EventBus {
111    ring: Mutex<VecDeque<LeanCtxEvent>>,
112}
113
114impl EventBus {
115    fn new() -> Self {
116        Self {
117            ring: Mutex::new(VecDeque::with_capacity(RING_CAPACITY)),
118        }
119    }
120
121    fn emit(&self, kind: EventKind) -> u64 {
122        let id = next_event_id();
123        let event = LeanCtxEvent {
124            id,
125            timestamp: chrono::Local::now()
126                .format("%Y-%m-%dT%H:%M:%S%.3f")
127                .to_string(),
128            kind,
129        };
130
131        {
132            let mut ring = self
133                .ring
134                .lock()
135                .unwrap_or_else(std::sync::PoisonError::into_inner);
136            if ring.len() >= RING_CAPACITY {
137                ring.pop_front();
138            }
139            ring.push_back(event.clone());
140        }
141
142        append_jsonl(&event);
143        id
144    }
145
146    fn events_since(&self, after_id: u64) -> Vec<LeanCtxEvent> {
147        let ring = self
148            .ring
149            .lock()
150            .unwrap_or_else(std::sync::PoisonError::into_inner);
151        ring.iter().filter(|e| e.id > after_id).cloned().collect()
152    }
153
154    fn latest_events(&self, n: usize) -> Vec<LeanCtxEvent> {
155        let ring = self
156            .ring
157            .lock()
158            .unwrap_or_else(std::sync::PoisonError::into_inner);
159        let len = ring.len();
160        let start = len.saturating_sub(n);
161        ring.iter().skip(start).cloned().collect()
162    }
163}
164
165fn bus() -> &'static EventBus {
166    static INSTANCE: OnceLock<EventBus> = OnceLock::new();
167    INSTANCE.get_or_init(EventBus::new)
168}
169
170fn jsonl_path() -> Option<std::path::PathBuf> {
171    crate::core::paths::state_dir()
172        .ok()
173        .map(|d| d.join("events.jsonl"))
174}
175
176fn event_sequence_path() -> Option<std::path::PathBuf> {
177    crate::core::paths::state_dir()
178        .ok()
179        .map(|d| d.join("events.seq"))
180}
181
182fn max_event_id_in_journal(path: &std::path::Path) -> u64 {
183    let Ok(content) = std::fs::read_to_string(path) else {
184        return 0;
185    };
186    content
187        .lines()
188        .filter_map(|line| serde_json::from_str::<LeanCtxEvent>(line).ok())
189        .map(|event| event.id)
190        .max()
191        .unwrap_or(0)
192}
193
194fn parse_sequence_record(record: &str) -> Option<u64> {
195    let (value, checksum) = record.trim().split_once(':')?;
196    let value = value.parse::<u64>().ok()?;
197    let checksum = checksum.parse::<u64>().ok()?;
198    (checksum == !value).then_some(value)
199}
200
201fn read_persisted_sequence(path: &std::path::Path) -> Option<u64> {
202    let content = std::fs::read_to_string(path).ok()?;
203    content.lines().rev().find_map(parse_sequence_record)
204}
205
206fn sequence_record(value: u64) -> String {
207    format!("{value}:{}", !value)
208}
209
210fn reserve_event_id_block_at(
211    sequence_path: &std::path::Path,
212    journal_path: &std::path::Path,
213    block_size: u64,
214) -> std::io::Result<(u64, u64)> {
215    use fs2::FileExt;
216
217    if block_size == 0 {
218        return Err(std::io::Error::other("event id block must not be empty"));
219    }
220    if let Some(parent) = sequence_path.parent() {
221        std::fs::create_dir_all(parent)?;
222    }
223    // Lock a stable companion inode. The append-only sequence journal can be
224    // repaired without allowing another process to bypass the lock.
225    let lock_path = sequence_path.with_extension("seq.lock");
226    let lock = std::fs::OpenOptions::new()
227        .create(true)
228        .truncate(false)
229        .read(true)
230        .write(true)
231        .open(lock_path)?;
232    lock.lock_exclusive()?;
233
234    let result = (|| {
235        let persisted = read_persisted_sequence(sequence_path);
236        let baseline = persisted.unwrap_or_else(|| {
237            max_event_id_in_journal(journal_path).max(max_event_id_in_journal(
238                &journal_path.with_extension("jsonl.old"),
239            ))
240        });
241        let first = baseline
242            .checked_add(1)
243            .ok_or_else(|| std::io::Error::other("event id space exhausted"))?;
244        let last = baseline
245            .checked_add(block_size)
246            .ok_or_else(|| std::io::Error::other("event id space exhausted"))?;
247
248        // A prior writer's record (or corruption) may leave the file without
249        // a trailing newline; appending directly would fuse onto that line
250        // and make both records unparsable. Start a fresh line first.
251        let needs_leading_newline = std::fs::read(sequence_path)
252            .ok()
253            .is_some_and(|bytes| !bytes.is_empty() && bytes.last() != Some(&b'\n'));
254
255        let mut sequence = std::fs::OpenOptions::new()
256            .create(true)
257            .append(true)
258            .open(sequence_path)?;
259        let mut record = Vec::new();
260        if needs_leading_newline {
261            record.push(b'\n');
262        }
263        record.extend_from_slice(sequence_record(last).as_bytes());
264        record.push(b'\n');
265        sequence.write_all(&record)?;
266        sequence.sync_data()?;
267        Ok((first, last))
268    })();
269
270    let _ = FileExt::unlock(&lock);
271    result
272}
273
274#[cfg(test)]
275fn next_event_id_at(
276    sequence_path: &std::path::Path,
277    journal_path: &std::path::Path,
278) -> std::io::Result<u64> {
279    reserve_event_id_block_at(sequence_path, journal_path, 1).map(|(first, _)| first)
280}
281
282fn next_event_id() -> u64 {
283    #[derive(Default)]
284    struct LocalBlock {
285        next: u64,
286        last: u64,
287    }
288
289    static BLOCK: OnceLock<Mutex<LocalBlock>> = OnceLock::new();
290    static FALLBACK: AtomicU64 = AtomicU64::new(0);
291    if !is_test_environment()
292        && let Some((sequence, journal)) = event_sequence_path().zip(jsonl_path())
293    {
294        let mut block = BLOCK
295            .get_or_init(|| Mutex::new(LocalBlock::default()))
296            .lock()
297            .unwrap_or_else(std::sync::PoisonError::into_inner);
298        if block.next <= block.last && block.next != 0 {
299            let id = block.next;
300            block.next = block.next.saturating_add(1);
301            return id;
302        }
303        if let Ok((first, last)) =
304            reserve_event_id_block_at(&sequence, &journal, EVENT_ID_BLOCK_SIZE)
305        {
306            block.next = first.saturating_add(1);
307            block.last = last;
308            return first;
309        }
310    }
311
312    let base = u64::try_from(chrono::Utc::now().timestamp_millis())
313        .unwrap_or_default()
314        .saturating_mul(1_000);
315    base.saturating_add(FALLBACK.fetch_add(1, Ordering::Relaxed))
316}
317
318fn is_test_environment() -> bool {
319    use std::sync::OnceLock;
320    static CACHED: OnceLock<bool> = OnceLock::new();
321    *CACHED.get_or_init(|| {
322        if cfg!(test) {
323            return true;
324        }
325        if std::env::var_os("__LEAN_CTX_SKIP_EVENTS").is_some() {
326            return true;
327        }
328        std::env::current_exe().is_ok_and(|p| {
329            let s = p.to_string_lossy();
330            s.contains("/deps/") || s.contains("\\deps\\")
331        })
332    })
333}
334
335fn append_jsonl(event: &LeanCtxEvent) {
336    if is_test_environment() {
337        return;
338    }
339    let Some(path) = jsonl_path() else { return };
340    let _ = append_jsonl_at(&path, event);
341}
342
343fn append_jsonl_at(path: &std::path::Path, event: &LeanCtxEvent) -> std::io::Result<()> {
344    use fs2::FileExt;
345    use std::io::Write;
346
347    if let Some(parent) = path.parent() {
348        std::fs::create_dir_all(parent)?;
349    }
350
351    // Lock a stable companion inode rather than events.jsonl itself: rotation
352    // renames the data file, so locking that inode would let another process
353    // open the replacement and bypass the lock.
354    let lock_path = path.with_extension("jsonl.lock");
355    let lock = std::fs::OpenOptions::new()
356        .create(true)
357        .truncate(false)
358        .read(true)
359        .write(true)
360        .open(lock_path)?;
361    lock.lock_exclusive()?;
362
363    let result = (|| {
364        if let Ok(content) = std::fs::read_to_string(path)
365            && content.lines().count() >= JSONL_MAX_LINES
366        {
367            let old = path.with_extension("jsonl.old");
368            let _ = std::fs::remove_file(&old);
369            std::fs::rename(path, old)?;
370        }
371
372        let json = serde_json::to_string(event).map_err(std::io::Error::other)?;
373        let mut line = json.into_bytes();
374        line.push(b'\n');
375        let mut file = std::fs::OpenOptions::new()
376            .create(true)
377            .append(true)
378            .open(path)?;
379        // One logical write while holding the interprocess lock prevents JSON
380        // objects and their trailing newlines from interleaving.
381        file.write_all(&line)
382    })();
383
384    let _ = FileExt::unlock(&lock);
385    result
386}
387
388// --- Public API ---
389
390pub fn emit(kind: EventKind) -> u64 {
391    bus().emit(kind)
392}
393
394pub fn events_since(after_id: u64) -> Vec<LeanCtxEvent> {
395    bus().events_since(after_id)
396}
397
398pub fn latest_events(n: usize) -> Vec<LeanCtxEvent> {
399    bus().latest_events(n)
400}
401
402#[derive(Default)]
403struct FileEventCache {
404    path: Option<std::path::PathBuf>,
405    mtime: Option<std::time::SystemTime>,
406    len: u64,
407    events: Vec<LeanCtxEvent>,
408}
409
410/// File-backed event load with a process-local cache keyed on (path, mtime, len).
411/// The dashboard polls this every 3 s; without the cache each poll re-read
412/// and re-parsed the entire JSONL (up to 10k lines) even when nothing changed.
413pub fn load_events_from_file(n: usize) -> Vec<LeanCtxEvent> {
414    static CACHE: OnceLock<Mutex<FileEventCache>> = OnceLock::new();
415    let Some(path) = jsonl_path() else {
416        return Vec::new();
417    };
418    let (mtime, len) = match std::fs::metadata(&path) {
419        Ok(m) => (m.modified().ok(), m.len()),
420        Err(_) => return Vec::new(),
421    };
422
423    let cache = CACHE.get_or_init(|| Mutex::new(FileEventCache::default()));
424    let mut guard = match cache.lock() {
425        Ok(g) => g,
426        Err(poisoned) => poisoned.into_inner(),
427    };
428
429    let fresh =
430        guard.path.as_deref() == Some(path.as_path()) && guard.mtime == mtime && guard.len == len;
431    if !fresh {
432        let Ok(content) = std::fs::read_to_string(&path) else {
433            return Vec::new();
434        };
435        guard.events = content
436            .lines()
437            .filter(|l| !l.trim().is_empty())
438            .filter_map(|l| serde_json::from_str(l).ok())
439            .collect();
440        guard.path = Some(path);
441        guard.mtime = mtime;
442        guard.len = len;
443    }
444
445    let start = guard.events.len().saturating_sub(n);
446    guard.events[start..].to_vec()
447}
448
449pub fn emit_tool_call(
450    tool: &str,
451    tokens_original: u64,
452    tokens_saved: u64,
453    mode: Option<String>,
454    duration_ms: u64,
455    path: Option<String>,
456) {
457    emit(EventKind::ToolCall {
458        tool: tool.to_string(),
459        tokens_original,
460        tokens_saved,
461        mode,
462        duration_ms,
463        path,
464    });
465}
466
467pub fn emit_cache_hit(path: &str, saved_tokens: u64) {
468    emit(EventKind::CacheHit {
469        path: path.to_string(),
470        saved_tokens,
471    });
472}
473
474pub fn emit_agent_action(agent_id: &str, action: &str, tool: Option<&str>) {
475    emit(EventKind::AgentAction {
476        agent_id: agent_id.to_string(),
477        action: action.to_string(),
478        tool: tool.map(std::string::ToString::to_string),
479    });
480}
481
482pub fn emit_budget_warning(role: &str, dimension: &str, used: &str, limit: &str, percent: u8) {
483    emit(EventKind::BudgetWarning {
484        role: role.to_string(),
485        dimension: dimension.to_string(),
486        used: used.to_string(),
487        limit: limit.to_string(),
488        percent,
489    });
490}
491
492pub fn emit_budget_exhausted(role: &str, dimension: &str, used: &str, limit: &str) {
493    emit(EventKind::BudgetExhausted {
494        role: role.to_string(),
495        dimension: dimension.to_string(),
496        used: used.to_string(),
497        limit: limit.to_string(),
498    });
499}
500
501pub fn emit_policy_violation(role: &str, tool: &str, reason: &str) {
502    emit(EventKind::PolicyViolation {
503        role: role.to_string(),
504        tool: tool.to_string(),
505        reason: reason.to_string(),
506    });
507}
508
509pub fn emit_role_changed(from: &str, to: &str) {
510    emit(EventKind::RoleChanged {
511        from: from.to_string(),
512        to: to.to_string(),
513    });
514}
515
516pub fn emit_profile_changed(from: &str, to: &str) {
517    emit(EventKind::ProfileChanged {
518        from: from.to_string(),
519        to: to.to_string(),
520    });
521}
522
523pub fn emit_slo_violation(slo_name: &str, metric: &str, threshold: f64, actual: f64, action: &str) {
524    emit(EventKind::SloViolation {
525        slo_name: slo_name.to_string(),
526        metric: metric.to_string(),
527        threshold,
528        actual,
529        action: action.to_string(),
530    });
531}
532
533pub fn emit_anomaly(metric: &str, expected: f64, actual: f64, deviation_factor: f64) {
534    emit(EventKind::Anomaly {
535        metric: metric.to_string(),
536        expected,
537        actual,
538        deviation_factor,
539    });
540}
541
542pub fn emit_verification_warning(warning_kind: &str, detail: &str, severity: &str) {
543    emit(EventKind::VerificationWarning {
544        warning_kind: warning_kind.to_string(),
545        detail: detail.to_string(),
546        severity: severity.to_string(),
547    });
548}
549
550pub fn emit_threshold_adapted(language: &str, arm: &str, old_threshold: f64, new_threshold: f64) {
551    emit(EventKind::ThresholdAdapted {
552        language: language.to_string(),
553        arm: arm.to_string(),
554        old_threshold,
555        new_threshold,
556    });
557}
558
559#[cfg(test)]
560mod tests {
561    use super::*;
562
563    #[test]
564    fn emit_returns_positive_id() {
565        let id = emit(EventKind::ToolCall {
566            tool: "ctx_read".to_string(),
567            tokens_original: 1000,
568            tokens_saved: 800,
569            mode: Some("map".to_string()),
570            duration_ms: 5,
571            path: Some("src/main.rs".to_string()),
572        });
573        assert!(id > 0);
574        let events = latest_events(100);
575        assert!(events.iter().any(|e| e.id == id));
576    }
577
578    #[test]
579    fn events_since_filters_correctly() {
580        let id1 = emit(EventKind::CacheHit {
581            path: "filter_test_a.rs".to_string(),
582            saved_tokens: 100,
583        });
584        let id2 = emit(EventKind::CacheHit {
585            path: "filter_test_b.rs".to_string(),
586            saved_tokens: 200,
587        });
588
589        let after = events_since(id1);
590        assert!(after.iter().any(|e| e.id == id2));
591        assert!(after.iter().all(|e| e.id > id1));
592    }
593
594    /// The (path, mtime, len) cache must never serve stale events: appending a
595    /// line changes the file length, which has nanosecond-independent
596    /// granularity (unlike mtime), so new events show up on the next poll.
597    #[test]
598    fn load_events_from_file_sees_appended_events() {
599        let path = jsonl_path().expect("test sandbox data dir");
600        if let Some(parent) = path.parent() {
601            std::fs::create_dir_all(parent).expect("create data dir");
602        }
603
604        let line_a = r#"{"id":900001,"timestamp":"2026-06-12T08:00:00.000","kind":{"type":"CacheHit","path":"cached_a.rs","saved_tokens":42}}"#;
605        std::fs::write(&path, format!("{line_a}\n")).expect("write events.jsonl");
606
607        let first = load_events_from_file(50);
608        assert!(
609            first.iter().any(|e| e.id == 900_001),
610            "initial load should parse the seeded event"
611        );
612
613        // Second call with unchanged file exercises the cached branch.
614        let cached = load_events_from_file(50);
615        assert_eq!(cached.len(), first.len());
616
617        let line_b = r#"{"id":900002,"timestamp":"2026-06-12T08:00:01.000","kind":{"type":"CacheHit","path":"cached_b.rs","saved_tokens":7}}"#;
618        {
619            use std::io::Write;
620            let mut f = std::fs::OpenOptions::new()
621                .append(true)
622                .open(&path)
623                .expect("append events.jsonl");
624            writeln!(f, "{line_b}").expect("append line");
625        }
626
627        let second = load_events_from_file(50);
628        assert!(
629            second.iter().any(|e| e.id == 900_002),
630            "append must invalidate the cache and surface the new event"
631        );
632    }
633
634    #[test]
635    fn events_jsonl_writer_child() {
636        let Some(path) = std::env::var_os("__LEAN_CTX_EVENTS_TEST_PATH") else {
637            return;
638        };
639        let writer = std::env::var("__LEAN_CTX_EVENTS_TEST_WRITER")
640            .expect("writer id")
641            .parse::<u64>()
642            .expect("numeric writer id");
643        for sequence in 0..100 {
644            let event = LeanCtxEvent {
645                id: writer * 100 + sequence,
646                timestamp: "2026-07-15T12:00:00.000".to_string(),
647                kind: EventKind::CacheHit {
648                    path: format!("writer-{writer}.rs"),
649                    saved_tokens: sequence,
650                },
651            };
652            append_jsonl_at(std::path::Path::new(&path), &event).expect("append event");
653        }
654    }
655
656    fn test_event(id: u64) -> LeanCtxEvent {
657        LeanCtxEvent {
658            id,
659            timestamp: "2026-07-15T12:00:00.000".to_string(),
660            kind: EventKind::CacheHit {
661                path: "event-id-test.rs".to_string(),
662                saved_tokens: 1,
663            },
664        }
665    }
666
667    #[test]
668    fn concurrent_processes_append_complete_json_lines() {
669        let dir = tempfile::tempdir().expect("temp dir");
670        let path = dir.path().join("events.jsonl");
671        let executable = std::env::current_exe().expect("test executable");
672
673        let mut children = Vec::new();
674        for writer in 0..4 {
675            children.push(
676                std::process::Command::new(&executable)
677                    .args([
678                        "--exact",
679                        "core::events::tests::events_jsonl_writer_child",
680                        "--nocapture",
681                    ])
682                    .env("__LEAN_CTX_EVENTS_TEST_PATH", &path)
683                    .env("__LEAN_CTX_EVENTS_TEST_WRITER", writer.to_string())
684                    .spawn()
685                    .expect("spawn writer"),
686            );
687        }
688        for mut child in children {
689            assert!(child.wait().expect("wait for writer").success());
690        }
691
692        let content = std::fs::read_to_string(&path).expect("read events");
693        let lines: Vec<_> = content.lines().collect();
694        assert_eq!(lines.len(), 400, "every append must produce one line");
695        for line in lines {
696            serde_json::from_str::<LeanCtxEvent>(line)
697                .unwrap_or_else(|error| panic!("invalid JSONL line: {error}: {line}"));
698        }
699    }
700
701    #[test]
702    fn concurrent_processes_serialize_rotation_and_append() {
703        let dir = tempfile::tempdir().expect("temp dir");
704        let path = dir.path().join("events.jsonl");
705        let seed = serde_json::to_string(&LeanCtxEvent {
706            id: 1,
707            timestamp: "2026-07-15T12:00:00.000".to_string(),
708            kind: EventKind::CacheHit {
709                path: "seed.rs".to_string(),
710                saved_tokens: 1,
711            },
712        })
713        .expect("serialize seed");
714        std::fs::write(&path, format!("{seed}\n").repeat(JSONL_MAX_LINES))
715            .expect("seed rotation threshold");
716
717        let executable = std::env::current_exe().expect("test executable");
718        let mut children = Vec::new();
719        for writer in 0..4 {
720            children.push(
721                std::process::Command::new(&executable)
722                    .args(["--exact", "core::events::tests::events_jsonl_writer_child"])
723                    .env("__LEAN_CTX_EVENTS_TEST_PATH", &path)
724                    .env("__LEAN_CTX_EVENTS_TEST_WRITER", writer.to_string())
725                    .spawn()
726                    .expect("spawn writer"),
727            );
728        }
729        for mut child in children {
730            assert!(child.wait().expect("wait for writer").success());
731        }
732
733        let old =
734            std::fs::read_to_string(path.with_extension("jsonl.old")).expect("rotated journal");
735        assert_eq!(old.lines().count(), JSONL_MAX_LINES);
736        assert!(
737            old.lines()
738                .all(|line| serde_json::from_str::<LeanCtxEvent>(line).is_ok()),
739            "rotated journal must contain complete JSON lines"
740        );
741
742        let current = std::fs::read_to_string(&path).expect("current journal");
743        assert_eq!(current.lines().count(), 400);
744        assert!(
745            current
746                .lines()
747                .all(|line| serde_json::from_str::<LeanCtxEvent>(line).is_ok()),
748            "replacement journal must contain complete JSON lines"
749        );
750    }
751
752    #[test]
753    fn persistent_event_id_writer_child() {
754        let Some(dir) = std::env::var_os("__LEAN_CTX_EVENT_ID_TEST_DIR") else {
755            return;
756        };
757        let writer = std::env::var("__LEAN_CTX_EVENT_ID_TEST_WRITER").expect("writer id");
758        let dir = std::path::PathBuf::from(dir);
759        let sequence = dir.join("events.seq");
760        let journal = dir.join("events.jsonl");
761        let (first, last) =
762            reserve_event_id_block_at(&sequence, &journal, 100).expect("reserve event id block");
763        let ids: Vec<String> = (first..=last).map(|id| id.to_string()).collect();
764        std::fs::write(dir.join(format!("writer-{writer}.ids")), ids.join("\n"))
765            .expect("write allocated ids");
766    }
767
768    #[test]
769    fn persistent_event_id_bootstraps_above_existing_journals() {
770        let dir = tempfile::tempdir().expect("temp dir");
771        let sequence = dir.path().join("events.seq");
772        let journal = dir.path().join("events.jsonl");
773        let old = journal.with_extension("jsonl.old");
774        std::fs::write(
775            &journal,
776            format!("{}\n", serde_json::to_string(&test_event(41)).unwrap()),
777        )
778        .unwrap();
779        std::fs::write(
780            &old,
781            format!("{}\n", serde_json::to_string(&test_event(73)).unwrap()),
782        )
783        .unwrap();
784
785        assert_eq!(next_event_id_at(&sequence, &journal).unwrap(), 74);
786        assert_eq!(next_event_id_at(&sequence, &journal).unwrap(), 75);
787    }
788
789    #[test]
790    fn corrupt_sequence_recovers_above_existing_journal() {
791        let dir = tempfile::tempdir().expect("temp dir");
792        let sequence = dir.path().join("events.seq");
793        let journal = dir.path().join("events.jsonl");
794        std::fs::write(&sequence, "76:broken").unwrap();
795        std::fs::write(
796            &journal,
797            format!("{}\n", serde_json::to_string(&test_event(80)).unwrap()),
798        )
799        .unwrap();
800
801        assert_eq!(next_event_id_at(&sequence, &journal).unwrap(), 81);
802        assert_eq!(read_persisted_sequence(&sequence), Some(81));
803    }
804
805    #[test]
806    fn concurrent_processes_allocate_unique_event_ids() {
807        let dir = tempfile::tempdir().expect("temp dir");
808        let executable = std::env::current_exe().expect("test executable");
809        let mut children = Vec::new();
810        for writer in 0..4 {
811            children.push(
812                std::process::Command::new(&executable)
813                    .args([
814                        "--exact",
815                        "core::events::tests::persistent_event_id_writer_child",
816                    ])
817                    .env("__LEAN_CTX_EVENT_ID_TEST_DIR", dir.path())
818                    .env("__LEAN_CTX_EVENT_ID_TEST_WRITER", writer.to_string())
819                    .spawn()
820                    .expect("spawn event id writer"),
821            );
822        }
823        for mut child in children {
824            assert!(child.wait().expect("wait for event id writer").success());
825        }
826
827        let mut ids = Vec::new();
828        for writer in 0..4 {
829            let content = std::fs::read_to_string(dir.path().join(format!("writer-{writer}.ids")))
830                .expect("read allocated ids");
831            ids.extend(content.lines().map(|line| line.parse::<u64>().unwrap()));
832        }
833        ids.sort_unstable();
834        ids.dedup();
835        assert_eq!(ids.len(), 400, "every process must receive unique ids");
836        assert_eq!(ids.first(), Some(&1));
837        assert_eq!(ids.last(), Some(&400));
838        let sequence_path = dir.path().join("events.seq");
839        assert_eq!(read_persisted_sequence(&sequence_path), Some(400));
840        assert_eq!(
841            std::fs::read_to_string(sequence_path)
842                .expect("read sequence journal")
843                .lines()
844                .count(),
845            4,
846            "four writers should persist four block reservations, not 400 ids"
847        );
848    }
849}
lean_ctx/core/events.rs

lean_ctx/core/
events.rs
