From 5f6f257998eefa8f6a6b667aeac06f384d1ab68c Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Tue, 13 Oct 2026 12:27:08 +0200 Subject: [PATCH] Writer: a failure that is not a JFJochException ends the run in error state too ProcessDataImage, ProcessEndMessage and NotifyReceiverOnFinalizedWrite caught only JFJochException, while ProcessStartMessage and FinalizeDataCollection already caught std::exception. Anything else - std::bad_alloc from a per-image array, filesystem_error from a path check, the JSON library refusing a string that is not UTF-8 while publishing a file-done notification - left CollectImages, and Run() logged "Exception not properly handled" and returned with the state never reaching Finalized: no END acknowledgement, no writer notification, the master's .tmp left behind until the next START. On the TCP image stream the broker then waited for the END acknowledgement indefinitely, because the writer's heartbeat thread keeps reporting it alive whatever state it is in. Now the same catch as at the other sites: error state, fatal ACK, END processed, Finalized. The new StreamWriter test provokes the JSON case through a sample name that is not UTF-8 with the file-done socket enabled, and checks the run ends Finalized with no master written. Co-Authored-By: Claude Fable 5.1 --- tests/StreamWriterTest.cpp | 50 ++++++++++++++++++++++++++++++++++++++ writer/StreamWriter.cpp | 16 +++++++++--- 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/tests/StreamWriterTest.cpp b/tests/StreamWriterTest.cpp index 662477ccf..1b767fb64 100644 --- a/tests/StreamWriterTest.cpp +++ b/tests/StreamWriterTest.cpp @@ -11,6 +11,7 @@ #include "../image_pusher/ZMQStream2Pusher.h" #include "../receiver/JFJochReceiverService.h" #include "../image_pusher/ZMQWriterNotificationPuller.h" +#include "../image_puller/TestImagePuller.h" TEST_CASE("StreamWriterTest_ZMQ", "[StreamWriter]") { RegisterHDF5Filter(); @@ -246,3 +247,52 @@ TEST_CASE("StreamWriterTest_ZMQ_MasterAfterAllDataFiles", "[StreamWriter]") { std::filesystem::remove_all("subdir4"); } + +// A failure that is not a JFJochException must end the run the same way as one that is: error state, +// END processed, state Finalized. Before this test the writer caught only its own exception type at the +// per-image and END sites, and anything else - here the JSON library refusing a sample name that is not +// UTF-8 while publishing a file-done notification - left Run() with the state stuck in Receiving, no END +// acknowledgement and no notification, and on TCP a broker waiting for that acknowledgement for ever. +TEST_CASE("StreamWriterTest_NonJFJochException_EndsRunInErrorState", "[StreamWriter]") { + RegisterHDF5Filter(); + Logger logger("StreamWriterTest_NonJFJochException"); + + DiffractionExperiment x(DetJF(2)); + x.FilePrefix("subdir5/StreamWriterTest5").NumTriggers(1).ImagesPerTrigger(2).ImagesPerFile(1) + .Compression(CompressionAlgorithm::NO_COMPRESSION).OverwriteExistingFiles(true) + .SampleName("\xff\xfe not utf-8"); + + StartMessage start_message; + x.FillMessage(start_message); + start_message.write_master_file = true; + + TestImagePuller puller; + // The file-done PUB socket is what makes the writer serialise the sample name to JSON. + StreamWriter writer(logger, puller, "ipc://*"); + + puller.Put({.cbor = std::make_shared(start_message)}); + std::vector image(x.GetPixelsNum()); + for (int i = 0; i < 2; i++) { + DataMessage message{}; + message.image = CompressedImage(image, x.GetXPixelsNum(), x.GetYPixelsNum()); + message.number = i; + puller.Put({.cbor = std::make_shared(message)}); + } + EndMessage end_message{}; + end_message.max_image_number = 2; + puller.Put({.cbor = std::make_shared(end_message)}); + + StreamWriterOutput out; + REQUIRE_NOTHROW(out = writer.Run()); + + // Image 0 filled its one-image data file; closing it published the notification, which threw. + CHECK(out.stats.state == StreamWriterState::Finalized); + CHECK(out.stats.processed_images == 0); + CHECK(out.data_file_stats.empty()); + // The master is never written for a run that ended in error; the data file was renamed into + // place before the notification was attempted. + CHECK(!std::filesystem::exists("subdir5/StreamWriterTest5_master.h5")); + CHECK(std::filesystem::exists("subdir5/StreamWriterTest5_data_000001.h5")); + + std::filesystem::remove_all("subdir5"); +} diff --git a/writer/StreamWriter.cpp b/writer/StreamWriter.cpp index af1bde36f..12bc6b8fa 100644 --- a/writer/StreamWriter.cpp +++ b/writer/StreamWriter.cpp @@ -178,7 +178,12 @@ void StreamWriter::ProcessDataImage() { if (verbose) logger.Info("Written"); NotifyTcpAck(TCPFrameType::DATA, true, false, TCPAckCode::None); - } catch (const JFJochException &e) { + } catch (const std::exception &e) { + // std::exception, not JFJochException, for the reason given in ProcessStartMessage: an + // exception that leaves this function skips the error state and the fatal ACK, and the + // broker then waits for an END ACK that never comes. Not every thrower below is ours - + // std::bad_alloc from a per-image array, filesystem_error from a path check, the JSON + // library when a file-done notification carries a string that is not UTF-8. logger.ErrorException(e); logger.Warning("Error writing image - switching to error state"); state = StreamWriterState::Error; @@ -206,7 +211,11 @@ void StreamWriter::ProcessEndMessage() { if ((image_puller_output.cbor->end_message->max_image_number == 0) && (max_image_number > 0)) image_puller_output.cbor->end_message->max_image_number = max_image_number; file_writer->WriteHDF5(*image_puller_output.cbor->end_message); - } catch (const JFJochException &e) { + } catch (const std::exception &e) { + // std::exception for the same reason as in ProcessDataImage. Here the cost of a leak is the + // highest: the END ACK and the notification below are never sent, the state never reaches + // Finalized, and on TCP the broker waits for the ACK for as long as the writer keeps + // heartbeating - which it does, whatever state it is in. logger.ErrorException(e); logger.Error("Error writing end message - switching to error state"); state = StreamWriterState::Error; @@ -411,7 +420,8 @@ void StreamWriter::NotifyReceiverOnFinalizedWrite(const std::string &detector_up s.SendTimeout(std::chrono::seconds(1)); s.Connect(detector_update_zmq_addr); s.Send(j.dump()); - } catch (const JFJochException &e) { + } catch (const std::exception &e) { + // std::exception: json::dump throws its own type when run_name is not valid UTF-8. logger.ErrorException(e); logger.Error("Error sending notification to detector update socket"); }