aprender-profile 0.29.0

//! Lamport logical clocks for causal ordering (Sprint 40 - Toyota Way: Poka-Yoke)
//!
//! This module implements Lamport logical clocks to establish **mathematically guaranteed**
//! causal ordering across distributed traces, eliminating false causality from clock skew.
//!
//! # Toyota Way Principle: Poka-Yoke (Error-Proofing)
//!
//! Physical timestamps ($`t_A` < `t_B`$) are unreliable for establishing causality due to:
//! - Clock skew across CPU cores (100-1000ns)
//! - NTP drift across machines (milliseconds)
//! - GPU async execution (kernel launch returns before kernel completes)
//!
//! Lamport clocks provide a **mathematical guarantee**: if event A causally precedes event B,
//! then `logical_clock(A) < logical_clock(B)`.
//!
//! # Peer-Reviewed Foundation
//!
//! **Lamport (1978). "Time, Clocks, and the Ordering of Events in a Distributed System." CACM.**
//! - **Theorem:** Event A → B iff `logical_clock(A)` < `logical_clock(B)`
//! - **Application:** Eliminates false causal edges from timestamp inference
//!
//! **Chow et al. (2014). "The Mystery Machine: End-to-end Performance Analysis." OSDI.**
//! - **Finding:** Timestamp-based causality has 15-30% false positive rate
//! - **Application:** Logical clocks provide provable correctness
//!
//! # Design
//!
//! ```text
//! ┌────────────────────────────────────────────────────────────────┐
//! │ Parent Process (trace_id: 4bf92f..., logical_clock: 42)       │
//! │   syscall() → GLOBAL_CLOCK.tick() → logical_clock = 43        │
//! └────────────────────────────────────────────────────────────────┘
//!                          │
//!                          │ fork() + env::set_var("RENACER_LOGICAL_CLOCK", "43")
//!                          ▼
//! ┌────────────────────────────────────────────────────────────────┐
//! │ Child Process                                                   │
//! │   on_init() → GLOBAL_CLOCK.sync(43)  // Inherit parent clock  │
//! │   syscall() → GLOBAL_CLOCK.tick() → logical_clock = 44        │
//! └────────────────────────────────────────────────────────────────┘
//!
//! Causal relationship proven by: 43 < 44 (Lamport ordering)
//! ```
//!
//! # Example
//!
//! ```no_run
//! use renacer::lamport_clock::LamportClock;
//!
//! // Create global clock
//! static GLOBAL_CLOCK: LamportClock = LamportClock::new();
//!
//! // Local event: increment clock
//! let t1 = GLOBAL_CLOCK.tick();
//!
//! // Send message to another process
//! // send_message(data, t1);
//!
//! // Receive message: synchronize clock
//! // let (data, remote_time) = receive_message();
//! // GLOBAL_CLOCK.sync(remote_time);
//! let t2 = GLOBAL_CLOCK.tick();
//!
//! // Guaranteed: t1 < t2 (causal ordering)
//! assert!(t1 < t2);
//! ```

use std::sync::atomic::{AtomicU64, Ordering};

/// Lamport logical clock for causal ordering
///
/// This is a monotonically increasing counter that establishes happens-before
/// relationships between events in a distributed system.
///
/// # Thread Safety
///
/// `LamportClock` is thread-safe and uses atomic operations for lock-free
/// increments. It can be safely shared across threads via `Arc` or used as
/// a global static.
///
/// # Performance
///
/// - `tick()`: ~10ns (single atomic `fetch_add`)
/// - `sync()`: ~20ns (atomic `fetch_max` + branch)
#[derive(Debug)]
pub struct LamportClock {
    counter: AtomicU64,
}

impl LamportClock {
    /// Create a new Lamport clock starting at 0
    ///
    /// # Example
    ///
    /// ```
    /// use renacer::lamport_clock::LamportClock;
    ///
    /// static CLOCK: LamportClock = LamportClock::new();
    /// ```
    pub const fn new() -> Self {
        Self { counter: AtomicU64::new(0) }
    }

