perf: stream the listing incrementally instead of load_all on open (M4)

Opening a listing walked the WHOLE segment (load_all) before showing anything --
a blank pane for seconds on a big .text. Now _prime loads just the viewport
around the cursor (appears instantly) and _grow streams the rest in the
background, growing virtual_size as pages land (throttled). Search finishes
loading first (needs the whole segment).

ListingModel gains a _load_lock so the grower and an in-view search can both
drive page loads without double-fetching; new public load_next_page.

Verified: ensure(1100)->partial(1500,incomplete), load_all->full(5036,complete);
3 concurrent loaders -> 0 duplicate heads; UI listing+search 30/0 (in-process).

2 files changed, 60 insertions, 5 deletions

diff --git a/idatui/app.py b/idatui/app.py
index cc9ab03..27dc33e 100644
--- a/idatui/app.py
+++ b/idatui/app.py
@@ -961,10 +961,16 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
         model = self.model
         if model is None:
             return
-        model.load_all()  # walk the whole segment's heads off the UI loop
-        self.app.call_from_thread(self._on_primed, len(model))
+        # Load just enough to render the viewport around the cursor, so the
+        # listing appears immediately even on a huge segment; the rest streams
+        # in via _grow. (load_all here would blank the pane for seconds.)
+        height = max(self.size.height, 1)
+        model.ensure(self.cursor + height + 2 * ListingModel.PAGE)
+        self.app.call_from_thread(self._on_primed, len(model), model.complete)
+        if not model.complete:
+            self._grow()
 
-    def _on_primed(self, total: int) -> None:
+    def _on_primed(self, total: int, complete: bool) -> None:
         if self.model is None:
             return
         self.total = total
@@ -979,7 +985,33 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
         self.refresh()
         self.post_message(ListingView.CursorMoved(self.cursor, self._cursor_ea()))
 
-    # -- search hooks (all heads are in memory after prime) --------------- #
+    @work(thread=True, exclusive=True, group="listing-grow")
+    def _grow(self) -> None:
+        """Stream the rest of the segment's heads in the background, growing the
+        virtual size as they land so the scrollbar/paging catch up."""
+        model = self.model
+        if model is None:
+            return
+        since = 0
+        while not model.complete:
+            if model.load_next_page() == 0:
+                break
+            if self.model is not model:  # a new load() replaced us
+                return
+            since += 1
+            if since >= 4:  # throttle repaints on huge segments
+                since = 0
+                self.app.call_from_thread(self._grew, len(model))
+        self.app.call_from_thread(self._grew, len(model))
+
+    def _grew(self, total: int) -> None:
+        if self.model is None or total <= self.total:
+            return
+        self.total = total
+        self.virtual_size = Size(0, total)
+        self.refresh()
+
+    # -- search hooks ----------------------------------------------------- #
     def _search_line_count(self) -> int:
         return self.total
 
@@ -987,7 +1019,19 @@ class ListingView(SearchMixin, NavMixin, ColumnCursor, ScrollView, can_focus=Tru
         return self._line_plain(i)
 
     def _search_ensure(self, done) -> None:
-        done()
+        # Search needs the whole segment; finish loading it first if streaming.
+        if self.model is not None and not self.model.complete:
+            self._search_load_all(done)
+        else:
+            done()
+
+    @work(thread=True, exclusive=True, group="listing-search-load")
+    def _search_load_all(self, done) -> None:  # type: ignore[no-untyped-def]
+        model = self.model
+        if model is not None:
+            model.load_all()
+            self.app.call_from_thread(self._grew, len(model))
+        self.app.call_from_thread(done)
 
     # -- rendering --------------------------------------------------------- #
     def render_line(self, y: int) -> Strip:
diff --git a/idatui/domain.py b/idatui/domain.py
index f26354c..e70182c 100644
--- a/idatui/domain.py
+++ b/idatui/domain.py
@@ -526,8 +526,19 @@ class ListingModel:
         self._next: int | None = seg_start  # next address to fetch from
         self._done = False
         self._lock = threading.Lock()
+        # Serializes page loads so a background grower and an in-view search can
+        # both drive loading without double-fetching the same page.
+        self._load_lock = threading.Lock()
+
+    def load_next_page(self) -> int:
+        """Load one more page of heads; returns how many were added."""
+        return self._load_next_page()
 
     def _load_next_page(self) -> int:
+        with self._load_lock:
+            return self._load_next_page_locked()
+
+    def _load_next_page_locked(self) -> int:
         with self._lock:
             if self._done or self._next is None:
                 return 0