receiver: live stills report last image -> all written, and Done once written

The serial (FPGA + GPU) live path now times the latency from the last image
to the end message out and the writer finished, as the rotation receiver
does: StillsGPUAnalysis::Written, called by FinalizeMeasurement after the
writer is finalised, sets last_image_to_written_s and only then the run's
outcome. Until then the status shows a "Writing" step, so the run no longer
reads Done (and its elapsed time no longer stops) before the end message and
the writer are through. The timeline's latency bracket and the headline
latency now show for serial collections.

The END message is serialised into a scratch buffer first, as on rotation:
merged data that would not fit the END buffer are dropped with a warning
instead of failing the collection.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UqXYdL4qBGYRiSXSGXfCLu

receiver/JFJochReceiverFPGA.cpp

@@ -26,7 +26,7 @@ JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experimen
                                        ZMQMetadataSocket *in_zmq_metadata_socket,
                                        IndexerThreadPool *indexing_thread_pool,
                                        bool in_gpu_analysis,
-                                       RugnuxLiveStatus *rugnux_status)
+                                       RugnuxLiveStatus *in_rugnux_status)
     : JFJochReceiver(in_experiment,
                      in_send_buf_ctrl,
                      in_image_sender,
@@ -48,7 +48,8 @@ JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experimen
       frame_transformation_ready((experiment.GetImageNum() > 0) ? frame_transformation_nthreads : 0),
       data_acquisition_ready(ndatastreams),
       gpu_analysis(in_gpu_analysis && experiment.GetImageNum() > 0
-                   && experiment.GetDetectorMode() == DetectorMode::Standard) {
+                   && experiment.GetDetectorMode() == DetectorMode::Standard),
+      rugnux_status(in_rugnux_status) {
     // The FPGA integration core has a fixed bin-count limit. Reject configurations it cannot
     // handle before starting acquisition, unless azimuthal integration is forced onto the CPU.
     const auto &azint_settings = experiment.GetAzimuthalIntegrationSettings();
@@ -707,6 +708,9 @@ void JFJochReceiverFPGA::FinalizeMeasurement() {
 
     if (push_images_to_writer)
         writer_error = image_pusher.Finalize();
+    // Last image -> all written, as the rugnux receiver reports it
+    if (gpu)
+        gpu->Written();
 
     current_status.SetProgress({});
     current_status.SetStatus(GetStatus());
@@ -861,4 +865,21 @@ void JFJochReceiverFPGA::AddToEndMessage(EndMessage &message) {
     if (const auto &m = gpu->GetMerged(); m && gpu->GetMergeExperiment().GetUnitCell())
         message.merged = MergedMtzTable(m->merge.merged, *gpu->GetMergeExperiment().GetUnitCell(),
                                         gpu->GetMergeExperiment(), OutputProvenance{});
+    if (!message.merged)
+        return;
+    // Every pusher serialises END into a buffer of MESSAGE_SIZE_FOR_START_END, and an END that does not fit there
+    // fails the collection. So it is serialised here first, as JFJochReceiverRugnux does; one that does not fit
+    // goes without the merged data.
+    const auto scratch = std::make_unique_for_overwrite<uint8_t[]>(MESSAGE_SIZE_FOR_START_END);
+    CBORStream2Serializer serializer(scratch.get(), MESSAGE_SIZE_FOR_START_END);
+    try {
+        serializer.SerializeSequenceEnd(message);
+    } catch (const JFJochException &e) {
+        const std::string text = fmt::format("the merged data do not fit in the end message ({} MiB) and are not "
+                                             "written", MESSAGE_SIZE_FOR_START_END / (1024 * 1024));
+        logger.Error("GPU analysis: {} - {}", text, e.what());
+        if (rugnux_status)
+            rugnux_status->Warning(text);
+        message.merged.reset();
+    }
 }

receiver/JFJochReceiverFPGA.h

@@ -90,6 +90,7 @@ class JFJochReceiverFPGA : public JFJochReceiver  {
     // Serial data with rugnux on: the converted frames are analysed on the GPU (StillsGPUAnalysis), and
     // the start message waits for its pre-pass.
     const bool gpu_analysis;
+    RugnuxLiveStatus *const rugnux_status;
     std::unique_ptr<StillsGPUAnalysis> gpu;
     void GPUSink(DataMessage &message, const AzimuthalIntegrationProfile &profile);
     void AddToEndMessage(EndMessage &message) override;

receiver/StillsGPUAnalysis.cpp

@@ -44,6 +44,8 @@ StillsGPUAnalysis::~StillsGPUAnalysis() {
     } catch (const std::exception &e) {
         logger.Error("GPU analysis: {}", e.what());
     }
+    if (status && outcome)
+        status->Finish(outcome->first, outcome->second);
 }
 
 void StillsGPUAnalysis::AnalysisError(const std::string &text) {
@@ -108,7 +110,7 @@ void StillsGPUAnalysis::Finish() {
     for (auto &f: analysis_futures)
         f.get();
     if (status) {
-        status->LastImage();
+        last_image = status->LastImage();
         Publish();
     }
     if (merge_future.valid()) {
@@ -126,17 +128,28 @@ void StillsGPUAnalysis::Finish() {
         if (setup_error)
             status->Finish(RugnuxLiveState::Failed, "GPU analysis setup failed");
         else {
+            // Reported by Written, once everything is written - the end message and the writer go on meanwhile.
+            status->Step("Writing");
             std::unique_lock ul(error_mutex);
             if (analysis_error.empty())
-                status->Finish(RugnuxLiveState::Done);
+                outcome.emplace(RugnuxLiveState::Done, "");
             else
-                status->Finish(RugnuxLiveState::Failed, analysis_error);
+                outcome.emplace(RugnuxLiveState::Failed, analysis_error);
         }
     }
     if (setup_error)
         std::rethrow_exception(setup_error);
 }
 
+void StillsGPUAnalysis::Written() {
+    if (!status || !outcome)
+        return;
+    if (last_image)
+        status->Latency(std::chrono::duration<double>(std::chrono::steady_clock::now() - last_image.value()).count());
+    status->Finish(outcome->first, outcome->second);
+    outcome.reset();
+}
+
 void StillsGPUAnalysis::PrePassThread() {
     std::map<int64_t, std::shared_ptr<HeldImage>> sample;
     {

receiver/StillsGPUAnalysis.h

@@ -82,6 +82,9 @@ public:
     // No frame comes after this: the pre-pass runs on what came, if it has not, and every frame is
     // analysed before it returns. Throws where the setup after the pre-pass failed.
     void Finish();
+    // After Finish, once the end message is out and the writer is done: the status gets the latency from the last
+    // image and then the run's outcome. Without it the outcome is reported when the analysis goes, with no latency.
+    void Written();
 
     // After Finish.
     [[nodiscard]] std::optional<UnitCell> GetConsensusUnitCell() const;
@@ -109,6 +112,9 @@ private:
     RugnuxLiveStills live;
     std::chrono::steady_clock::time_point ready_time;
     void Publish();
+    // Kept by Finish for Written
+    std::optional<std::chrono::steady_clock::time_point> last_image;
+    std::optional<std::pair<RugnuxLiveState, std::string>> outcome;
 
     Setup setup;
     std::unique_ptr<IndexAndRefine> indexer;

tests/JFJochReceiverProcessingTest.cpp

@@ -182,6 +182,13 @@ TEST_CASE("JFJochIntegrationTest_ZMQ_crystal_spot_and_index_GPUStills", "[JFJoch
     CHECK(receiver_out.status.images_sent == experiment.GetImageNum());
     CHECK(!receiver_out.status.cancelled);
 
+    // Done once everything is written, with the latency from the last image to then
+    const auto live = service.GetRugnuxStatus();
+    CHECK(live.state == RugnuxLiveState::Done);
+    REQUIRE(live.last_image_to_written_s);
+    REQUIRE(live.last_image_s);
+    CHECK(live.elapsed_s >= live.last_image_s.value() + live.last_image_to_written_s.value() - 1e-6);
+
     // No progress value at the end of measurement
     REQUIRE(!service.GetProgress().has_value());
 

tests/StillsGPUAnalysisTest.cpp

@@ -110,7 +110,18 @@ TEST_CASE("StillsGPUAnalysis_FailingFrameDoesNotStopAnalysis", "[StillsGPUAnalys
         REQUIRE(gpu.Hold(Frame(buffer, i)));
     REQUIRE_NOTHROW(gpu.Finish());
     CHECK(sunk == 3);
-    CHECK(status.Get().state == RugnuxLiveState::Failed);
-    CHECK(!status.Get().message.empty());
+    // The outcome waits for everything to be written, and comes after the latency.
+    CHECK(status.Get().state == RugnuxLiveState::Running);
+    CHECK(!status.Get().last_image_to_written_s);
+    gpu.Written();
+    const auto s = status.Get();
+    CHECK(s.state == RugnuxLiveState::Failed);
+    CHECK(!s.message.empty());
+    REQUIRE(s.last_image_to_written_s);
+    REQUIRE(s.last_image_s);
+    CHECK(s.elapsed_s >= s.last_image_s.value() + s.last_image_to_written_s.value() - 1e-6);
+    REQUIRE(!s.steps.empty());
+    CHECK(s.steps.back().title == "Writing");
+    CHECK(s.steps.back().duration_s);
     CHECK(buffer.GetStatus().preparation_slots == 0);
 }