laminar_core/tpc/

core_handle.rs

//! # Core Handle
//!
//! Manages a single core's reactor thread with SPSC queue communication.
//!
//! ## Architecture
//!
//! Each `CoreHandle` spawns a dedicated thread that:
//! 1. Sets CPU affinity to pin to a specific core
//! 2. Creates a `Reactor` with its own state partition
//! 3. Optionally initializes an `io_uring` ring for async I/O (Linux only)
//! 4. Drains the inbox SPSC queue for incoming events
//! 5. Processes events through the reactor
//! 6. Pushes outputs to the outbox SPSC queue
//!
//! ## Communication
//!
//! - **Inbox**: Main thread → Core thread (events, watermarks, commands)
//! - **Outbox**: Core thread → Main thread (outputs)
//!
//! Both use lock-free SPSC queues for minimal latency.
//!
//! ## `io_uring` Integration
//!
//! On Linux with the `io-uring` feature enabled, each core thread can have its own
//! `io_uring` ring for high-performance async I/O. This enables:
//! - SQPOLL mode for syscall-free submission
//! - Registered buffers for zero-copy I/O
//! - Per-core I/O isolation for thread-per-core architecture

use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::thread::{self, JoinHandle};

use tokio::sync::Notify;

#[cfg(all(target_os = "linux", feature = "io-uring"))]
use crate::io_uring::IoUringConfig;

use crate::alloc::HotPathGuard;
use crate::budget::TaskBudget;
use crate::checkpoint::CheckpointBarrier;
use crate::numa::NumaTopology;
use crate::operator::{CheckpointCompleteData, Event, Operator, OperatorState, Output};
use crate::reactor::{Reactor, ReactorConfig};
use crate::storage_io::{IoCompletion, StorageIo, SyncStorageIo};

use super::backpressure::{
    BackpressureConfig, CreditAcquireResult, CreditGate, CreditMetrics, OverflowStrategy,
};
use super::spsc::SpscQueue;
use super::TpcError;

/// Messages sent to a core thread.
#[derive(Debug)]
pub enum CoreMessage {
    /// Process an event from a specific source.
    Event {
        /// Index of the source that produced this event.
        source_idx: usize,
        /// The event data.
        event: Event,
    },
    /// Watermark advancement
    Watermark(i64),
    /// Request a checkpoint
    CheckpointRequest(u64),
    /// Checkpoint barrier from a source.
    ///
    /// The core thread flushes the reactor before forwarding this
    /// as `Output::Barrier` so the coordinator can track alignment.
    Barrier {
        /// Index of the source that emitted this barrier.
        source_idx: usize,
        /// The checkpoint barrier.
        barrier: CheckpointBarrier,
    },
    /// Graceful shutdown
    Shutdown,
}

/// Output tagged with its source index for coordinator routing.
///
/// Core threads wrap every `Output` in a `TaggedOutput` so the
/// coordinator can map outputs back to the originating source.
#[derive(Debug)]
pub struct TaggedOutput {
    /// Index of the source that produced this output.
    pub source_idx: usize,
    /// The operator output.
    pub output: Output,
}

/// Configuration for a core handle.
#[derive(Debug, Clone)]
pub struct CoreConfig {
    /// Core ID (0-indexed)
    pub core_id: usize,
    /// CPU ID to pin to (None = no pinning)
    pub cpu_affinity: Option<usize>,
    /// Inbox queue capacity
    pub inbox_capacity: usize,
    /// Outbox queue capacity
    pub outbox_capacity: usize,
    /// Reactor configuration
    pub reactor_config: ReactorConfig,
    /// Backpressure configuration
    pub backpressure: BackpressureConfig,
    /// Enable NUMA-aware memory allocation
    pub numa_aware: bool,
    /// Enable per-core storage I/O backend.
    ///
    /// When true, creates a [`StorageIo`] instance for this core. On Linux
    /// with the `io-uring` feature and a supported kernel, this uses
    /// `io_uring` with SQPOLL. Otherwise falls back to synchronous I/O.
    pub enable_storage_io: bool,
    /// `io_uring` configuration (Linux only, requires `io-uring` feature)
    #[cfg(all(target_os = "linux", feature = "io-uring"))]
    pub io_uring_config: Option<IoUringConfig>,
}

impl Default for CoreConfig {
    fn default() -> Self {
        Self {
            core_id: 0,
            cpu_affinity: None,
            inbox_capacity: 8192,
            outbox_capacity: 8192,
            reactor_config: ReactorConfig::default(),
            backpressure: BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        }
    }
}

/// Handle to a core's reactor thread.
///
/// Provides lock-free communication with the core via SPSC queues
/// and credit-based flow control for backpressure.
pub struct CoreHandle {
    /// Core ID
    core_id: usize,
    /// NUMA node for this core
    numa_node: usize,
    /// Inbox queue (main thread writes, core reads)
    inbox: Arc<SpscQueue<CoreMessage>>,
    /// Outbox queue (core writes, main thread reads)
    outbox: Arc<SpscQueue<TaggedOutput>>,
    /// Storage I/O completion queue (Ring 0 writes, Ring 2 reads).
    /// Ring 0 pushes completions from `StorageIo::poll_completions`.
    /// Ring 2 drains during checkpoint to feed `PerCoreWalManager::check_all_completions`.
    io_completion_outbox: Arc<SpscQueue<IoCompletion>>,
    /// Credit gate for backpressure (sender acquires, receiver releases)
    credit_gate: Arc<CreditGate>,
    /// Thread handle (None if thread hasn't started or has been joined)
    thread: Option<JoinHandle<Result<(), TpcError>>>,
    /// Core thread's `Thread` handle for `unpark()` wakeups.
    core_thread_handle: thread::Thread,
    /// Shutdown flag
    shutdown: Arc<AtomicBool>,
    /// Events processed counter
    events_processed: Arc<AtomicU64>,
    /// Outputs dropped due to full outbox
    outputs_dropped: Arc<AtomicU64>,
    /// Running state
    is_running: Arc<AtomicBool>,
}

