chore: nitpicks after multithreading review

2026-09-10 07:16:23 -04:00 · 2026-09-05 15:16:50 +02:00 · 2026-09-05 15:16:50 +02:00 · ade829872b
parent 0392eeb70d
commit ade829872b
12 changed files with 495 additions and 163 deletions
--- a/ARCHITECTURE.md
+++ b/ARCHITECTURE.md
@ -428,7 +428,8 @@ Source (stdin bytes or child process stdout)
  └── parallel_bufread()  (all inputs)
        ├─ Thread 1: I/O reader — reads 256 KB chunks, splits at line boundaries,
        │             assigns monotonic sequence numbers, sends to MPMC channel
-        ├─ Thread N: workers — receive chunks, validate UTF-8,
+        ├─ Thread 1: bounded dispatcher — uses in-flight tokens to limit queued/running jobs
+        ├─ Pool workers: receive chunk jobs, validate UTF-8,
        │             create DefaultSkimItem::new(line, ansi, trans_fields, matching_fields, delimiter)
        │               .hidden_fields(hidden_fields, delimiter)
        │             (handles ANSI stripping, --nth / --with-nth / --hide-nth inline),
@ -491,7 +492,7 @@ Engines are composable through the factory pattern. Starting from `Matcher::crea
 options
  │
  ├── if regex mode:
-  │     RegexEngineFactory
+  │     RegexEngineFactory (configured with the same RankBuilder / --tiebreak criteria)
  │       └─ if normalize: NormalizedEngineFactory(RegexEngineFactory)
  │
  └── else (fuzzy/exact mode):
@ -581,12 +582,12 @@ Matcher::run(query, item_pool, thread_pool, …)
        │          └─ merge_worker_results(worker_results, no_sort, …)
        │               ├─ concatenate k sorted runs
        │               ├─ sort() (stable; driftsort detects k runs → O(n log k))
-        │               └─ write into SpinLock<Option<ProcessedItems>>
+        │               └─ validate query generation and write into Mutex<Option<ProcessedItems>>
        │
        └─ stopped.store(true)
 ```

-Interruption is cooperative: each chunk checks `interrupt.load(Relaxed)` before processing. `MatcherControl::kill()` sets `interrupt = true`; `MatcherControl::drop()` also calls `kill()`.
+Interruption is cooperative: each chunk checks `interrupt.load(Relaxed)` before processing. `MatcherControl::kill()` sets `interrupt = true`; `MatcherControl::drop()` also calls `kill()`. Each forced restart increments a shared query generation. Published `ProcessedItems` carry that generation; both publication and consumption reject stale generations, so a cancelled matcher cannot replace newer results.

 ### Ranking & Sorting

@ -828,13 +829,14 @@ The `StatusInfo` struct rendered inside the input line shows:
 `ItemList` (`src/tui/item_list.rs`) maintains:

 - `items: Vec<MatchedItem>` — the currently displayed matched items
- `processed_items: Arc<SpinLock<Option<ProcessedItems>>>` — shared with matcher
+- `processed_items: Arc<Mutex<Option<ProcessedItems>>>` — shared with matcher
+- `matcher_generation: Arc<AtomicUsize>` — identifies the active query generation
 - `selection: Vec<usize>` — indices of multi-selected items
 - `current: usize` — focused item index (0 = bottom in default layout)
 - `offset: usize` — scroll offset (number of items scrolled)
 - `manual_hscroll: i16` — user-driven horizontal scroll

-On each render, `ItemList::render()` checks `processed_items` and swaps them in atomically via the `SpinLock`. Depending on `MergeStrategy`:
+On each render, `ItemList::render()` checks `processed_items`, rejects results from an old query generation, and swaps current results in through the mutex. Depending on `MergeStrategy`:

 - `Replace`: replaces `items` entirely.
 - `SortedMerge`: performs an O(n+m) merge preserving order.
@ -868,7 +870,7 @@ Pre-selection is applied when items first appear: `DefaultSkimSelector::should_s

 `Preview` (`src/tui/preview.rs`) renders a side/top/bottom pane showing expanded information about the focused item. Its stored content is one of three variants:

-**Plain text mode** (no `pty`): spawns `sh -c <cmd>` on Unix or `cmd /c <cmd>` on Windows. On Windows, `Command::raw_arg` is used so `cmd.exe` receives shell metacharacters exactly as written. The child captures stdout (capped at `PREVIEW_MAX_BYTES`), parses it with `ansi_to_tui::IntoText`, stores as `PreviewContent::Text`, and sends `Event::PreviewReady`.
+**Plain text mode** (no `pty`): spawns `sh -c <cmd>` on Unix or `cmd /c <cmd>` on Windows. On Windows, `Command::raw_arg` is used so `cmd.exe` receives shell metacharacters exactly as written. The worker drains stdout and stderr concurrently, but retains at most `PREVIEW_MAX_BYTES` from each stream. Cancellation terminates the child process group (the process tree on Windows), so selection changes do not leave old preview commands running. Successful stdout or failed stderr is parsed with `ansi_to_tui::IntoText`, stored as `PreviewContent::Text`, and followed by `Event::PreviewReady`.

 **PTY mode** (`--preview-window pty`): creates a real pseudo-terminal pair via `portable_pty`. The child process sees a properly sized terminal (via `ROWS`/`COLUMNS` env and PTY dimensions). Output is parsed by a `vt100::Parser` with a scrollback buffer, stored as `PreviewContent::Terminal(Arc<RwLock<vt100::Parser>>)`. This enables interactive preview programs (e.g. `bat`, `delta`).

@ -893,12 +895,14 @@ else if pty mode:
          → Event::PreviewReady when EOF

 else:
-  sh -c <cmd>
-  thread: wait for output → content.write() = PreviewContent::Text(…)
+  start shell in a dedicated process group with piped stdout + stderr
+  thread: drain both streams with bounded retention; poll child status
+          → cancellation kills the process group
+          → content.write() = PreviewContent::Text(…)
          → Event::PreviewReady
 ```

-Scroll state: `scroll_y`, `scroll_x` (in lines/columns). `page_up/down`, `scroll_up/down/left/right` modify these. `PreviewPosition` supports fixed, percentage, and negative offsets. When `PreviewReady` fires, an optional offset expression (from `--preview-window +expr`) is evaluated to auto-scroll to the matched line.
+Scroll state: `scroll_y`, `scroll_x` (in lines/columns) and `total_lines` use `usize`; conversion to ratatui's `u16` coordinates saturates at render time. `page_up/down`, `scroll_up/down/left/right` modify these. `PreviewPosition` supports fixed, percentage, and negative offsets. When `PreviewReady` fires, an optional offset expression (from `--preview-window +expr`) is evaluated to auto-scroll to the matched line.

 ### Header Widget

@ -1275,21 +1279,23 @@ Main thread (Tokio runtime)
  │
  └─ Tokio task: Tui event pump (crossterm EventStream + tick timer)

-ThreadPool (N = num_cpus OS threads, persistent)
-  ├─ Matcher coordinator (1 slot per match run)
-  └─ Worker threads (N-1 slots per match run)
+Matcher ThreadPool (persistent)
+  └─ Worker threads process atomic match chunks; one separate coordinator thread waits for completion

