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"); }