viewer: fix HTTP live-follow OOM via datasetLoaded backpressure + shared pixel mask

The viewer could grow to ~100 GB RAM when live-following an HTTP broker. The
rc.153 images_in_flight backpressure only throttled imageLoaded; the heavy
per-frame payload rides datasetLoaded, fanned out over ~10 queued cross-thread
connections with no cap. In HTTPSyncDataset follow mode (entered when an operator
clicks an image while following live) RefreshDatasetOnly_i emits a fresh full
dataset every autoload tick with no imageLoaded, so the gate never engaged and
the queued events - each pinning a full JFJochReaderDataset (full-detector
PixelMask + per-image plots) - accumulated without bound.

Backpressure datasetLoaded the same way as imageLoaded: a datasets_in_flight
counter (cap 2), all emits routed through EmitDatasetLoaded_i, and
AutoLoadTimerExpired gated on it (covers HTTPSyncDataset). The window routes the
worker's datasetLoaded through a single OnDatasetReady sink that fans out
synchronously via datasetReady and acks with datasetConsumed. Under load stale
datasets are dropped; the next tick sends the latest.

Share the pixel mask instead of deep-copying it: JFJochReaderDataset::pixel_mask
is now shared_ptr<const PixelMask>, so per-frame dataset copies share the ~72 MB
mask. UpdateUserMask does copy-on-write; JFJochHttpReader caches the mask by
arm_date so a live refresh reuses one shared mask per acquisition.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

reader/HDF5MetadataSource.cpp

@@ -692,7 +692,7 @@ HDF5MetadataSource::OpenResult HDF5MetadataSource::Open(const std::string &filen
                 );
             if (mask_tmp.empty())
                 mask_tmp = std::vector<uint32_t>(image_size_x * image_size_y);
-            dataset->pixel_mask = PixelMask(mask_tmp);
+            dataset->pixel_mask = std::make_shared<const PixelMask>(mask_tmp);
         }
 
         ReadROIMetadata(*master_file, *dataset);

reader/JFJochHttpReader.cpp

@@ -19,6 +19,8 @@ void JFJochHttpReader::Close() {
     SetStartMessage({});
     last_image_buffer_counter = {};
     last_op_http_sync = false;
+    cached_pixel_mask.reset();
+    cached_pixel_mask_arm_date.clear();
 }
 
 ImageBufferStatus JFJochHttpReader::GetImageBufferStatus() const {
@@ -162,8 +164,17 @@ std::shared_ptr<JFJochReaderDataset> JFJochHttpReader::UpdateDataset_i() {
                 std::chrono::microseconds(std::lround(msg->start_message->count_time * 1e6))
         );
 
-        if (!msg->start_message->pixel_mask.empty())
-            dataset->pixel_mask = PixelMask(msg->start_message->pixel_mask.begin()->second);
+        if (!msg->start_message->pixel_mask.empty()) {
+            // The pixel mask is constant for an acquisition; build the full-detector (~tens of MB)
+            // PixelMask once per arm and share it across the per-refresh dataset snapshots, so a live
+            // dataset refresh does not reconstruct and copy it on every tick.
+            if (!cached_pixel_mask || cached_pixel_mask_arm_date != dataset->arm_date) {
+                cached_pixel_mask =
+                    std::make_shared<const PixelMask>(msg->start_message->pixel_mask.begin()->second);
+                cached_pixel_mask_arm_date = dataset->arm_date;
+            }
+            dataset->pixel_mask = cached_pixel_mask;
+        }
 
         dataset->experiment.NumTriggers(1);
         dataset->experiment.ImagesPerTrigger(msg->start_message->number_of_images);

reader/JFJochHttpReader.h

@@ -15,6 +15,11 @@ class JFJochHttpReader : public JFJochReader {
     std::optional<int64_t> last_image_buffer_counter;
     bool last_op_http_sync = false;
 
+    // Cache of the (constant-per-acquisition) pixel mask, keyed on arm date, so the per-refresh
+    // dataset rebuild reuses one shared mask instead of reconstructing it every tick.
+    std::shared_ptr<const PixelMask> cached_pixel_mask;
+    std::string cached_pixel_mask_arm_date;
+
     ImageBufferStatus GetImageBufferStatus() const;
 
     bool LoadImage_i(std::shared_ptr<JFJochReaderDataset> &dataset,

reader/JFJochReader.cpp

@@ -95,7 +95,10 @@ void JFJochReader::UpdateUserMask(const std::vector<uint32_t> &mask) {
     if (!dataset)
         return;
     auto new_dataset = std::make_shared<JFJochReaderDataset>(*dataset);
-    new_dataset->pixel_mask.LoadUserMask(dataset->experiment, mask);
+    // Copy-on-write: the mask is shared with the old snapshot, so edit a fresh copy, not in place.
+    auto new_mask = std::make_shared<PixelMask>(*dataset->pixel_mask);
+    new_mask->LoadUserMask(dataset->experiment, mask);
+    new_dataset->pixel_mask = new_mask;
     dataset = new_dataset;
 }
 

reader/JFJochReaderDataset.h

@@ -7,6 +7,7 @@
 #include <optional>
 #include <vector>
 #include <map>
+#include <memory>
 
 #include "../common/DiffractionGeometry.h"
 #include "../common/DiffractionExperiment.h"
@@ -17,7 +18,10 @@ struct JFJochReaderDataset {
     std::string arm_date;
 
     DiffractionExperiment experiment;
-    PixelMask pixel_mask;
+    // Shared, not copied, across dataset snapshots: the mask is constant for a run, so a per-frame
+    // dataset refresh / mutable copy must not clone the full-detector (~tens of MB) mask. Held as
+    // shared_ptr<const>; the rare edit (user mask) builds a fresh mask (copy-on-write). Never null.
+    std::shared_ptr<const PixelMask> pixel_mask = std::make_shared<const PixelMask>();
 
     std::optional<int64_t> error_value;
 

reader/JFJochReaderImage.cpp

@@ -99,7 +99,7 @@ void JFJochReaderImage::ProcessInputImage(const void *data, size_t npixel, int64
     top_pixels.reserve(top_pixels_acc.Capacity());
 
     bool has_input_mask = false;
-    const auto &mask = dataset->pixel_mask.GetMask();
+    const auto &mask = dataset->pixel_mask->GetMask();
     if (mask.size() == npixel)
         has_input_mask = true;