-Reader threads (OS threads, per-invocation):
-  ├─ collect_items thread: blocks on SkimItemReceiver (recv_timeout 5ms), calls ItemPool::append
+Reader ThreadPool (persistent)
+  └─ Short chunk jobs parse items; an in-flight token limit bounds the work queue
+
+Reader threads (OS threads, per invocation):
+  ├─ collect_items thread: blocks on SkimItemReceiver (recv_timeout 1ms), calls ItemPool::append
  ├─ I/O reader thread: reads large byte chunks, splits lines, assigns sequence numbers
-  ├─ Worker threads (N): parse lines, create DefaultSkimItem (ANSI strip + field transforms inline)
+  ├─ Bounded dispatcher thread: submits chunk jobs only while an in-flight token is available
  ├─ Reorder thread: sequence-ordered output; drops tx_pipeline_done on EOF
-  └─ Killer thread (command inputs only): waits for rx_interrupt or rx_pipeline_done;
-       kills child process when either fires
+  └─ Killer thread: waits for rx_interrupt or rx_pipeline_done; kills a command child if present

-Preview thread (OS thread, per preview spawn):
-  └─ reads PTY/child stdout or decodes image path → PreviewContent Arc<RwLock>
-      → sends Event::PreviewReady
+Preview threads (OS threads, per preview spawn):
+  ├─ PTY reader, image decoder, or plain-child monitor
+  └─ Plain mode also has bounded stdout and stderr drain threads
+      → writes PreviewContent Arc<RwLock> and sends Event::PreviewReady

 IPC handler task (Tokio, per connection):
  └─ reads RON actions → sends Event::Action to TUI channel
@ -1300,9 +1306,9 @@ Popup stdin relay thread (OS thread, only in --popup/--tmux mode):

 **Synchronization primitives used:**

- `Arc<SpinLock<Option<ProcessedItems>>>` — matcher-to-ItemList result handoff
+- `Arc<Mutex<Option<ProcessedItems>>>` — matcher-to-ItemList result handoff without CPU-spinning under merge contention
 - `Arc<AtomicBool>` — `needs_render` (matcher → event loop), `stopped` / `interrupt` (MatcherControl)
- `Arc<AtomicUsize>` — `processed` / `matched` counters, reader `components_to_stop`
+- `Arc<AtomicUsize>` — `processed` / `matched` counters, matcher query generation, reader `components_to_stop`
 - `Arc<tokio::sync::Notify>` — `items_available` (ItemPool → Skim::tick wakeup)
 - `Arc<std::sync::RwLock<PreviewContent>>` — preview thread → Preview widget
 - `kanal::Sender/Receiver<Vec<Arc<dyn SkimItem>>>` — item batches through pipeline
@ -1324,28 +1330,28 @@ The global allocator is `mimalloc` (v3), chosen for its low-latency multi-thread
 | `Skim::init_tui_with` | `src/skim.rs:307` | Install a caller-provided TUI backend |
 | `Skim::enter` | `src/skim.rs:394` | Enter terminal, resolve image picker, start listener/event pump |
 | `Skim::should_enter` | `src/skim.rs:438` | Filter/select-1/exit-0/sync gate |
-| `Skim::output` | `src/skim.rs:555` | Collect & return SkimOutput |
-| `Skim::tick` | `src/skim.rs:636` | Single async event loop iteration |
-| `App::from_options` | `src/tui/app.rs:289` | Build all widgets from options |
-| `App::run_preview` | `src/tui/app.rs:503` | Expand cmd, debounce, call Preview::spawn |
-| `App::handle_event` | `src/tui/app.rs:628` | Dispatch all Event variants |
-| `App::handle_action` | `src/tui/app.rs:833` | Apply action follow-up bindings |
-| `App::dispatch_conditional` | `src/tui/app.rs:852` | Dispatch the selected conditional subaction chain without follow-up bindings |
-| `App::dispatch_action` | `src/tui/app.rs:875` | Dispatch one Action variant without follow-up bindings |
+| `Skim::output` | `src/skim.rs:563` | Collect & return SkimOutput |
+| `Skim::tick` | `src/skim.rs:644` | Single async event loop iteration |
+| `App::from_options` | `src/tui/app.rs:291` | Build all widgets from options |
+| `App::run_preview` | `src/tui/app.rs:507` | Expand cmd, debounce, call Preview::spawn |
+| `App::handle_event` | `src/tui/app.rs:632` | Dispatch all Event variants |
+| `App::handle_action` | `src/tui/app.rs:837` | Apply action follow-up bindings |
+| `App::dispatch_conditional` | `src/tui/app.rs:856` | Dispatch the selected conditional subaction chain without follow-up bindings |
+| `App::dispatch_action` | `src/tui/app.rs:879` | Dispatch one Action variant without follow-up bindings |
 | `Tui::run_execute` | `src/tui/backend.rs:363` | Suspend reader, run `execute` child with its own tty stdin, restart reader |
-| `App::restart_matcher` | `src/tui/app.rs:1352` | Kill old match pass, start new one |
-| `App::expand_cmd` | `src/tui/app.rs:1428` | Substitute `{}`, `{q}`, `{n}` etc. |
-| `App::handle_mouse` | `src/tui/app.rs:1496` | Handle mouse behavior and emit `double-click` |
-| `Widget::render (App)` | `src/tui/app.rs:151` | Root render; calls all sub-widgets |
-| `Matcher::run` | `src/matcher.rs:~260` | Parallel match dispatch |
-| `merge_worker_results` | `src/matcher.rs:28` | Merge k sorted runs → ProcessedItems |
+| `App::restart_matcher` | `src/tui/app.rs:1378` | Kill old match pass, start new one |
+| `App::expand_cmd` | `src/tui/app.rs:1465` | Substitute `{}`, `{q}`, `{n}` etc. |
+| `App::handle_mouse` | `src/tui/app.rs:1533` | Handle mouse behavior and emit `double-click` |
+| `Widget::render (App)` | `src/tui/app.rs:152` | Root render; calls all sub-widgets |
+| `Matcher::run` | `src/matcher.rs:294` | Parallel match dispatch |
+| `merge_worker_results` | `src/matcher.rs:36` | Merge k sorted runs → ProcessedItems |
 | `ItemPool::append` | `src/item.rs:469` | Add items, notify matcher |
 | `ItemPool::take` | `src/item.rs:502` | Take un-matched items for matcher |
 | `DefaultSkimItem::new` | `src/helper/item.rs:64` | ANSI strip, field transform, matching ranges (hidden ranges set later via `hidden_fields` builder) |
-| `SkimItemReader::parallel_bufread` | `src/helper/item_reader.rs:287` | Unified parallel pipeline (all inputs) |
-| `spawn_io_reader` | `src/helper/item_reader.rs:378` | I/O reader thread: chunk reads + line splitting |
-| `spawn_reorder_thread` | `src/helper/item_reader.rs:483` | Reorder thread: ordered output + pipeline-done signal |
-| `Preview::spawn` | `src/tui/preview.rs:319` | Start image, PTY, or plain preview worker |
+| `SkimItemReader::parallel_bufread` | `src/helper/item_reader.rs:284` | Unified parallel pipeline (all inputs) |
+| `spawn_io_reader` | `src/helper/item_reader.rs:380` | I/O reader thread: chunk reads + line splitting |
+| `spawn_reorder_thread` | `src/helper/item_reader.rs:485` | Reorder thread: ordered output + pipeline-done signal |
+| `Preview::spawn` | `src/tui/preview.rs:371` | Start image, PTY, or plain preview worker |
 | `Tui::new_with_height_and_backend` | `src/tui/backend.rs:81` | Terminal init + viewport sizing |
 | `Tui::enter` | `src/tui/backend.rs:134` | Enable raw mode + terminal setup |
 | `Tui::start` | `src/tui/backend.rs:235` | Spawn crossterm EventStream task (fresh cancellation token each call) |