impl CoreHandle {
    /// Spawns a new core thread with the given configuration.
    ///
    /// # Errors
    ///
    /// Returns an error if the thread cannot be spawned.
    pub fn spawn(config: CoreConfig) -> Result<Self, TpcError> {
        let notify = Arc::new(Notify::new());
        let flag = Arc::new(AtomicBool::new(false));
        Self::spawn_with_notify(config, Vec::new(), flag, notify)
    }

    /// Spawns a new core thread with operators.
    ///
    /// # Errors
    ///
    /// Returns an error if the thread cannot be spawned.
    #[allow(clippy::needless_pass_by_value)]
    pub fn spawn_with_operators(
        config: CoreConfig,
        operators: Vec<Box<dyn Operator>>,
    ) -> Result<Self, TpcError> {
        let notify = Arc::new(Notify::new());
        let flag = Arc::new(AtomicBool::new(false));
        Self::spawn_with_notify(config, operators, flag, notify)
    }

    /// Spawns a new core thread with operators and shared signaling.
    ///
    /// `has_new_data` is a lock-free flag set by the core thread when
    /// outputs are available. Only the false→true transition triggers
    /// `output_notify.notify_one()`, ensuring at most one mutex
    /// acquisition per coordinator drain cycle across all cores.
    ///
    /// # Errors
    ///
    /// Returns an error if the thread cannot be spawned.
    #[allow(clippy::needless_pass_by_value)]
    pub fn spawn_with_notify(
        config: CoreConfig,
        operators: Vec<Box<dyn Operator>>,
        has_new_data: Arc<AtomicBool>,
        output_notify: Arc<Notify>,
    ) -> Result<Self, TpcError> {
        let core_id = config.core_id;
        let cpu_affinity = config.cpu_affinity;
        let reactor_config = config.reactor_config.clone();

        // Detect NUMA topology for NUMA-aware allocation
        let topology = NumaTopology::detect();
        let numa_node =
            cpu_affinity.map_or_else(|| topology.current_node(), |cpu| topology.node_for_cpu(cpu));

        let inbox = Arc::new(SpscQueue::new(config.inbox_capacity));
        let outbox: Arc<SpscQueue<TaggedOutput>> = Arc::new(SpscQueue::new(config.outbox_capacity));
        // I/O completion queue: Ring 0 produces, Ring 2 consumes during checkpoint.
        // Capacity 256 matches default io_uring ring_entries.
        let io_completion_outbox: Arc<SpscQueue<IoCompletion>> = Arc::new(SpscQueue::new(256));
        let credit_gate = Arc::new(CreditGate::new(config.backpressure.clone()));
        let shutdown = Arc::new(AtomicBool::new(false));
        let events_processed = Arc::new(AtomicU64::new(0));
        let outputs_dropped = Arc::new(AtomicU64::new(0));
        let is_running = Arc::new(AtomicBool::new(false));

        // Create per-core storage I/O backend (cold path — allocation is fine here).
        let storage_io: Option<Box<dyn StorageIo>> = if config.enable_storage_io {
            // On Linux with io-uring feature, try io_uring first.
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            {
                if let Some(ref uring_cfg) = config.io_uring_config {
                    match crate::storage_io::UringStorageIo::new(core_id, uring_cfg) {
                        Ok(backend) => {
                            tracing::info!(
                                "Core {core_id}: using io_uring storage I/O (SQPOLL={})",
                                backend.uses_sqpoll()
                            );
                            Some(Box::new(backend))
                        }
                        Err(e) => {
                            tracing::warn!(
                                "Core {core_id}: io_uring init failed ({e}), falling back to sync"
                            );
                            Some(Box::new(SyncStorageIo::new()))
                        }
                    }
                } else {
                    Some(Box::new(SyncStorageIo::new()))
                }
            }
            // On non-Linux or without io-uring feature, use sync backend.
            #[cfg(not(all(target_os = "linux", feature = "io-uring")))]
            {
                Some(Box::new(SyncStorageIo::new()))
            }
        } else {
            None
        };

        let thread_context = CoreThreadContext {
            core_id,
            cpu_affinity,
            reactor_config,
            numa_aware: config.numa_aware,
            numa_node,
            #[cfg(target_os = "linux")]
            numa_topology: topology,
            inbox: Arc::clone(&inbox),
            outbox: Arc::clone(&outbox),
            credit_gate: Arc::clone(&credit_gate),
            shutdown: Arc::clone(&shutdown),
            events_processed: Arc::clone(&events_processed),
            outputs_dropped: Arc::clone(&outputs_dropped),
            is_running: Arc::clone(&is_running),
            has_new_data: Arc::clone(&has_new_data),
            output_notify: Arc::clone(&output_notify),
        };

        let thread = thread::Builder::new()
            .name(format!("laminar-core-{core_id}"))
            .spawn({
                let io_cq = Arc::clone(&io_completion_outbox);
                move || core_thread_main(&thread_context, operators, storage_io, io_cq)
            })
            .map_err(|e| TpcError::SpawnFailed {
                core_id,
                message: e.to_string(),
            })?;

        // Capture the thread handle for unpark() wakeups before we store
        // the JoinHandle (which consumes it on join).
        let core_thread_handle = thread.thread().clone();

        // Wait for thread to signal it's running
        while !is_running.load(Ordering::Acquire) {
            thread::yield_now();
        }

        Ok(Self {
            core_id,
            numa_node,
            inbox,
            outbox,
            io_completion_outbox,
            credit_gate,
            thread: Some(thread),
            core_thread_handle,
            shutdown,
            events_processed,
            outputs_dropped,
            is_running,
        })
    }

    /// Returns the core ID.
    #[must_use]
    pub fn core_id(&self) -> usize {
        self.core_id
    }

    /// Returns the NUMA node for this core.
    #[must_use]
    pub fn numa_node(&self) -> usize {
        self.numa_node
    }

    /// Returns true if the core thread is running.
    #[must_use]
    pub fn is_running(&self) -> bool {
        self.is_running.load(Ordering::Acquire)
    }