    /// Create a new Lamport clock with a specific starting value
    ///
    /// This is useful when resuming from a saved state or inheriting a
    /// logical clock from a parent process.
    ///
    /// # Arguments
    ///
    /// * `initial_value` - Starting value for the clock
    ///
    /// # Example
    ///
    /// ```
    /// use renacer::lamport_clock::LamportClock;
    ///
    /// // Child process inherits parent's logical clock
    /// let parent_clock = std::env::var("RENACER_LOGICAL_CLOCK")
    ///     .ok()
    ///     .and_then(|s| s.parse().ok())
    ///     .unwrap_or(0);
    ///
    /// let clock = LamportClock::with_value(parent_clock);
    /// ```
    pub const fn with_value(initial_value: u64) -> Self {
        Self { counter: AtomicU64::new(initial_value) }
    }

    /// Increment the clock on a local event (happens-before tick)
    ///
    /// This should be called whenever the current process performs an event
    /// that should be recorded in the causal order (e.g., syscall, span start).
    ///
    /// # Returns
    ///
    /// The new logical timestamp after incrementing.
    ///
    /// # Performance
    ///
    /// ~10ns (single atomic `fetch_add` with `SeqCst` ordering)
    ///
    /// # Example
    ///
    /// ```
    /// use renacer::lamport_clock::LamportClock;
    ///
    /// static CLOCK: LamportClock = LamportClock::new();
    ///
    /// fn on_syscall() {
    ///     let logical_time = CLOCK.tick();
    ///     println!("Syscall at logical time: {}", logical_time);
    /// }
    /// ```
    pub fn tick(&self) -> u64 {
        self.counter.fetch_add(1, Ordering::SeqCst)
    }

    /// Synchronize the clock on message receive (happens-before sync)
    ///
    /// This implements Lamport's synchronization rule:
    /// ```text
    /// local_clock = max(local_clock, remote_clock) + 1
    /// ```
    ///
    /// This should be called when receiving a message/event from another
    /// process or thread (e.g., fork, RPC response, span context propagation).
    ///
    /// # Arguments
    ///
    /// * `remote_clock` - The logical timestamp from the remote event
    ///
    /// # Performance
    ///
    /// ~20ns (atomic `fetch_max` + `fetch_add`)
    ///
    /// # Example
    ///
    /// ```
    /// use renacer::lamport_clock::LamportClock;
    ///
    /// static CLOCK: LamportClock = LamportClock::new();
    ///
    /// fn on_fork_child() {
    ///     // Inherit parent's logical clock from environment
    ///     if let Ok(parent_clock_str) = std::env::var("RENACER_LOGICAL_CLOCK") {
    ///         if let Ok(parent_clock) = parent_clock_str.parse::<u64>() {
    ///             CLOCK.sync(parent_clock);
    ///         }
    ///     }
    /// }
    /// ```
    pub fn sync(&self, remote_clock: u64) {
        // Atomic max operation: local_clock = max(local_clock, remote_clock)
        self.counter.fetch_max(remote_clock, Ordering::SeqCst);

        // Increment by 1 (Lamport rule)
        self.counter.fetch_add(1, Ordering::SeqCst);
    }

    /// Get the current logical timestamp without incrementing
    ///
    /// This is useful for reading the clock value without advancing it
    /// (e.g., for logging, debugging, or exporting to child processes).
    ///
    /// # Returns
    ///
    /// The current logical timestamp.
    ///
    /// # Example
    ///
    /// ```
    /// use renacer::lamport_clock::LamportClock;
    ///
    /// static CLOCK: LamportClock = LamportClock::new();
    ///
    /// fn propagate_to_child() {
    ///     let current_time = CLOCK.now();
    ///     std::env::set_var("RENACER_LOGICAL_CLOCK", current_time.to_string());
    ///     // fork() or spawn child process
    /// }
    /// ```
    pub fn now(&self) -> u64 {
        self.counter.load(Ordering::SeqCst)
    }

    /// Reset the clock to 0
    ///
    /// This is primarily useful for testing. In production, clocks should
    /// never be reset as it violates monotonicity.
    #[cfg(test)]
    pub(crate) fn reset(&self) {
        self.counter.store(0, Ordering::SeqCst);
    }
}