--- a/Cargo.toml
+++ b/Cargo.toml
@ -88,7 +88,7 @@ kanal = "0.1.1"
 log = "0.4.31"
 memchr = "2.8.1"
 mimalloc = { version = "0.1.48", features = ["v3"] }
-nix = { version = "0.31.3", features = ["fs", "poll"] }
+nix = { version = "0.31.3", features = ["fs", "poll", "signal"] }
 portable-pty = "0.9.0"
 ratatui = "0.30.0"
 ratatui-image = { version = "11.0.4", features = ["crossterm"], default-features = false, optional = true }
--- a/src/helper/item_reader.rs
+++ b/src/helper/item_reader.rs
@ -265,16 +265,13 @@ impl SkimItemReader {
    /// 1. **I/O thread** (dedicated) — reads large byte chunks (~256 KB) from
    ///    `source`, splitting on line boundaries, and sends them tagged with
    ///    monotonic sequence numbers into a bounded channel.
-    /// 2. **Dispatcher thread** (dedicated, lightweight) — drains that channel
-    ///    and submits one pool job per chunk.  The bounded channel provides
-    ///    natural back-pressure on the I/O thread when the pool is busy.
+    /// 2. **Bounded dispatcher** — submits chunk jobs to the pool while a token
+    ///    limit bounds queued and running work. It stops draining the input
+    ///    channel when that limit is reached, which applies back-pressure to I/O.
    /// 3. **Pool jobs** — parse lines, validate UTF-8, apply ANSI stripping and
    ///    field transforms, and create `DefaultSkimItem` + `Arc` per line.
-    ///    Because these jobs share the same pool as the matcher, reader and
-    ///    matcher compete for the same thread budget rather than over-subscribing
-    ///    available CPU cores.
-    /// 4. **Reorder thread** (dedicated) — collects `(seq, items)` from pool
-    ///    jobs and emits them in sequence order so downstream index assignment
+    /// 4. **Reorder thread** (dedicated) — collects `(seq, items)` from workers
+    ///    and emits them in sequence order so downstream index assignment
    ///    and `--tac` behaviour are correct.
    ///
    /// When `child` is `Some`, a **killer thread** is also spawned.  It waits
@ -303,18 +300,27 @@ impl SkimItemReader {
        // Stage 1: I/O thread.
        Self::spawn_io_reader(source, tx_chunks, line_ending);

-        // Stage 2: dispatcher thread — bridges the bounded channel to the pool.
+        // Stage 2: dispatch at most a fixed number of queued or running jobs.
+        // A worker returns its token only after it sends the parsed result.
+        let max_in_flight = num_threads * 4;
+        let (tx_permits, rx_permits) = std::sync::mpsc::sync_channel(max_in_flight);
+        for _ in 0..max_in_flight {
+            tx_permits.send(()).expect("permit receiver is alive");
+        }
        thread::spawn(move || {
            while let Ok((seq, chunk)) = rx_chunks.recv() {
+                if rx_permits.recv().is_err() {
+                    break;
+                }
                let tx = tx_results.clone();
+                let return_permit = tx_permits.clone();
                let opt = option.clone();
                pool.spawn(move || {
                    let result = Self::process_chunk(seq, &chunk, &opt);
                    let _ = tx.send(result);
+                    let _ = return_permit.send(());
                });
            }
-            // rx_chunks closed → all chunks dispatched; tx_results dropped here
-            // so the reorder thread exits once the last pool job finishes.
        });

        // A zero-capacity channel used as a completion signal: the reorder
--- a/src/matcher.rs
+++ b/src/matcher.rs
@ -2,14 +2,13 @@
 use crate::thread_pool::{self, ThreadPool};
 use crate::tui::item_list::{MergeStrategy, ProcessedItems};
 use std::rc::Rc;
-use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};

 use crate::engine::normalized::NormalizedEngineFactory;
 use crate::engine::split::SplitMatchEngineFactory;
 use crate::item::{ItemPool, MatchedItem, RankBuilder};
 use crate::prelude::{AndOrEngineFactory, ExactOrFuzzyEngineFactory, RegexEngineFactory};
-use crate::spinlock::SpinLock;
 use crate::{CaseMatching, MatchEngineFactory, SkimItem, SkimOptions};

 /// Merges per-worker match results and writes them into `processed_items`.
@ -37,10 +36,16 @@ fn input_index(tac: bool, start: usize, batch_len: usize, batch_index: usize) ->
 fn merge_worker_results(
    worker_results: Vec<Vec<MatchedItem>>,
    no_sort: bool,
-    processed_items: &SpinLock<Option<ProcessedItems>>,
+    processed_items: &Mutex<Option<ProcessedItems>>,
    merge_strategy: MergeStrategy,
+    generation: usize,
+    current_generation: &AtomicUsize,
    needs_render: &AtomicBool,
 ) {
+    if current_generation.load(Ordering::Acquire) != generation {
+        return;
+    }
+
    let total_len: usize = worker_results.iter().map(Vec::len).sum();
    let mut items = Vec::with_capacity(total_len);
    for chunk in worker_results {
@ -55,18 +60,32 @@ fn merge_worker_results(

    trace!("matcher stop, total matched: {}", items.len());

-    // Single lock, single write into processed_items.
-    let mut guard = processed_items.lock();
+    // Validate while holding the result lock so an old matcher cannot overwrite
+    // results that belong to a newer query generation.
+    let mut guard = processed_items
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    if current_generation.load(Ordering::Acquire) != generation {
+        return;
+    }
    if matches!(merge_strategy, MergeStrategy::Replace) {
        *guard = Some(ProcessedItems {
            items,
            merge: MergeStrategy::Replace,
+            generation,
        });
        drop(guard);
        needs_render.store(true, Ordering::Relaxed);
        return;
    }
    match &mut *guard {
+        Some(existing) if existing.generation != generation => {
+            *guard = Some(ProcessedItems {
+                items,
+                merge: MergeStrategy::Replace,
+                generation,
+            });
+        }
        Some(existing) => {
            if no_sort {
                if matches!(merge_strategy, MergeStrategy::Prepend) {
@ -84,6 +103,7 @@ fn merge_worker_results(
            *guard = Some(ProcessedItems {
                items,
                merge: merge_strategy,
+                generation,
            });
        }
    }
@ -200,13 +220,14 @@ impl Matcher {
    #[must_use]
    pub fn create_engine_factory_with_builder(options: &SkimOptions) -> (Rc<dyn MatchEngineFactory>, Arc<RankBuilder>) {
        if options.regex {
-            let regex_factory = RegexEngineFactory::builder();
+            let rank_builder = Arc::new(RankBuilder::new(options.tiebreak.clone()).tac(options.tac));
+            let regex_factory = RegexEngineFactory::builder().rank_builder(rank_builder.clone());
            let factory: Rc<dyn MatchEngineFactory> = if options.normalize {
                Rc::new(NormalizedEngineFactory::new(regex_factory))
            } else {
                Rc::new(regex_factory)
            };
-            (factory, Arc::new(RankBuilder::default().tac(options.tac)))
+            (factory, rank_builder)
        } else {
            let rank_builder = Arc::new(RankBuilder::new(options.tiebreak.clone()).tac(options.tac));
            log::debug!("Creating matcher for algo {:?}", options.algorithm);
@ -275,10 +296,12 @@ impl Matcher {
        query: &str,
        item_pool: &Arc<ItemPool>,
        thread_pool: &Arc<ThreadPool>,
-        processed_items: Arc<SpinLock<Option<ProcessedItems>>>,
+        processed_items: Arc<Mutex<Option<ProcessedItems>>>,
        merge_strategy: MergeStrategy,
        no_sort: bool,
        tac: bool,
+        generation: usize,
+        current_generation: Arc<AtomicUsize>,
        needs_render: Arc<AtomicBool>,
    ) -> MatcherControl {
        let matcher_engine = self.engine_factory.create_engine_with_case(query, self.case_matching);
@ -401,7 +424,15 @@ impl Matcher {
                        return;
                    }

-                    merge_worker_results(worker_results, no_sort, &processed_items, merge_strategy, &needs_render);
+                    merge_worker_results(
+                        worker_results,
+                        no_sort,
+                        &processed_items,
+                        merge_strategy,
+                        generation,
+                        &current_generation,
+                        &needs_render,
+                    );
                },
            );
            stopped.store(true, Ordering::Relaxed);
@ -469,31 +500,123 @@ mod tests {
        assert!(engine.match_item(&"foobar".to_string()).is_some());
    }

+    #[test]
+    fn regex_factory_uses_configured_tiebreak() {
+        let options = SkimOptionsBuilder::default()
+            .regex(true)
+            .tiebreak(vec![crate::RankCriteria::Length])
+            .build()
+            .unwrap();
+        let (factory, rank_builder) = Matcher::create_engine_factory_with_builder(&options);
+        let engine = factory.create_engine("a");
+        let mut matches: Vec<_> = ["aaaa", "a", "aaa"]
+            .into_iter()
+            .enumerate()
+            .map(|(index, text)| {
+                let item: Arc<dyn SkimItem> = Arc::new(text.to_string());
+                let mut result = engine.match_item(item.as_ref()).unwrap();
+                result.rank.index = i32::try_from(index).unwrap();
+                MatchedItem::new(item, result.rank, Some(result.matched_range), &rank_builder)
+            })
+            .collect();
+        matches.sort();
+
+        let output: Vec<_> = matches.iter().map(|item| item.text().into_owned()).collect();
+        assert_eq!(output, ["a", "aaa", "aaaa"]);
+    }
+
+    fn merge_test_results(
+        worker_results: Vec<Vec<MatchedItem>>,
+        no_sort: bool,
+        processed_items: &Mutex<Option<ProcessedItems>>,
+        merge_strategy: MergeStrategy,
+        needs_render: &AtomicBool,
+    ) {
+        let generation = AtomicUsize::new(0);
+        merge_worker_results(
+            worker_results,
+            no_sort,
+            processed_items,
+            merge_strategy,
+            0,
+            &generation,
+            needs_render,
+        );
+    }
+
    #[test]
    fn merge_worker_results_replace_sorts() {
-        let processed = SpinLock::new(None);
+        let processed = Mutex::new(None);
        let needs_render = AtomicBool::new(false);
        let workers = vec![vec![matched("b", 1)], vec![matched("a", 0)]];
-        merge_worker_results(workers, false, &processed, MergeStrategy::Replace, &needs_render);
+        merge_test_results(workers, false, &processed, MergeStrategy::Replace, &needs_render);

        assert!(needs_render.load(Ordering::Relaxed));
-        let guard = processed.lock();
+        let guard = processed.lock().unwrap();
        let items = &guard.as_ref().unwrap().items;
        assert_eq!(items.len(), 2);
    }

+    #[test]
+    fn stale_generation_cannot_publish_results() {
+        let processed = Mutex::new(None);
+        let needs_render = AtomicBool::new(false);
+        let generation = AtomicUsize::new(2);
+
+        merge_worker_results(
+            vec![vec![matched("stale", 0)]],
+            false,
+            &processed,
+            MergeStrategy::Replace,
+            1,
+            &generation,
+            &needs_render,
+        );
+
+        assert!(processed.lock().unwrap().is_none());
+        assert!(!needs_render.load(Ordering::Relaxed));
+    }
+
+    #[test]
+    fn current_results_replace_pending_stale_generation() {
+        let processed = Mutex::new(Some(ProcessedItems {
+            items: vec![matched("stale", 0)],
+            merge: MergeStrategy::SortedMerge,
+            generation: 0,
+        }));
+        let needs_render = AtomicBool::new(false);
+        let generation = AtomicUsize::new(1);
+
+        merge_worker_results(
+            vec![vec![matched("current", 1)]],
+            false,
+            &processed,
+            MergeStrategy::SortedMerge,
+            1,
+            &generation,
+            &needs_render,
+        );
+
+        let guard = processed.lock().unwrap();
+        let result = guard.as_ref().unwrap();
+        assert_eq!(result.generation, 1);
+        assert_eq!(result.items.len(), 1);
+        assert_eq!(result.items[0].text(), "current");
+        assert!(matches!(result.merge, MergeStrategy::Replace));
+    }
+
    #[test]
    fn merge_worker_results_no_sort_preserves_chunk_order() {
-        let processed = SpinLock::new(None);
+        let processed = Mutex::new(None);
        let needs_render = AtomicBool::new(false);
        let workers = vec![
            vec![matched("a", 0), matched("b", 1)],
            vec![matched("c", 2), matched("d", 3)],
            vec![matched("e", 4), matched("f", 5)],
        ];
-        merge_worker_results(workers, true, &processed, MergeStrategy::Replace, &needs_render);
+        merge_test_results(workers, true, &processed, MergeStrategy::Replace, &needs_render);

-        let guard = processed.lock();
+        let guard = processed.lock().unwrap();
        let items = &guard.as_ref().unwrap().items;
        let indexes: Vec<i32> = items.iter().map(|item| item.rank.index).collect();
        assert_eq!(indexes, vec![0, 1, 2, 3, 4, 5]);
@ -501,11 +624,11 @@ mod tests {

    #[test]
    fn merge_worker_results_append_no_sort_extends_existing() {
-        let processed = SpinLock::new(None);
+        let processed = Mutex::new(None);
        let needs_render = AtomicBool::new(false);

        // First append establishes the existing list.
-        merge_worker_results(
+        merge_test_results(
            vec![vec![matched("a", 0)]],
            true,
            &processed,
@ -513,7 +636,7 @@ mod tests {
            &needs_render,
        );
        // Second append with no_sort extends the existing list in place.
-        merge_worker_results(
+        merge_test_results(
            vec![vec![matched("b", 1)]],
            true,
            &processed,
@ -521,7 +644,7 @@ mod tests {
            &needs_render,
        );

-        let guard = processed.lock();
+        let guard = processed.lock().unwrap();
        assert_eq!(guard.as_ref().unwrap().items.len(), 2);
    }

@ -538,17 +661,17 @@ mod tests {

    #[test]
    fn merge_worker_results_prepend_no_sort_places_new_batch_first() {
-        let processed = SpinLock::new(None);
+        let processed = Mutex::new(None);
        let needs_render = AtomicBool::new(false);

-        merge_worker_results(
+        merge_test_results(
            vec![vec![matched("c", 2), matched("b", 1), matched("a", 0)]],
            true,
            &processed,
            MergeStrategy::Prepend,
            &needs_render,
        );
-        merge_worker_results(
+        merge_test_results(
            vec![vec![matched("e", 4), matched("d", 3)]],
            true,
            &processed,
@ -556,7 +679,7 @@ mod tests {
            &needs_render,
        );

-        let guard = processed.lock();
+        let guard = processed.lock().unwrap();
        let indexes: Vec<i32> = guard
            .as_ref()
            .unwrap()
--- a/src/reader.rs
+++ b/src/reader.rs
@ -10,7 +10,9 @@ use crate::{SkimItem, SkimItemReceiver};
 use std::cell::RefCell;
 use std::rc::Rc;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::thread::JoinHandle;
+use std::time::Duration;

 /// Trait for collecting items from command output
 pub trait CommandCollector {
@ -39,6 +41,7 @@ pub struct ReaderControl {
    tx_interrupt: Sender<i32>,
    tx_interrupt_cmd: Option<Sender<i32>>,
    components_to_stop: Arc<AtomicUsize>,
+    collector_handle: Option<JoinHandle<()>>,
    items: Arc<SpinLock<Vec<Arc<dyn SkimItem>>>>,
 }

@ -52,7 +55,14 @@ impl ReaderControl {

        let _ = self.tx_interrupt_cmd.clone().map(|tx| tx.send(1));
        let _ = self.tx_interrupt.send(1);
-        while self.components_to_stop.load(Ordering::SeqCst) != 0 {}
+        if let Some(handle) = self.collector_handle.take() {
+            let _ = handle.join();
+        }
+        // Command collectors can own additional components outside the reader's
+        // join handle. Wait without consuming a CPU while they process the signal.
+        while self.components_to_stop.load(Ordering::Acquire) != 0 {
+            std::thread::sleep(Duration::from_millis(1));
+        }
    }

    /// Takes all items collected so far
@ -123,12 +133,14 @@ impl Reader {
        );

        let components_to_stop_clone = components_to_stop.clone();
-        let tx_interrupt = collect_items(components_to_stop_clone, rx_item, move |items| _ = app_tx.send(items));
+        let (tx_interrupt, collector_handle) =
+            collect_items(components_to_stop_clone, rx_item, move |items| _ = app_tx.send(items));

        ReaderControl {
            tx_interrupt,
            tx_interrupt_cmd,
            components_to_stop,
+            collector_handle: Some(collector_handle),
            items,
        }
    }
@ -149,7 +161,7 @@ impl Reader {
        );

        let components_to_stop_clone = components_to_stop.clone();
-        let tx_interrupt = collect_items(components_to_stop_clone, rx_item, move |items| {
+        let (tx_interrupt, collector_handle) = collect_items(components_to_stop_clone, rx_item, move |items| {
            item_pool.append(items);
        });
        debug!("collect: started ({components_to_stop:?} components)");
@ -158,6 +170,7 @@ impl Reader {
            tx_interrupt,
            tx_interrupt_cmd,
            components_to_stop,
+            collector_handle: Some(collector_handle),
            items,
        }
    }
@ -172,18 +185,19 @@ impl Default for Reader {
    }
 }

-fn collect_items<F>(components_to_stop: Arc<AtomicUsize>, rx_item: SkimItemReceiver, callback: F) -> Sender<i32>
+fn collect_items<F>(
+    components_to_stop: Arc<AtomicUsize>,
+    rx_item: SkimItemReceiver,
+    callback: F,
+) -> (Sender<i32>, JoinHandle<()>)
 where
    F: Fn(Vec<Arc<dyn SkimItem>>) + Send + 'static,
 {
    let (tx_interrupt, rx_interrupt) = crate::prelude::bounded(8);

-    let started = Arc::new(AtomicBool::new(false));
-    let started_clone = started.clone();
-    std::thread::spawn(move || {
+    components_to_stop.fetch_add(1, Ordering::AcqRel);
+    let handle = std::thread::spawn(move || {
        debug!("collect_item start");
-        components_to_stop.fetch_add(1, Ordering::SeqCst);
-        started_clone.store(true, Ordering::SeqCst); // notify parent that it is started

        loop {
            if let Ok(Some(msg)) = rx_interrupt.try_recv() {
@ -205,15 +219,11 @@ where
            }
        }

-        components_to_stop.fetch_sub(1, Ordering::SeqCst);
+        components_to_stop.fetch_sub(1, Ordering::AcqRel);
        debug!("collect_item stop");
    });

-    while !started.load(Ordering::SeqCst) {
-        // busy waiting for the thread to start. (components_to_stop is added)
-    }
-
-    tx_interrupt
+    (tx_interrupt, handle)
 }

 #[cfg(test)]
--- a/src/skim.rs
+++ b/src/skim.rs
@ -470,6 +470,7 @@ where
                .item_list
                .processed_items
                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .take()
                .unwrap_or_default()
                .items
@ -521,7 +522,14 @@ where
                app.matcher_control.get_num_matched()
            );
            if app.matcher_control.get_num_matched() == min_items_before_enter - 1 {
-                app.item_list.items = app.item_list.processed_items.lock().take().unwrap_or_default().items;
+                app.item_list.items = app
+                    .item_list
+                    .processed_items
+                    .lock()
+                    .unwrap_or_else(std::sync::PoisonError::into_inner)
+                    .take()
+                    .unwrap_or_default()
+                    .items;
                debug!("early exit, result: {:?}", app.results());
                return false;
            }
--- a/src/spinlock.rs
+++ b/src/spinlock.rs
@ -26,9 +26,9 @@ pub struct SpinLockGuard<'a, T: ?Sized + 'a> {
 }

 impl<'a, T: ?Sized + 'a> SpinLockGuard<'a, T> {
-    /// Creates a new guard for the given lock
-    pub fn new(pool: &'a SpinLock<T>) -> SpinLockGuard<'a, T> {
-        Self { __lock: pool }
+    /// Creates a guard after its lock has been acquired.
+    fn new(lock: &'a SpinLock<T>) -> SpinLockGuard<'a, T> {
+        Self { __lock: lock }
    }
 }

--- a/src/tui/app.rs
+++ b/src/tui/app.rs
@ -582,7 +582,7 @@ impl App {
                            u16::try_from(u32::from(self.preview.rows) * u32::from(p) / 100).unwrap_or(u16::MAX)
                        }
                    };
-                    self.preview.scroll_y = v_scroll;
+                    self.preview.scroll_y = usize::from(v_scroll);
                    self.preview.scroll_down(v_offset);

                    let h_scroll = match preview_position.h_scroll {
@ -599,7 +599,7 @@ impl App {
                            u16::try_from(u32::from(self.preview.cols) * u32::from(p) / 100).unwrap_or(u16::MAX)
                        }
                    };
-                    self.preview.scroll_x = h_scroll.saturating_add(h_offset);
+                    self.preview.scroll_x = usize::from(h_scroll.saturating_add(h_offset));
                }
                ItemPreview::TextWithPos(t, preview_position) | ItemPreview::AnsiWithPos(t, preview_position) => self
                    .preview
@ -1381,6 +1381,12 @@ impl App {
        if self.query_below_min_length() {
            // Query is too short, clear items and don't run matcher
            self.matcher_control.kill();
+            self.item_list.matcher_generation.fetch_add(1, Ordering::AcqRel);
+            self.item_list
+                .processed_items
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .take();
            self.item_list.items.clear();
            self.item_list.current = 0;
            self.item_list.offset = 0;
@ -1413,6 +1419,12 @@ impl App {
                self.item_pool.reset();
            }

+            let generation = if force {
+                self.item_list.matcher_generation.fetch_add(1, Ordering::AcqRel) + 1
+            } else {
+                self.item_list.matcher_generation.load(Ordering::Acquire)
+            };
+
            let merge_strategy = if force {
                MergeStrategy::Replace
            } else if no_sort && self.options.tac {
@ -1431,6 +1443,8 @@ impl App {
                merge_strategy,
                no_sort,
                self.options.tac,
+                generation,
+                self.item_list.matcher_generation.clone(),
                self.needs_render.clone(),
            );
            // A new search is in flight; arm the `result`/`zero`/`one` events to
--- a/src/tui/item_list.rs
+++ b/src/tui/item_list.rs
@ -1,5 +1,6 @@
 use std::rc::Rc;
-use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};

 use indexmap::IndexSet;
 use ratatui::widgets::{
@ -10,7 +11,6 @@ use regex::Regex;

 use crate::item::MatchedItem;
 use crate::options::feature_flag;
-use crate::spinlock::SpinLock;
 use crate::theme::ColorTheme;
 use crate::tui::BorderType;
 use crate::tui::item_renderer::ItemRenderer;
@ -36,6 +36,7 @@ pub(crate) enum MergeStrategy {
 pub(crate) struct ProcessedItems {
    pub(crate) items: Vec<MatchedItem>,
    pub(crate) merge: MergeStrategy,
+    pub(crate) generation: usize,
 }

 impl Default for ProcessedItems {
@ -43,6 +44,7 @@ impl Default for ProcessedItems {
        Self {
            items: Vec::new(),
            merge: MergeStrategy::Replace,
+            generation: 0,
        }
    }
 }
@ -52,7 +54,8 @@ impl Default for ProcessedItems {
 pub struct ItemList {
    pub(crate) items: Vec<MatchedItem>,
    pub(crate) selection: IndexSet<MatchedItem>,
-    pub(crate) processed_items: Arc<SpinLock<Option<ProcessedItems>>>,
+    pub(crate) processed_items: Arc<Mutex<Option<ProcessedItems>>>,
+    pub(crate) matcher_generation: Arc<AtomicUsize>,
    pub(crate) direction: ListDirection,
    pub(crate) offset: usize,
    /// How many leading sub-lines of items[offset] have been scrolled off the top.
@ -420,7 +423,8 @@ impl SkimWidget for ItemList {
            (None, 0)
        };

-        let processed_items = Arc::new(SpinLock::new(None));
+        let processed_items = Arc::new(Mutex::new(None));
+        let matcher_generation = Arc::new(AtomicUsize::new(0));

        let interactive = options.interactive;
        let no_clear_if_empty = options.no_clear_if_empty;
@ -429,6 +433,7 @@ impl SkimWidget for ItemList {
        // Spawn background processing thread with the appropriate configuration
        Self {
            processed_items,
+            matcher_generation,
            reserved: 0, // header_lines are now displayed in the Header widget, not ItemList
            direction: match options.layout {
                TuiLayout::Default => ratatui::widgets::ListDirection::BottomToTop,
@ -511,7 +516,13 @@ impl SkimWidget for ItemList {
        // Check for pre-processed items from background thread (non-blocking).
        // Bind the result separately so the lock guard is dropped before a merge
        // mutates the item list.
-        let processed = this.processed_items.lock().take();
+        let processed = this
+            .processed_items
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .take();
+        let current_generation = this.matcher_generation.load(Ordering::Acquire);
+        let processed = processed.filter(|result| result.generation == current_generation);
        let items_updated = if let Some(processed) = processed {
            debug!("Render: Got {} processed items", processed.items.len());

--- a/src/tui/item_list_tests.rs
+++ b/src/tui/item_list_tests.rs
@ -240,7 +240,11 @@ fn render_list(il: &mut ItemList, w: u16, h: u16) {
 }

 fn set_processed(il: &ItemList, items: Vec<MatchedItem>, merge: MergeStrategy) {
-    *il.processed_items.lock() = Some(ProcessedItems { items, merge });
+    *il.processed_items.lock().unwrap() = Some(ProcessedItems {
+        items,
+        merge,
+        generation: il.matcher_generation.load(std::sync::atomic::Ordering::Acquire),
+    });
 }

 #[test]
@ -253,6 +257,21 @@ fn render_applies_replace_strategy() {
    assert_eq!(il.items[0].text(), "new");
 }

+#[test]
+fn render_discards_results_from_stale_generation() {
+    let mut il = list(2);
+    *il.processed_items.lock().unwrap() = Some(ProcessedItems {
+        items: vec![matched("stale", 0)],
+        merge: MergeStrategy::Replace,
+        generation: 0,
+    });
+    il.matcher_generation.store(1, std::sync::atomic::Ordering::Release);
+
+    render_list(&mut il, 20, 5);
+    assert_eq!(il.items.len(), 2);
+    assert!(il.items.iter().all(|item| item.text() != "stale"));
+}
+
 #[test]
 fn render_applies_append_strategy() {
    let mut il = list(2);
--- a/src/tui/preview.rs
+++ b/src/tui/preview.rs
@ -15,9 +15,10 @@ use tui_term::widget::PseudoTerminal;

 use std::env;
 use std::io::Read;
-use std::sync::{Arc, RwLock, mpsc};
+use std::process::{Child, Stdio};
+use std::sync::{Arc, Mutex, RwLock, mpsc};
 use std::thread::JoinHandle;
-use std::time::Instant;
+use std::time::{Duration, Instant};

 use super::statusline::spinner_char;
 use super::util::{find_csi_end, find_osc_end, handle_csi_query, handle_osc_query};
@ -31,6 +32,50 @@ use crate::{SkimItem, SkimOptions};
 pub type PreviewCallbackFn = dyn Fn(Vec<Arc<dyn SkimItem>>) -> Vec<String> + Send + Sync + 'static;
 const PREVIEW_MAX_BYTES: usize = 1024 * 1024;
 const VT_SCROLLBACK: usize = 100_000;
+type PlainChild = Arc<Mutex<Option<Child>>>;
+
+fn read_bounded(mut reader: impl Read) -> Vec<u8> {
+    let mut output = Vec::with_capacity(PREVIEW_MAX_BYTES);
+    let mut buffer = [0; 8192];
+    loop {
+        match reader.read(&mut buffer) {
+            Ok(0) | Err(_) => break,
+            Ok(read) => {
+                let retained = PREVIEW_MAX_BYTES.saturating_sub(output.len()).min(read);
+                output.extend_from_slice(&buffer[..retained]);
+            }
+        }
+    }
+    output
+}
+
+fn terminate_plain_child(child: &PlainChild) {
+    let Ok(mut guard) = child.lock() else {
+        return;
+    };
+    let Some(child) = guard.as_mut() else {
+        return;
+    };
+
+    #[cfg(unix)]
+    if let Ok(process_group) = i32::try_from(child.id()) {
+        use nix::sys::signal::{Signal, killpg};
+        use nix::unistd::Pid;
+
+        let _ = killpg(Pid::from_raw(process_group), Signal::SIGKILL);
+    }
+
+    #[cfg(windows)]
+    {
+        let _ = std::process::Command::new("taskkill")
+            .args(["/PID", &child.id().to_string(), "/T", "/F"])
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .status();
+    }
+
+    let _ = child.kill();
+}

 /// Preview content options
 pub(crate) enum PreviewContent {
@ -81,11 +126,12 @@ pub struct Preview {
    pub cmd: String,
    pub rows: u16,
    pub cols: u16,
-    pub scroll_y: u16,
-    pub scroll_x: u16,
+    pub scroll_y: usize,
+    pub scroll_x: usize,
    pub thread_handle: Option<JoinHandle<()>>,
    /// Channel to signal thread interruption
    interrupt_tx: Option<mpsc::Sender<()>>,
+    plain_child: Option<PlainChild>,
    pub theme: Arc<ColorTheme>,
    /// Border type
    pub border: BorderType,
@ -97,7 +143,7 @@ pub struct Preview {
    image: bool,
    #[cfg(feature = "image")]
    image_picker: Option<Picker>,
-    pub total_lines: u16,
+    pub total_lines: usize,
    loading: bool,
    spinner_start: Instant,
 }
@ -150,18 +196,12 @@ impl Preview {
    }

    /// Convert a Size value to an actual offset based on preview dimensions
-    fn size_to_offset(&self, size: super::Size, is_vertical: bool) -> u16 {
+    fn size_to_offset(&self, size: super::Size, is_vertical: bool) -> usize {
+        let dimension = if is_vertical { self.rows } else { self.cols };
        match size {
-            super::Size::Fixed(n) => n,
-            super::Size::Percent(p) => {
-                let dimension = if is_vertical { self.rows } else { self.cols };
-                // Result is at most dimension (a u16), so truncation cannot occur.
-                u16::try_from(u32::from(dimension) * u32::from(p) / 100).unwrap_or(u16::MAX)
-            }
-            super::Size::Neg(n) => {
-                let dimension = if is_vertical { self.rows } else { self.cols };
-                dimension.saturating_sub(n)
-            }
+            super::Size::Fixed(n) => usize::from(n),
+            super::Size::Percent(p) => usize::from(dimension) * usize::from(p) / 100,
+            super::Size::Neg(n) => usize::from(dimension.saturating_sub(n)),
        }
    }

@ -226,7 +266,7 @@ impl Preview {
        let Ok(mut content) = self.content.write() else {
            return Err(eyre::eyre!("Failed to acquire content for writing"));
        };
-        self.total_lines = text.lines.len().try_into().unwrap();
+        self.total_lines = text.lines.len();
        *content = PreviewContent::Text(text);
        self.scroll_y = 0;
        self.scroll_x = 0;
@ -256,7 +296,7 @@ impl Preview {
    }

    pub fn scroll_up(&mut self, lines: u16) {
-        self.scroll_y = self.scroll_y.saturating_sub(lines);
+        self.scroll_y = self.scroll_y.saturating_sub(usize::from(lines));
    }

    pub fn scroll_down(&mut self, lines: u16) {
@ -265,26 +305,26 @@ impl Preview {
            self.total_lines, self.rows
        );
        if self.total_lines > 0 {
-            self.scroll_y = self
-                .scroll_y
-                .saturating_add(lines)
-                .min(self.total_lines.saturating_sub(self.rows.saturating_sub(1)));
+            self.scroll_y = self.scroll_y.saturating_add(usize::from(lines)).min(
+                self.total_lines
+                    .saturating_sub(usize::from(self.rows.saturating_sub(1))),
+            );
        } else {
            // We might not have the actual total_lines value
-            self.scroll_y = self.scroll_y.saturating_add(lines);
+            self.scroll_y = self.scroll_y.saturating_add(usize::from(lines));
        }
    }

    pub fn scroll_left(&mut self, cols: u16) {
-        self.scroll_x = self.scroll_x.saturating_sub(cols);
+        self.scroll_x = self.scroll_x.saturating_sub(usize::from(cols));
    }

    pub fn scroll_right(&mut self, cols: u16) {
-        self.scroll_x = self.scroll_x.saturating_add(cols);
+        self.scroll_x = self.scroll_x.saturating_add(usize::from(cols));
    }

    pub fn set_offset(&mut self, offset: u16) {
-        self.scroll_y = offset.saturating_sub(1); // -1 because line numbers are 1-indexed
+        self.scroll_y = usize::from(offset.saturating_sub(1)); // -1 because line numbers are 1-indexed
    }

    pub fn page_up(&mut self) {
@ -302,6 +342,11 @@ impl Preview {
            let _ = tx.send(());
        }

+        if let Some(child) = self.plain_child.take() {
+            trace!("killing plain preview child process group");
+            terminate_plain_child(&child);
+        }
+
        if let Some(mut child) = self.pty_child.take() {
            trace!("killing pty child process");
            match child.try_wait() {
@ -490,39 +535,83 @@ impl Preview {
            shell_cmd
                .env("ROWS", self.rows.to_string())
                .env("COLUMNS", self.cols.to_string())
-                .env("PAGER", "");
+                .env("PAGER", "")
+                .stdout(Stdio::piped())
+                .stderr(Stdio::piped());
            if let Ok(cwd) = env::current_dir() {
                shell_cmd.current_dir(cwd);
            }
+            #[cfg(unix)]
+            {
+                use std::os::unix::process::CommandExt as _;
+                shell_cmd.process_group(0);
+            }
+            #[cfg(windows)]
+            {
+                use std::os::windows::process::CommandExt as _;
+                const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
+                shell_cmd.creation_flags(CREATE_NEW_PROCESS_GROUP);
+            }

            let (interrupt_tx, interrupt_rx) = mpsc::channel();
            self.interrupt_tx = Some(interrupt_tx);

-            self.thread_handle = Some(std::thread::spawn(move || {
-                if interrupt_rx.try_recv().is_ok() {
-                    return;
-                }
-
-                let try_out = shell_cmd.output();
-                if try_out.is_err() {
-                    log::info!("Shell cmd in error: {try_out:?}");
+            let mut child = match shell_cmd.spawn() {
+                Ok(child) => child,
+                Err(error) => {
+                    log::info!("Shell cmd in error: {error:?}");
                    let _ = event_tx_clone.blocking_send(Event::PreviewReady);
-                    return;
+                    return Ok(());
                }
+            };
+            let stdout = child.stdout.take().expect("stdout was configured as piped");
+            let stderr = child.stderr.take().expect("stderr was configured as piped");
+            let child = Arc::new(Mutex::new(Some(child)));
+            self.plain_child = Some(child.clone());

-                let mut out = try_out.unwrap();
+            self.thread_handle = Some(std::thread::spawn(move || {
+                let stdout_reader = std::thread::spawn(move || read_bounded(stdout));
+                let stderr_reader = std::thread::spawn(move || read_bounded(stderr));

-                if interrupt_rx.try_recv().is_ok() {
-                    return;
-                }
-
-                if let Ok(mut c) = content.write() {
-                    if out.status.success() {
-                        out.stdout.resize(PREVIEW_MAX_BYTES.min(out.stdout.len()), 0);
-                        *c = PreviewContent::Text(out.stdout.into_text().unwrap_or_default());
-                    } else {
-                        *c = PreviewContent::Text(out.stderr.clone().into_text().unwrap_or_default());
+                let status = loop {
+                    match interrupt_rx.recv_timeout(Duration::from_millis(10)) {
+                        Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => {
+                            terminate_plain_child(&child);
+                            break None;
+                        }
+                        Err(mpsc::RecvTimeoutError::Timeout) => {}
                    }
+
+                    let wait_result = match child.lock() {
+                        Ok(mut guard) => guard.as_mut().map(Child::try_wait),
+                        Err(_) => break None,
+                    };
+                    match wait_result {
+                        Some(Ok(Some(status))) => break Some(status),
+                        Some(Ok(None)) => {}
+                        Some(Err(error)) => {
+                            log::info!("Failed to wait for preview command: {error:?}");
+                            break None;
+                        }
+                        None => break None,
+                    }
+                };
+
+                // A shell can exit while a background descendant still owns the pipes.
+                // Terminate the whole process group before joining the drain threads.
+                terminate_plain_child(&child);
+                let stdout = stdout_reader.join().unwrap_or_default();
+                let stderr = stderr_reader.join().unwrap_or_default();
+                if let Ok(mut guard) = child.lock() {
+                    guard.take();
+                }
+
+                let Some(status) = status else {
+                    return;
+                };
+                if let Ok(mut c) = content.write() {
+                    let output = if status.success() { stdout } else { stderr };
+                    *c = PreviewContent::Text(output.into_text().unwrap_or_default());
                }

                trace!("sending ready ping");
@ -538,12 +627,14 @@ impl Preview {
        area: ratatui::layout::Rect,
        buf: &mut ratatui::prelude::Buffer,
        text: &Text,
-    ) -> u16 {
+    ) -> usize {
        // Calculate total lines in content
-        let total_lines: u16 = text.lines.len().try_into().unwrap();
+        let total_lines = text.lines.len();

-        // Create paragraph with optional block
-        let mut paragraph = Paragraph::new(text.clone()).scroll((self.scroll_y, self.scroll_x));
+        // Ratatui terminal coordinates are u16. Saturate previews that exceed that range.
+        let scroll_y = u16::try_from(self.scroll_y).unwrap_or(u16::MAX);
+        let scroll_x = u16::try_from(self.scroll_x).unwrap_or(u16::MAX);
+        let mut paragraph = Paragraph::new(text.clone()).scroll((scroll_y, scroll_x));

        // Enable wrapping if wrap is true
        if self.wrap {
@ -552,7 +643,7 @@ impl Preview {

        // Add scroll position indicator at top-right if scrolled
        if self.scroll_y > 0 && total_lines > 0 {
-            let current_line = (self.scroll_y + 1) as usize; // +1 because scroll_y is 0-indexed but we want 1-indexed display
+            let current_line = self.scroll_y.saturating_add(1); // Display line numbers are 1-indexed.
            let title = format!("{current_line}/{total_lines}");

            outer = outer.title_top(Line::from(title).alignment(Alignment::Right).reversed());
@ -569,23 +660,21 @@ impl Preview {
        area: ratatui::layout::Rect,
        buf: &mut ratatui::prelude::Buffer,
        parser: &std::sync::RwLock<tui_term::vt100::Parser>,
-    ) -> u16 {
-        let mut total_lines = 0u16;
+    ) -> usize {
+        let mut total_lines = 0usize;
        // For terminal content, manipulate scrollback to implement scrolling
        if let Ok(mut parser_guard) = parser.try_write() {
            let scrollback_len = parser_guard.screen().scrollback();
            // Reset scrollback to its full size first
            parser_guard.screen_mut().set_scrollback(VT_SCROLLBACK);
            // If the scrollback is not empty, we seem to be off by one
-            total_lines = (scrollback_len.saturating_sub(1) + parser_guard.screen().contents().lines().count())
-                .try_into()
-                .unwrap();
+            total_lines = scrollback_len.saturating_sub(1) + parser_guard.screen().contents().lines().count();
            if self.scroll_y > 0 {
                trace!("scrolling in vt buffer: {}/{}", self.scroll_y, total_lines);
                // Reduce scrollback by scroll_y to show earlier content
                parser_guard
                    .screen_mut()
-                    .set_scrollback(scrollback_len.saturating_sub(self.scroll_y.into()));
+                    .set_scrollback(scrollback_len.saturating_sub(self.scroll_y));
            }
        }

@ -674,6 +763,7 @@ impl SkimWidget for Preview {
            scroll_x: 0,
            thread_handle: None,
            interrupt_tx: None,
+            plain_child: None,
            pty: None,
            pty_child: None,
            #[cfg(feature = "image")]
--- a/src/tui/preview_tests.rs
+++ b/src/tui/preview_tests.rs
@ -5,7 +5,7 @@ use ratatui::layout::Size;
 #[cfg(feature = "image")]
 use ratatui_image::picker::Picker;

-use super::Preview;
+use super::{PREVIEW_MAX_BYTES, Preview, PreviewContent, read_bounded};

 #[cfg(feature = "image")]
 fn image(width: u32, height: u32) -> DynamicImage {
@ -60,6 +60,51 @@ fn content_loads_text_and_resets_scroll() {
    assert!(!p.is_loading());
 }

+#[test]
+fn large_text_content_does_not_overflow_line_count() {
+    let input = "x\n".repeat(70_000);
+    let mut preview = Preview::default();
+    preview.content(input.as_bytes()).unwrap();
+    assert_eq!(preview.total_lines, 70_000);
+
+    let content = preview.content.read().unwrap();
+    let PreviewContent::Text(text) = &*content else {
+        panic!("expected text preview");
+    };
+    let area = ratatui::layout::Rect::new(0, 0, 20, 5);
+    let mut buffer = ratatui::buffer::Buffer::empty(area);
+    assert_eq!(
+        preview.render_text(ratatui::widgets::Block::new(), area, &mut buffer, text),
+        70_000
+    );
+}
+
+#[test]
+fn bounded_reader_discards_output_after_limit() {
+    let input = vec![b'x'; PREVIEW_MAX_BYTES + 4096];
+    let output = read_bounded(std::io::Cursor::new(input));
+    assert_eq!(output.len(), PREVIEW_MAX_BYTES);
+}
+
+#[cfg(unix)]
+#[test]
+fn plain_preview_can_be_cancelled() {
+    use ratatui::backend::TestBackend;
+    use std::time::{Duration, Instant};
+
+    let mut preview = Preview::default();
+    preview.pty = None;
+    let mut tui =
+        super::super::Tui::new_with_height_and_backend(TestBackend::new(20, 5), super::super::Size::Percent(100))
+            .unwrap();
+    preview.spawn(&mut tui, "sleep 30").unwrap();
+
+    let started = Instant::now();
+    preview.kill();
+    preview.thread_handle.take().unwrap().join().unwrap();
+    assert!(started.elapsed() < Duration::from_secs(2));
+}
+
 #[test]
 fn vertical_scroll_clamps_to_content() {
    let mut p = Preview::default();