Only hard re-render `Viz`s matching backfill deats

Avoid unnecessarily re-rendering the wrong (1min OHLC history) chart
and/or other such charts with update tasks listening to the sampler
stream. Instead only redraw in tasks which are updating vizs which match
the actual details of the backfill event.

We can probably also eventually match against a range tuple (emitted in
the msg) and then have the task further only update the formatter layer
unless the range is actually in view?

piker/data/_sampling.py

@@ -296,7 +296,11 @@ class Sampler:
         self,
         info: dict | None = None,
     ) -> None:
-        for period_s in self.subscribers:
+
+        # NOTE: take a copy of subs since removals can happen
+        # during the broadcast checkpoint which can cause
+        # a `RuntimeError` on interation of the underlying `dict`.
+        for period_s in list(self.subscribers):
             await self.broadcast(
                 period_s,
                 info=info,

piker/ui/_display.py

@@ -209,9 +209,11 @@ async def increment_history_view(
     # min_istream: tractor.MsgStream,
     ds: DisplayState,
 ):
-    hist_chart = ds.hist_chart
+    hist_chart: ChartPlotWidget = ds.hist_chart
-    hist_viz = ds.hist_viz
+    hist_viz: Viz = ds.hist_viz
+    viz: Viz = ds.viz
     assert 'hist' in hist_viz.shm.token['shm_name']
+    name: str = hist_viz.name
 
     # TODO: seems this is more reliable at keeping the slow
     # chart incremented in view more correctly?
@@ -221,15 +223,13 @@ async def increment_history_view(
     # - we need increment logic that only does the view shift
     #   call when the uppx permits/needs it
 
-    async with open_sample_stream(1.) as min_istream:
-
     # draw everything from scratch on first entry!
     for curve_name, hist_viz in hist_chart._vizs.items():
-            log.info(f'FORCING CURVE REDRAW -> {curve_name}')
+        log.info(f'Forcing hard redraw -> {curve_name}')
         hist_viz.update_graphics(force_redraw=True)
 
+    async with open_sample_stream(1.) as min_istream:
         async for msg in min_istream:
-            # print(f'SAMPLER MSG: {msg}')
 
             profiler = Profiler(
                 msg=f'History chart cycle for: `{ds.fqme}`',
@@ -239,19 +239,28 @@ async def increment_history_view(
                 # ms_threshold=4,
             )
 
+            # NOTE: when a backfill msg is broadcast from the
+            # history mgmt layer, we match against the equivalent
+            # `Viz` and "hard re-render" (i.e. re-allocate the
+            # in-mem xy-array formats managed in
+            # `.data._formatters) its curve graphics to fill
+            # on-chart gaps.
+            # TODO: specifically emit/handle range tuples?
+            # - samplerd could emit the actual update range via
+            #   tuple and then we only enter the below block if that
+            #   range is detected as in-view?
             if (
-                'backfilling' in msg
+                (bf_wut := msg.get('backfilling', False))
             ):
-                # for curve_name, hist_viz in hist_chart._vizs.items():
+                viz_name, timeframe = bf_wut
-                print(f'FORCING REDRAW!! {hist_viz.name}')
+                if viz_name == name:
+                    log.info(f'Forcing hard redraw -> {name}@{timeframe}')
+                    match timeframe:
+                        case 60:
                             hist_viz.update_graphics(force_redraw=True)
+                        case 1:
+                            viz.update_graphics(force_redraw=True)
 
-            # l3 = ds.viz.shm.array[-3:]
-            # print(
-            #     f'fast step for {ds.flume.mkt.fqme}:\n'
-            #     f'{list(l3["time"])}\n'
-            #     f'{l3}\n'
-            # )
             # check if slow chart needs an x-domain shift and/or
             # y-range resize.
             (