impl Default for LamportClock {
    fn default() -> Self {
        Self::new()
    }
}

/// Global Lamport clock for renacer
///
/// This is used throughout the tracing system to assign logical timestamps
/// to spans and events.
///
/// # Example
///
/// ```
/// use renacer::lamport_clock::GLOBAL_CLOCK;
///
/// fn record_span() {
///     let logical_time = GLOBAL_CLOCK.tick();
///     println!("Span at logical time: {}", logical_time);
/// }
/// ```
pub static GLOBAL_CLOCK: LamportClock = LamportClock::new();

/// Initialize the global clock from environment variable (for child processes)
///
/// This should be called early in the process initialization to inherit the
/// parent process's logical clock.
///
/// # Example
///
/// ```
/// use renacer::lamport_clock::init_from_env;
///
/// // Initialize logical clock from parent process
/// init_from_env();
///
/// // Rest of program...
/// ```
pub fn init_from_env() {
    if let Ok(clock_str) = std::env::var("RENACER_LOGICAL_CLOCK") {
        if let Ok(parent_clock) = clock_str.parse::<u64>() {
            GLOBAL_CLOCK.sync(parent_clock);
            eprintln!("DEBUG: Initialized Lamport clock from parent: {parent_clock}");
        }
    }
}

/// Propagate the global clock to a child process via environment variable
///
/// This should be called before forking or spawning a child process.
///
/// # Example
///
/// ```no_run
/// use renacer::lamport_clock::propagate_to_env;
/// use std::process::Command;
///
/// fn spawn_child() {
///     propagate_to_env();
///
///     Command::new("./child_process")
///         .spawn()
///         .expect("Failed to spawn child");
/// }
/// ```
pub fn propagate_to_env() {
    let current_time = GLOBAL_CLOCK.now();
    std::env::set_var("RENACER_LOGICAL_CLOCK", current_time.to_string());
}

#[cfg(test)]
mod tests {
    use super::*;
    use serial_test::serial;

    #[test]
    fn test_tick_increments() {
        let clock = LamportClock::new();
        assert_eq!(clock.tick(), 0);
        assert_eq!(clock.tick(), 1);
        assert_eq!(clock.tick(), 2);
        assert_eq!(clock.now(), 3);
    }

    #[test]
    fn test_sync_with_higher_remote_clock() {
        let clock = LamportClock::new();
        clock.tick(); // local = 1

        // Receive message from process with clock = 10
        clock.sync(10);

        // After sync: local = max(1, 10) + 1 = 11
        assert_eq!(clock.now(), 11);
    }

    #[test]
    fn test_sync_with_lower_remote_clock() {
        let clock = LamportClock::new();
        for _ in 0..5 {
            clock.tick(); // local = 5
        }

        // Receive message from process with clock = 2
        clock.sync(2);

        // After sync: local = max(5, 2) + 1 = 6
        assert_eq!(clock.now(), 6);
    }

    #[test]
    fn test_happens_before_ordering() {
        let clock_a = LamportClock::new();
        let clock_b = LamportClock::new();

        // Process A: local events
        let t1 = clock_a.tick(); // t1 = 0
        let t2 = clock_a.tick(); // t2 = 1

        // Process A sends message to Process B
        // (t2 is sent with the message)

        // Process B receives message and syncs
        clock_b.sync(t2); // clock_b = max(0, 1) + 1 = 2
        let t3 = clock_b.tick(); // t3 = 2

        // Verify causal ordering: t1 < t2 < t3
        assert!(t1 < t2);
        assert!(t2 < t3);
    }

    #[test]
    fn test_with_value() {
        let clock = LamportClock::with_value(100);
        assert_eq!(clock.now(), 100);
        assert_eq!(clock.tick(), 100);
        assert_eq!(clock.now(), 101);
    }

    #[test]
    fn test_reset() {
        let clock = LamportClock::new();
        clock.tick();
        clock.tick();
        assert_eq!(clock.now(), 2);

        clock.reset();
        assert_eq!(clock.now(), 0);
    }