    /// Returns the core thread's handle for `unpark()` wakeups.
    ///
    /// Source I/O threads call `unpark()` on this handle after pushing
    /// to the inbox SPSC queue, waking the core from `park_timeout()`.
    #[must_use]
    pub fn core_thread_handle(&self) -> &thread::Thread {
        &self.core_thread_handle
    }

    /// Returns the number of events processed by this core.
    #[must_use]
    pub fn events_processed(&self) -> u64 {
        self.events_processed.load(Ordering::Relaxed)
    }

    /// Returns the number of outputs dropped due to a full outbox.
    #[must_use]
    pub fn outputs_dropped(&self) -> u64 {
        self.outputs_dropped.load(Ordering::Relaxed)
    }

    /// Sends a message to the core with credit-based flow control.
    ///
    /// This method respects the backpressure configuration:
    /// - `Block`: Spins until credits available, then sends
    /// - `Drop`: Returns Ok but drops the message if no credits
    /// - `Error`: Returns error if no credits available
    ///
    /// # Errors
    ///
    /// Returns an error if the inbox queue is full or credits exhausted (with Error strategy).
    pub fn send(&self, message: CoreMessage) -> Result<(), TpcError> {
        // Try to acquire a credit
        match self.credit_gate.try_acquire() {
            CreditAcquireResult::Acquired => {
                // Have credit, try to push
                self.inbox.push(message).map_err(|_| TpcError::QueueFull {
                    core_id: self.core_id,
                })
            }
            CreditAcquireResult::WouldBlock => {
                // Check overflow strategy
                if self.credit_gate.config().overflow_strategy == OverflowStrategy::Block {
                    // Spin until we get a credit
                    self.credit_gate.acquire_blocking(1);
                    self.inbox.push(message).map_err(|_| TpcError::QueueFull {
                        core_id: self.core_id,
                    })
                } else {
                    // Error strategy - return error
                    Err(TpcError::Backpressure {
                        core_id: self.core_id,
                    })
                }
            }
            CreditAcquireResult::Dropped => {
                // Drop strategy - silently drop, already recorded in metrics
                Ok(())
            }
        }
    }

    /// Tries to send a message without blocking.
    ///
    /// Returns `Err` if no credits available or queue is full.
    /// Does not block regardless of overflow strategy.
    ///
    /// # Errors
    ///
    /// Returns an error if credits exhausted or queue full.
    pub fn try_send(&self, message: CoreMessage) -> Result<(), TpcError> {
        match self.credit_gate.try_acquire() {
            CreditAcquireResult::Acquired => {
                self.inbox.push(message).map_err(|_| TpcError::QueueFull {
                    core_id: self.core_id,
                })
            }
            CreditAcquireResult::WouldBlock | CreditAcquireResult::Dropped => {
                Err(TpcError::Backpressure {
                    core_id: self.core_id,
                })
            }
        }
    }

    /// Sends an event to the core with credit-based flow control.
    ///
    /// The `source_idx` identifies which source produced this event,
    /// allowing the coordinator to route outputs back to the correct source.
    ///
    /// # Errors
    ///
    /// Returns an error if the inbox queue is full or backpressure applies.
    pub fn send_event(&self, source_idx: usize, event: Event) -> Result<(), TpcError> {
        self.send(CoreMessage::Event { source_idx, event })
    }

    /// Tries to send an event without blocking.
    ///
    /// # Errors
    ///
    /// Returns an error if credits exhausted or queue full.
    pub fn try_send_event(&self, source_idx: usize, event: Event) -> Result<(), TpcError> {
        self.try_send(CoreMessage::Event { source_idx, event })
    }

    /// Returns a reference to the inbox queue (for direct SPSC push from I/O threads).
    #[must_use]
    pub fn inbox(&self) -> &Arc<SpscQueue<CoreMessage>> {
        &self.inbox
    }

    /// Returns a reference to the outbox queue (for direct SPSC pop).
    #[must_use]
    pub fn outbox(&self) -> &Arc<SpscQueue<TaggedOutput>> {
        &self.outbox
    }

    /// Polls the outbox for tagged outputs.
    ///
    /// Returns up to `max_count` outputs, each tagged with the source index.
    ///
    /// # Note
    ///
    /// This method allocates memory. For zero-allocation polling, use
    /// [`poll_outputs_into`](Self::poll_outputs_into) or [`poll_each`](Self::poll_each) instead.
    #[must_use]
    pub fn poll_outputs(&self, max_count: usize) -> Vec<TaggedOutput> {
        self.outbox.pop_batch(max_count)
    }

    /// Polls the outbox for tagged outputs into a pre-allocated buffer (zero-allocation).
    ///
    /// Outputs are appended to `buffer`. Returns the number of outputs added.
    /// The buffer should have sufficient capacity to avoid reallocation.
    #[inline]
    pub fn poll_outputs_into(&self, buffer: &mut Vec<TaggedOutput>, max_count: usize) -> usize {
        let start_len = buffer.len();

        self.outbox.pop_each(max_count, |output| {
            buffer.push(output);
            true
        });

        buffer.len() - start_len
    }

    /// Polls the outbox with a callback for each tagged output (zero-allocation).
    ///
    /// Processing stops when:
    /// - `max_count` outputs have been processed
    /// - The outbox becomes empty
    /// - The callback returns `false`
    ///
    /// Returns the number of outputs processed.
    #[inline]
    pub fn poll_each<F>(&self, max_count: usize, f: F) -> usize
    where
        F: FnMut(TaggedOutput) -> bool,
    {
        self.outbox.pop_each(max_count, f)
    }

    /// Polls a single tagged output from the outbox.
    #[must_use]
    pub fn poll_output(&self) -> Option<TaggedOutput> {
        self.outbox.pop()
    }

    /// Drain storage I/O completions from this core.
    ///
    /// Ring 0 pushes completions from `StorageIo::poll_completions` into a
    /// per-core SPSC queue. Ring 2 (checkpoint coordinator) calls this to
    /// retrieve them and pass to `PerCoreWalManager::check_all_completions`.
    pub fn drain_io_completions(&self, out: &mut Vec<IoCompletion>) {
        self.io_completion_outbox.pop_each(256, |c| {
            out.push(c);
            true
        });
    }

