mx/Jungfraujoch
3
2
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases 81 Wiki Activity

Writer: a failure that is not a JFJochException ends the run in error state too

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
leonarski_f
2026-10-13 12:27:08 +02:00
co-authored by Claude Fable 5.1
parent 23da53d2c1
commit 5f6f257998
2 changed files with 63 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/StreamWriterTest.cpp
+50
View File
@@ -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<CBORStream2DeserializerOutput>(start_message)});
std::vector<int16_t> 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<CBORStream2DeserializerOutput>(message)});
}
EndMessage end_message{};
end_message.max_image_number = 2;
puller.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(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");
}