    #[test]
    fn test_thread_safety() {
        use std::sync::Arc;
        use std::thread;

        let clock = Arc::new(LamportClock::new());
        let mut handles = vec![];

        // Spawn 10 threads, each incrementing 100 times
        for _ in 0..10 {
            let clock_clone = clock.clone();
            let handle = thread::spawn(move || {
                for _ in 0..100 {
                    clock_clone.tick();
                }
            });
            handles.push(handle);
        }

        for handle in handles {
            handle.join().expect("test");
        }

        // Final clock value should be 1000
        assert_eq!(clock.now(), 1000);
    }

    #[test]
    fn test_env_propagation() {
        let clock = LamportClock::new();
        for _ in 0..42 {
            clock.tick();
        }

        // Simulate propagating to child process
        let current = clock.now();
        std::env::set_var("TEST_RENACER_LOGICAL_CLOCK", current.to_string());

        // Simulate child process reading from env
        let clock_str = std::env::var("TEST_RENACER_LOGICAL_CLOCK").expect("test");
        let parent_clock = clock_str.parse::<u64>().expect("test");

        let child_clock = LamportClock::new();
        child_clock.sync(parent_clock);

        // Child should start after parent
        assert!(child_clock.now() > current);
        assert_eq!(child_clock.now(), 43); // max(0, 42) + 1
    }

    #[test]
    fn test_default_trait() {
        let clock = LamportClock::default();
        assert_eq!(clock.now(), 0);
        assert_eq!(clock.tick(), 0);
        assert_eq!(clock.now(), 1);
    }

    #[test]
    #[serial]
    fn test_propagate_to_env() {
        // Reset global clock for test isolation
        GLOBAL_CLOCK.reset();

        // Tick a few times
        GLOBAL_CLOCK.tick();
        GLOBAL_CLOCK.tick();
        GLOBAL_CLOCK.tick();

        // Propagate to env
        propagate_to_env();

        // Verify env var was set
        let env_val = std::env::var("RENACER_LOGICAL_CLOCK").expect("test");
        assert_eq!(env_val, "3");
    }

    #[test]
    #[serial]
    fn test_init_from_env_with_valid_clock() {
        // Reset global clock for test isolation
        GLOBAL_CLOCK.reset();

        // Set env var simulating parent clock
        std::env::set_var("RENACER_LOGICAL_CLOCK", "50");

        // Init from env
        init_from_env();

        // Global clock should be synced: max(0, 50) + 1 = 51
        assert_eq!(GLOBAL_CLOCK.now(), 51);

        // Clean up
        std::env::remove_var("RENACER_LOGICAL_CLOCK");
        GLOBAL_CLOCK.reset();
    }

    #[test]
    #[serial]
    fn test_init_from_env_with_missing_env() {
        // Reset global clock for test isolation
        GLOBAL_CLOCK.reset();

        // Ensure env var is not set
        std::env::remove_var("RENACER_LOGICAL_CLOCK");

        // Init from env - should do nothing
        init_from_env();

        // Global clock should remain at 0
        assert_eq!(GLOBAL_CLOCK.now(), 0);
    }

    #[test]
    #[serial]
    fn test_init_from_env_with_invalid_value() {
        // Reset global clock for test isolation
        GLOBAL_CLOCK.reset();

        // Set env var to invalid value
        std::env::set_var("RENACER_LOGICAL_CLOCK", "not_a_number");

        // Init from env - should do nothing due to parse error
        init_from_env();

        // Global clock should remain at 0
        assert_eq!(GLOBAL_CLOCK.now(), 0);

        // Clean up
        std::env::remove_var("RENACER_LOGICAL_CLOCK");
    }

    #[test]
    #[serial]
    fn test_global_clock_tick() {
        // Reset for isolation
        GLOBAL_CLOCK.reset();

        let t1 = GLOBAL_CLOCK.tick();
        let t2 = GLOBAL_CLOCK.tick();
        let t3 = GLOBAL_CLOCK.tick();

        assert_eq!(t1, 0);
        assert_eq!(t2, 1);
        assert_eq!(t3, 2);
        assert_eq!(GLOBAL_CLOCK.now(), 3);

        GLOBAL_CLOCK.reset();
    }
}