    /// Returns the number of pending messages in the inbox.
    #[must_use]
    pub fn inbox_len(&self) -> usize {
        self.inbox.len()
    }

    /// Returns the number of pending outputs in the outbox.
    #[must_use]
    pub fn outbox_len(&self) -> usize {
        self.outbox.len()
    }

    /// Returns true if backpressure is currently active.
    #[must_use]
    pub fn is_backpressured(&self) -> bool {
        self.credit_gate.is_backpressured()
    }

    /// Returns the number of available credits.
    #[must_use]
    pub fn available_credits(&self) -> usize {
        self.credit_gate.available()
    }

    /// Returns the maximum credits configured.
    #[must_use]
    pub fn max_credits(&self) -> usize {
        self.credit_gate.max_credits()
    }

    /// Returns the credit metrics.
    #[must_use]
    pub fn credit_metrics(&self) -> &CreditMetrics {
        self.credit_gate.metrics()
    }

    /// Signals the core to shut down gracefully.
    pub fn shutdown(&self) {
        self.shutdown.store(true, Ordering::Release);
        // Also send a shutdown message to wake up the thread
        let _ = self.inbox.push(CoreMessage::Shutdown);
        // Unpark the core thread in case it's sleeping in park_timeout()
        self.core_thread_handle.unpark();
    }

    /// Waits for the core thread to finish.
    ///
    /// # Errors
    ///
    /// Returns an error if the thread panicked or returned an error.
    pub fn join(mut self) -> Result<(), TpcError> {
        if let Some(handle) = self.thread.take() {
            handle.join().map_err(|_| TpcError::SpawnFailed {
                core_id: self.core_id,
                message: "Thread panicked".to_string(),
            })?
        } else {
            Ok(())
        }
    }

    /// Sends a shutdown signal and waits for the thread to finish.
    ///
    /// # Errors
    ///
    /// Returns an error if the thread cannot be joined cleanly.
    pub fn shutdown_and_join(self) -> Result<(), TpcError> {
        self.shutdown();
        self.join()
    }
}

impl Drop for CoreHandle {
    fn drop(&mut self) {
        // Signal shutdown if not already done
        self.shutdown.store(true, Ordering::Release);
        // Try to send shutdown message (may fail if queue is full, that's OK)
        let _ = self.inbox.push(CoreMessage::Shutdown);
        // Unpark the core thread in case it's sleeping in park_timeout()
        self.core_thread_handle.unpark();

        // Join the thread if we haven't already
        if let Some(handle) = self.thread.take() {
            let _ = handle.join();
        }
    }
}

impl std::fmt::Debug for CoreHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CoreHandle")
            .field("core_id", &self.core_id)
            .field("numa_node", &self.numa_node)
            .field("is_running", &self.is_running())
            .field("events_processed", &self.events_processed())
            .field("outputs_dropped", &self.outputs_dropped())
            .field("inbox_len", &self.inbox_len())
            .field("outbox_len", &self.outbox_len())
            .field("available_credits", &self.available_credits())
            .field("is_backpressured", &self.is_backpressured())
            .finish_non_exhaustive()
    }
}

/// Context passed to the core thread.
pub(super) struct CoreThreadContext {
    pub(super) core_id: usize,
    cpu_affinity: Option<usize>,
    reactor_config: ReactorConfig,
    numa_aware: bool,
    numa_node: usize,
    #[cfg(target_os = "linux")]
    numa_topology: NumaTopology,
    pub(super) inbox: Arc<SpscQueue<CoreMessage>>,
    pub(super) outbox: Arc<SpscQueue<TaggedOutput>>,
    pub(super) credit_gate: Arc<CreditGate>,
    pub(super) shutdown: Arc<AtomicBool>,
    pub(super) events_processed: Arc<AtomicU64>,
    pub(super) outputs_dropped: Arc<AtomicU64>,
    pub(super) is_running: Arc<AtomicBool>,
    pub(super) has_new_data: Arc<AtomicBool>,
    pub(super) output_notify: Arc<Notify>,
}

/// Initializes the core thread: sets CPU affinity, NUMA allocator, `io_uring`, and creates the reactor.
fn init_core_thread(
    ctx: &CoreThreadContext,
    operators: Vec<Box<dyn Operator>>,
) -> Result<Reactor, TpcError> {
    // Set CPU affinity if requested
    if let Some(cpu_id) = ctx.cpu_affinity {
        set_cpu_affinity(ctx.core_id, cpu_id)?;
    }

    // Bind memory allocations to local NUMA node (Linux only, after CPU affinity)
    #[cfg(target_os = "linux")]
    if ctx.numa_aware {
        if let Err(e) = ctx.numa_topology.bind_local_memory() {
            tracing::warn!(core_id = ctx.core_id, ?e, "NUMA memory bind failed");
        }
    }

    // Log NUMA information if NUMA-aware mode is enabled
    if ctx.numa_aware {
        tracing::info!(
            "Core {} starting on NUMA node {}",
            ctx.core_id,
            ctx.numa_node
        );
    }

    // Create the reactor with configured settings
    let mut reactor_config = ctx.reactor_config.clone();
    reactor_config.cpu_affinity = ctx.cpu_affinity;

    let mut reactor = Reactor::new(reactor_config).map_err(|e| TpcError::ReactorError {
        core_id: ctx.core_id,
        source: e,
    })?;

    // Add operators
    for op in operators {
        reactor.add_operator(op);
    }

    Ok(reactor)
}

/// Main function for the core thread.
#[allow(clippy::too_many_lines, clippy::needless_pass_by_value)]
fn core_thread_main(
    ctx: &CoreThreadContext,
    operators: Vec<Box<dyn Operator>>,
    mut storage_io: Option<Box<dyn StorageIo>>,
    io_completion_outbox: Arc<SpscQueue<IoCompletion>>,
) -> Result<(), TpcError> {
    let mut reactor = init_core_thread(ctx, operators)?;

    // Signal that we're running
    ctx.is_running.store(true, Ordering::Release);

    // Reusable buffer for reactor outputs (avoids per-poll Vec allocation)
    let mut poll_buffer: Vec<Output> = Vec::with_capacity(256);

    // Reusable buffer for storage I/O completions (zero-alloc across iterations)
    let mut io_completions: Vec<IoCompletion> = Vec::with_capacity(64);

    // Pre-allocated checkpoint Box reused across checkpoint cycles.
    // Taken when a CheckpointRequest arrives, replenished after the inbox drain.
    let mut checkpoint_slot: Option<Box<CheckpointCompleteData>> =
        Some(Box::new(CheckpointCompleteData {
            checkpoint_id: 0,
            operator_states: Vec::new(),
        }));
    // Reusable buffer for operator states (avoids Vec alloc per checkpoint)
    let mut checkpoint_states_buf: Vec<OperatorState> = Vec::new();

    // Track the last source_idx seen for tagging reactor outputs.
    // When events from multiple sources are interleaved, reactor outputs
    // are attributed to source 0 (the coordinator resolves this).
    let mut last_source_idx: usize = 0;

    // Rate-limit submit errors to avoid allocation storms on the hot path.
    let mut submit_error_count: u64 = 0;

    // Tiered idle strategy: tracks consecutive iterations with no work.
    // 0..64   → spin_loop() (PAUSE/YIELD hint, ~ns wake)
    // 64..128 → thread::yield_now() (OS scheduler yield, ~μs wake)
    // 128+    → thread::park_timeout(1ms) (sleep, woken by unpark())
    let mut idle_spins: u32 = 0;

    // Wrap the main loop in catch_unwind so an operator panic does not
    // silently kill this core thread.  On panic we set is_running = false
    // (the coordinator checks this) and propagate the error.
    let panic_result = catch_unwind(AssertUnwindSafe(|| -> Result<(), TpcError> {
        // Main loop
        loop {
            // Check for shutdown
            if ctx.shutdown.load(Ordering::Acquire) {
                break;
            }

            // Hot path guard for inbox processing
            let _guard = HotPathGuard::enter("CoreThread::process_inbox");

            // Task budget tracking for batch processing
            let batch_budget = TaskBudget::ring0_batch();

            // Drain inbox and track messages processed for credit release
            let mut had_work = false;
            let mut messages_processed = 0usize;

            while let Some(message) = ctx.inbox.pop() {
                match message {
                    CoreMessage::Event { source_idx, event } => {
                        last_source_idx = source_idx;
                        if let Err(e) = reactor.submit(event) {
                            submit_error_count += 1;
                            if submit_error_count.is_power_of_two() {
                                tracing::error!(
                                    "Core {}: Failed to submit event (n={}): {e}",
                                    ctx.core_id,
                                    submit_error_count,
                                );
                            }
                        }
                        messages_processed += 1;
                        had_work = true;
                    }
                    CoreMessage::Watermark(timestamp) => {
                        // Advance the reactor's watermark so downstream operators
                        // see the updated event-time progress on the next poll().
                        reactor.advance_watermark(timestamp);
                        messages_processed += 1;
                        had_work = true;
                    }
                    CoreMessage::CheckpointRequest(checkpoint_id) => {
                        // Snapshot operator states into the pre-allocated Box.
                        // The Box is taken and replenished after the inbox drain loop.
                        reactor.trigger_checkpoint_into(&mut checkpoint_states_buf);
                        let mut data = checkpoint_slot.take().unwrap_or_else(|| {
                            Box::new(CheckpointCompleteData {
                                checkpoint_id: 0,
                                operator_states: Vec::new(),
                            })
                        });
                        data.checkpoint_id = checkpoint_id;
                        std::mem::swap(&mut data.operator_states, &mut checkpoint_states_buf);
                        // Checkpoint completion must not be dropped — the coordinator
                        // would stall waiting for a response that never arrives.
                        // Spin-wait: checkpoints are infrequent (seconds apart).
                        let mut cp_out = TaggedOutput {
                            source_idx: 0,
                            output: Output::CheckpointComplete(data),
                        };
                        loop {
                            match ctx.outbox.push(cp_out) {
                                Ok(()) => break,
                                Err(returned) => {
                                    if ctx.shutdown.load(Ordering::Acquire) {
                                        break;
                                    }
                                    cp_out = returned;
                                    std::hint::spin_loop();
                                }
                            }
                        }
                        messages_processed += 1;
                        had_work = true;
                    }
                    CoreMessage::Barrier {
                        source_idx,
                        barrier,
                    } => {
                        // Fully drain the reactor before forwarding the barrier.
                        // A single poll_into may only process batch_size events;
                        // we must loop until the queue is empty so no pre-barrier
                        // event slips past the barrier.
                        loop {
                            poll_buffer.clear();
                            reactor.poll_into(&mut poll_buffer);
                            if poll_buffer.is_empty() {
                                break;
                            }
                            for output in poll_buffer.drain(..) {
                                if ctx
                                    .outbox
                                    .push(TaggedOutput { source_idx, output })
                                    .is_err()
                                {
                                    ctx.outputs_dropped.fetch_add(1, Ordering::Relaxed);
                                }
                            }
                        }
                        // Barrier must not be dropped — a missing barrier breaks
                        // checkpoint alignment. Spin-wait until outbox has space.
                        let mut barrier_out = TaggedOutput {
                            source_idx,
                            output: Output::Barrier(barrier),
                        };
                        loop {
                            match ctx.outbox.push(barrier_out) {
                                Ok(()) => break,
                                Err(returned) => {
                                    if ctx.shutdown.load(Ordering::Acquire) {
                                        break;
                                    }
                                    barrier_out = returned;
                                    std::hint::spin_loop();
                                }
                            }
                        }
                        messages_processed += 1;
                        had_work = true;
                    }
                    CoreMessage::Shutdown => {
                        // Release credits for any messages we've processed so far
                        if messages_processed > 0 {
                            ctx.credit_gate.release(messages_processed);
                            messages_processed = 0;
                        }
                        break;
                    }
                }

                // Check if batch budget is almost exceeded and break to process reactor
                // This ensures Ring 0 latency guarantees by limiting batch processing time
                if batch_budget.almost_exceeded() {
                    break;
                }
            }

            // Release credits for processed messages
            // This signals to senders that we have capacity for more
            if messages_processed > 0 {
                ctx.credit_gate.release(messages_processed);
            }

            // Poll storage I/O completions (non-blocking, zero-alloc).
            // Push completions to the SPSC outbox for Ring 2 (checkpoint coordinator)
            // to drain and feed to PerCoreWalManager::check_all_completions().
            if let Some(ref mut sio) = storage_io {
                io_completions.clear();
                sio.poll_completions(&mut io_completions);
                for completion in &io_completions {
                    // Spin-wait on push failure. A dropped completion
                    // permanently stalls the WAL writer (pending_sync
                    // never clears, blocking all future syncs).
                    let mut c = *completion;
                    loop {
                        match io_completion_outbox.push(c) {
                            Ok(()) => break,
                            Err(returned) => {
                                if ctx.shutdown.load(Ordering::Acquire) {
                                    break;
                                }
                                c = returned;
                                std::hint::spin_loop();
                            }
                        }
                    }
                }
                if !io_completions.is_empty() {
                    had_work = true;
                }
            }

            // Replenish checkpoint slot if it was consumed (infrequent — once per checkpoint).
            // checkpoint_states_buf was drained by the swap above; move its capacity into the
            // new Box so both buffers retain their allocations across cycles.
            if checkpoint_slot.is_none() {
                checkpoint_states_buf.clear();
                let mut data = Box::new(CheckpointCompleteData {
                    checkpoint_id: 0,
                    operator_states: Vec::new(),
                });
                std::mem::swap(&mut data.operator_states, &mut checkpoint_states_buf);
                checkpoint_slot = Some(data);
            }

            // Process events in reactor (reuses poll_buffer capacity across iterations)
            poll_buffer.clear();
            reactor.poll_into(&mut poll_buffer);
            ctx.events_processed
                .fetch_add(poll_buffer.len() as u64, Ordering::Relaxed);

            // Push outputs to outbox tagged with the last source index
            for output in poll_buffer.drain(..) {
                if ctx
                    .outbox
                    .push(TaggedOutput {
                        source_idx: last_source_idx,
                        output,
                    })
                    .is_err()
                {
                    ctx.outputs_dropped.fetch_add(1, Ordering::Relaxed);
                }
                had_work = true;
            }

            // Tiered idle: progressively back off to avoid burning CPU
            // when no work is available. Source I/O threads call unpark()
            // after pushing to the inbox, giving sub-ms wake latency.
            if had_work {
                // Signal the coordinator that outputs are available.
                // Lock-free: atomic swap is ~5ns. Only the false→true
                // transition calls notify_one(), so at most ONE core per
                // coordinator cycle hits the Notify mutex — no N-way contention.
                if !ctx.has_new_data.swap(true, Ordering::Release) {
                    ctx.output_notify.notify_one();
                }
                idle_spins = 0;
            } else {
                idle_spins = idle_spins.saturating_add(1);
                if idle_spins < 64 {
                    std::hint::spin_loop();
                } else if idle_spins < 128 {
                    thread::yield_now();
                } else {
                    thread::park_timeout(std::time::Duration::from_millis(1));
                }
            }
        }

        // Drain any remaining events before shutdown
        poll_buffer.clear();
        reactor.poll_into(&mut poll_buffer);
        for output in poll_buffer.drain(..) {
            let _ = ctx.outbox.push(TaggedOutput {
                source_idx: last_source_idx,
                output,
            });
        }

        Ok(())
    })); // end catch_unwind

    // Always clear is_running so coordinator detects core death.
    ctx.is_running.store(false, Ordering::Release);
    // Signal coordinator — it may be parked waiting for output.
    ctx.output_notify.notify_one();

    match panic_result {
        Ok(inner) => inner,
        Err(payload) => {
            let message = if let Some(s) = payload.downcast_ref::<&str>() {
                (*s).to_string()
            } else if let Some(s) = payload.downcast_ref::<String>() {
                s.clone()
            } else {
                "unknown panic".to_string()
            };
            tracing::error!("Core {}: operator panic caught: {message}", ctx.core_id,);
            Err(TpcError::OperatorPanic {
                core_id: ctx.core_id,
                message,
            })
        }
    }
}

/// Sets CPU affinity for the current thread.
fn set_cpu_affinity(core_id: usize, cpu_id: usize) -> Result<(), TpcError> {
    #[cfg(target_os = "linux")]
    {
        use libc::{cpu_set_t, sched_setaffinity, CPU_SET, CPU_ZERO};
        use std::mem;

        // SAFETY: We're calling libc functions with valid parameters.
        // The cpu_set_t is properly initialized with CPU_ZERO.
        // The process ID 0 refers to the current thread.
        #[allow(unsafe_code)]
        unsafe {
            let mut set: cpu_set_t = mem::zeroed();
            CPU_ZERO(&mut set);
            CPU_SET(cpu_id, &mut set);

            let result = sched_setaffinity(0, mem::size_of::<cpu_set_t>(), &raw const set);
            if result != 0 {
                return Err(TpcError::AffinityFailed {
                    core_id,
                    message: format!(
                        "sched_setaffinity failed: {}",
                        std::io::Error::last_os_error()
                    ),
                });
            }
        }
    }

    #[cfg(target_os = "windows")]
    {
        use windows_sys::Win32::System::Threading::{GetCurrentThread, SetThreadAffinityMask};

        if cpu_id >= usize::BITS as usize {
            return Err(TpcError::AffinityFailed {
                core_id,
                message: format!(
                    "cpu_id {cpu_id} >= {} — Windows processor groups not yet supported",
                    usize::BITS
                ),
            });
        }
        // SAFETY: We're calling Windows API functions with valid parameters.
        // GetCurrentThread returns a pseudo-handle that doesn't need to be closed.
        // The mask is a valid CPU mask for the specified core.
        #[allow(unsafe_code)]
        unsafe {
            let mask: usize = 1 << cpu_id;
            let result = SetThreadAffinityMask(GetCurrentThread(), mask);
            if result == 0 {
                return Err(TpcError::AffinityFailed {
                    core_id,
                    message: format!(
                        "SetThreadAffinityMask failed: {}",
                        std::io::Error::last_os_error()
                    ),
                });
            }
        }
    }

    #[cfg(not(any(target_os = "linux", target_os = "windows")))]
    {
        let _ = (core_id, cpu_id);
        // No-op on other platforms
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::operator::{OperatorState, OutputVec, Timer};
    use arrow_array::{Int64Array, RecordBatch};
    use std::sync::Arc;
    use std::time::Duration;

    // Simple passthrough operator for testing
    struct PassthroughOperator;

    impl Operator for PassthroughOperator {
        fn process(
            &mut self,
            event: &Event,
            _ctx: &mut crate::operator::OperatorContext,
        ) -> OutputVec {
            let mut output = OutputVec::new();
            output.push(Output::Event(event.clone()));
            output
        }

        fn on_timer(
            &mut self,
            _timer: Timer,
            _ctx: &mut crate::operator::OperatorContext,
        ) -> OutputVec {
            OutputVec::new()
        }

        fn checkpoint(&self) -> OperatorState {
            OperatorState {
                operator_id: "passthrough".to_string(),
                data: vec![],
            }
        }

        fn restore(&mut self, _state: OperatorState) -> Result<(), crate::operator::OperatorError> {
            Ok(())
        }
    }

    fn make_event(value: i64) -> Event {
        let array = Arc::new(Int64Array::from(vec![value]));
        let batch = RecordBatch::try_from_iter(vec![("value", array as _)]).unwrap();
        Event::new(value, batch)
    }

    #[test]
    fn test_core_handle_spawn() {
        let config = CoreConfig {
            core_id: 0,
            cpu_affinity: None, // Don't pin in tests
            inbox_capacity: 1024,
            outbox_capacity: 1024,
            reactor_config: ReactorConfig::default(),
            backpressure: super::BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        };

        let handle = CoreHandle::spawn(config).unwrap();
        assert!(handle.is_running());
        assert_eq!(handle.core_id(), 0);

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_core_handle_with_operator() {
        let config = CoreConfig {
            core_id: 0,
            cpu_affinity: None,
            inbox_capacity: 1024,
            outbox_capacity: 1024,
            reactor_config: ReactorConfig::default(),
            backpressure: super::BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        };

        let handle =
            CoreHandle::spawn_with_operators(config, vec![Box::new(PassthroughOperator)]).unwrap();

        // Send an event
        let event = make_event(42);
        handle.send_event(0, event).unwrap();

        // Wait a bit for processing
        thread::sleep(Duration::from_millis(50));

        // Poll for outputs
        let outputs = handle.poll_outputs(10);
        assert!(!outputs.is_empty());
        // Verify source_idx is preserved
        assert_eq!(outputs[0].source_idx, 0);

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_core_handle_multiple_events() {
        let config = CoreConfig {
            core_id: 1,
            cpu_affinity: None,
            inbox_capacity: 1024,
            outbox_capacity: 1024,
            reactor_config: ReactorConfig::default(),
            backpressure: super::BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        };

        let handle =
            CoreHandle::spawn_with_operators(config, vec![Box::new(PassthroughOperator)]).unwrap();

        // Send multiple events from source 0
        for i in 0..100 {
            handle.send_event(0, make_event(i)).unwrap();
        }

        // Wait for processing
        thread::sleep(Duration::from_millis(100));

        // Poll all outputs
        let mut total_outputs = 0;
        loop {
            let outputs = handle.poll_outputs(1000);
            if outputs.is_empty() {
                break;
            }
            total_outputs += outputs.len();
        }

        // Should have at least some outputs (may have watermarks too)
        assert!(total_outputs >= 100);

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_core_handle_shutdown() {
        let config = CoreConfig::default();
        let handle = CoreHandle::spawn(config).unwrap();

        assert!(handle.is_running());

        handle.shutdown();

        // Wait for shutdown
        thread::sleep(Duration::from_millis(100));

        // Thread should have stopped
        assert!(!handle.is_running());
    }

    #[test]
    fn test_core_handle_debug() {
        let config = CoreConfig {
            core_id: 42,
            ..Default::default()
        };
        let handle = CoreHandle::spawn(config).unwrap();

        let debug_str = format!("{handle:?}");
        assert!(debug_str.contains("CoreHandle"));
        assert!(debug_str.contains("42"));

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_core_config_default() {
        let config = CoreConfig::default();
        assert_eq!(config.core_id, 0);
        assert!(config.cpu_affinity.is_none());
        assert_eq!(config.inbox_capacity, 8192);
        assert_eq!(config.outbox_capacity, 8192);
        assert!(!config.numa_aware);
    }

    #[test]
    fn test_core_handle_numa_node() {
        let config = CoreConfig {
            core_id: 0,
            cpu_affinity: None,
            numa_aware: true,
            ..Default::default()
        };

        let handle = CoreHandle::spawn(config).unwrap();
        // On any system, numa_node should be a valid value (0 on non-NUMA systems)
        assert!(handle.numa_node() < 64);

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_poll_outputs_into() {
        let config = CoreConfig {
            core_id: 0,
            cpu_affinity: None,
            inbox_capacity: 1024,
            outbox_capacity: 1024,
            reactor_config: ReactorConfig::default(),
            backpressure: super::BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        };

        let handle =
            CoreHandle::spawn_with_operators(config, vec![Box::new(PassthroughOperator)]).unwrap();

        // Send events
        for i in 0..10 {
            handle.send_event(0, make_event(i)).unwrap();
        }

        // Wait for processing
        thread::sleep(Duration::from_millis(100));

        // Poll into pre-allocated buffer
        let mut buffer = Vec::with_capacity(100);
        let count = handle.poll_outputs_into(&mut buffer, 100);

        assert!(count > 0);
        assert_eq!(buffer.len(), count);

        // Reuse buffer - no new allocation
        let cap_before = buffer.capacity();
        buffer.clear();
        let _ = handle.poll_outputs_into(&mut buffer, 100);
        assert_eq!(buffer.capacity(), cap_before); // Capacity unchanged

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_poll_each() {
        let config = CoreConfig {
            core_id: 0,
            cpu_affinity: None,
            inbox_capacity: 1024,
            outbox_capacity: 1024,
            reactor_config: ReactorConfig::default(),
            backpressure: super::BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        };

        let handle =
            CoreHandle::spawn_with_operators(config, vec![Box::new(PassthroughOperator)]).unwrap();

        // Send events
        for i in 0..10 {
            handle.send_event(0, make_event(i)).unwrap();
        }

        // Wait for processing
        thread::sleep(Duration::from_millis(100));

        // Poll with callback
        let mut event_count = 0;
        let count = handle.poll_each(100, |tagged| {
            if matches!(tagged.output, Output::Event(_)) {
                event_count += 1;
            }
            true
        });

        assert!(count > 0);
        assert!(event_count > 0);

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_poll_each_early_stop() {
        let config = CoreConfig {
            core_id: 0,
            cpu_affinity: None,
            inbox_capacity: 1024,
            outbox_capacity: 1024,
            reactor_config: ReactorConfig::default(),
            backpressure: super::BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        };

        let handle =
            CoreHandle::spawn_with_operators(config, vec![Box::new(PassthroughOperator)]).unwrap();

        // Send events
        for i in 0..20 {
            handle.send_event(0, make_event(i)).unwrap();
        }

        // Wait for processing
        thread::sleep(Duration::from_millis(100));

        // Poll with early stop after 5 items
        let mut processed = 0;
        let count = handle.poll_each(100, |_| {
            processed += 1;
            processed < 5 // Stop after 5
        });

        assert_eq!(count, 5);
        assert_eq!(processed, 5);

        // There should be more items remaining
        let remaining = handle.outbox_len();
        assert!(remaining > 0);

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_watermark_propagation() {
        let config = CoreConfig {
            core_id: 0,
            cpu_affinity: None,
            inbox_capacity: 1024,
            outbox_capacity: 1024,
            reactor_config: ReactorConfig::default(),
            backpressure: super::BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        };

        let handle =
            CoreHandle::spawn_with_operators(config, vec![Box::new(PassthroughOperator)]).unwrap();

        // Send a watermark advancement
        handle.send(CoreMessage::Watermark(5000)).unwrap();

        // Wait for processing
        thread::sleep(Duration::from_millis(50));

        // Poll outputs - should contain a watermark
        let outputs = handle.poll_outputs(100);
        let has_watermark = outputs
            .iter()
            .any(|o| matches!(o.output, Output::Watermark(_)));
        assert!(
            has_watermark,
            "Expected watermark output after Watermark message"
        );

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_checkpoint_triggering() {
        let config = CoreConfig {
            core_id: 0,
            cpu_affinity: None,
            inbox_capacity: 1024,
            outbox_capacity: 1024,
            reactor_config: ReactorConfig::default(),
            backpressure: super::BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        };

        let handle =
            CoreHandle::spawn_with_operators(config, vec![Box::new(PassthroughOperator)]).unwrap();

        // Send a checkpoint request
        handle.send(CoreMessage::CheckpointRequest(42)).unwrap();

        // Wait for processing
        thread::sleep(Duration::from_millis(50));

        // Poll outputs - should contain a CheckpointComplete
        let outputs = handle.poll_outputs(100);
        let checkpoint = outputs
            .iter()
            .find(|o| matches!(o.output, Output::CheckpointComplete(_)));
        assert!(checkpoint.is_some(), "Expected CheckpointComplete output");

        if let Some(TaggedOutput {
            output: Output::CheckpointComplete(data),
            ..
        }) = checkpoint
        {
            assert_eq!(data.checkpoint_id, 42);
            // One operator (PassthroughOperator)
            assert_eq!(data.operator_states.len(), 1);
            assert_eq!(data.operator_states[0].operator_id, "passthrough");
        }

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_outputs_dropped_counter() {
        let config = CoreConfig {
            core_id: 0,
            cpu_affinity: None,
            inbox_capacity: 1024,
            outbox_capacity: 4, // Very small outbox to force drops
            reactor_config: ReactorConfig::default(),
            backpressure: super::BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        };

        let handle =
            CoreHandle::spawn_with_operators(config, vec![Box::new(PassthroughOperator)]).unwrap();

        // Send many events without polling - outbox should fill up
        for i in 0..100 {
            let _ = handle.send_event(0, make_event(i));
        }

        // Wait for processing to fill and overflow the outbox
        thread::sleep(Duration::from_millis(200));

        // Some outputs should have been dropped
        let dropped = handle.outputs_dropped();
        assert!(
            dropped > 0,
            "Expected some outputs to be dropped with outbox_capacity=4"
        );

        handle.shutdown_and_join().unwrap();
    }

    #[test]
    fn test_barrier_handling() {
        let config = CoreConfig {
            core_id: 0,
            cpu_affinity: None,
            inbox_capacity: 1024,
            outbox_capacity: 1024,
            reactor_config: ReactorConfig::default(),
            backpressure: super::BackpressureConfig::default(),
            numa_aware: false,
            enable_storage_io: false,
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            io_uring_config: None,
        };

        let handle =
            CoreHandle::spawn_with_operators(config, vec![Box::new(PassthroughOperator)]).unwrap();

        // Send some events then a barrier
        handle.send_event(0, make_event(1)).unwrap();
        handle.send_event(0, make_event(2)).unwrap();
        handle
            .send(CoreMessage::Barrier {
                source_idx: 0,
                barrier: CheckpointBarrier::new(99, 1),
            })
            .unwrap();

        // Wait for processing
        thread::sleep(Duration::from_millis(100));

        // Poll outputs - should contain events then a barrier
        let outputs = handle.poll_outputs(100);
        let has_barrier = outputs
            .iter()
            .any(|o| matches!(o.output, Output::Barrier(_)));
        assert!(has_barrier, "Expected Barrier output");

        // Barrier should have correct source_idx
        let barrier_tagged = outputs
            .iter()
            .find(|o| matches!(o.output, Output::Barrier(_)))
            .unwrap();
        assert_eq!(barrier_tagged.source_idx, 0);

        handle.shutdown_and_join().unwrap();
    }
}