diff --git a/docs/JFJOCH_WRITER.md b/docs/JFJOCH_WRITER.md index 4cba4066..40b094e8 100644 --- a/docs/JFJOCH_WRITER.md +++ b/docs/JFJOCH_WRITER.md @@ -48,6 +48,27 @@ After file is all saved and closed, it is renamed to remove the suffix. By default, renaming won't happen if this would overwrite existing file. However, this behavior can be changed by setting `overwrite` parameter to true in the file writer configuration. +### When the overwrite conflict is reported +An existing output file is a fatal condition (unless `overwrite` is true). *When* it is detected +depends on whether the transport between the broker and the writer has a back-channel to report the +failure before acquisition starts: + +* **Direct HDF5 pusher and TCP writer (back-channel available).** The conflict is detected at + **start**: the writer that owns the master file checks whether it already exists and refuses to + start. The direct pusher raises the error in-process; the TCP writer returns a START-failure + acknowledgement. Either way the broker learns immediately and aborts the data collection *before* + the detector is armed — no images are taken and nothing is written. Only the master file is + checked up front: in a multi-writer setup the per-image data files are staggered across writers, + and checking them at start would make each writer inspect files it never writes (and race the + writers that do). Data-file conflicts are instead caught by their owning writer at the final + rename, which for the TCP path surfaces as a write-failure acknowledgement to the broker. +* **ZeroMQ writer (no back-channel).** The ZeroMQ image stream is fire-and-forget: the writer has no + way to tell the broker to stop, and the broker would keep streaming images regardless. The writer + therefore does **not** fail at start. It writes the whole series to the `..tmp` files as + usual and only fails at the final rename, leaving the `.tmp` files on disk. This is deliberate: the + acquired images are preserved (in `.tmp` form) rather than being dropped by a writer that aborted + mid-stream. Rename the `.tmp` files by hand, or re-run with `overwrite` set, to recover them. + ## Finalized files information Creates PUB socket to inform about finalized data files. For each closed file, the socket will send a JSON message, with the following structure: diff --git a/tests/HDF5WritingTest.cpp b/tests/HDF5WritingTest.cpp index 35af40a0..39a73a47 100644 --- a/tests/HDF5WritingTest.cpp +++ b/tests/HDF5WritingTest.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include "../common/DiffractionExperiment.h" #include "../writer/HDF5Objects.h" @@ -1445,37 +1446,73 @@ TEST_CASE("HDF5Objects_VDS_reverse_strided", "[HDF5][Unit]") { remove("scratch_vds_reverse_strided.h5"); REQUIRE(H5Fget_obj_count(H5F_OBJ_ALL, H5F_OBJ_ALL) == 0); } -// Reproduces the "pusher stuck after overwrite failure" bug: if EndDataCollection -// throws while finalizing (here: master file already exists, overwrite off), the -// pusher must still tear down its writer so a subsequent collection can start. -TEST_CASE("HDF5FilePusher_overwrite_failure_recovers", "[HDF5FilePusher][Repro]") { +// Overwrite is detected up front for back-channel transports (default) by checking +// the master file only - never the staggered per-writer data files. The ZeroMQ path +// (no back-channel) must opt out and keep writing .tmp files instead. +TEST_CASE("FileWriter_overwrite_detected_at_start", "[HDF5][Overwrite]") { RegisterHDF5Filter(); DiffractionExperiment x(DetJF4M()); - x.FilePrefix("pusher_overwrite_repro").ImagesPerTrigger(1) + x.FilePrefix("fw_overwrite_start").ImagesPerTrigger(3).ImagesPerFile(2) + .Compression(CompressionAlgorithm::NO_COMPRESSION) + .SetFileWriterFormat(FileWriterFormat::NXmxVDS).OverwriteExistingFiles(false); + + StartMessage start_message; + x.FillMessage(start_message); + REQUIRE(start_message.write_master_file.value_or(false)); // this writer owns the master + + // A stray data file (owned by another, staggered writer) must NOT trip the check - + // only the master file is inspected. The temporary writer cleans up its own tmp. + { std::ofstream(HDF5Metadata::DataFileName(start_message, 0)) << "blocker"; } + REQUIRE_NOTHROW(FileWriter(start_message)); + remove(HDF5Metadata::DataFileName(start_message, 0).c_str()); + + // The master file, on the other hand, does collide. + { std::ofstream(HDF5Metadata::MasterFileName(start_message)) << "blocker"; } + + // Back-channel transport (direct HDF5 / TCP): fail fast in the constructor. + REQUIRE_THROWS_AS(FileWriter(start_message), JFJochException); + + // ZeroMQ transport (no back-channel): must not throw - it will write .tmp and + // only fail at the final rename. The un-finalized writer cleans up its own tmp. + REQUIRE_NOTHROW(FileWriter(start_message, /*check_overwrite_at_start=*/false)); + + remove(HDF5Metadata::MasterFileName(start_message).c_str()); + REQUIRE(H5Fget_obj_count(H5F_OBJ_ALL, H5F_OBJ_ALL) == 0); +} + +// If EndDataCollection throws while finalizing (here the master file appears mid-run, +// so the early check can't catch it), the pusher must still tear down its writer so a +// subsequent collection can start instead of dying with "already writing images". +TEST_CASE("HDF5FilePusher_finalize_failure_recovers", "[HDF5FilePusher][Repro]") { + RegisterHDF5Filter(); + + DiffractionExperiment x(DetJF4M()); + x.FilePrefix("pusher_finalize_repro").ImagesPerTrigger(1) .Compression(CompressionAlgorithm::NO_COMPRESSION) .SetFileWriterFormat(FileWriterFormat::NXmxVDS) .OverwriteExistingFiles(false); StartMessage start_message; x.FillMessage(start_message); - EndMessage end_message{}; - // Make the master-file rename fail at EndDataCollection time. - { std::ofstream(HDF5Metadata::MasterFileName(start_message)) << "blocker"; } - HDF5FilePusher pusher; pusher.StartDataCollection(start_message); - // Finalization fails because the target master file already exists. + // Create the conflict after the start-time check has already passed. + { std::ofstream(HDF5Metadata::MasterFileName(start_message)) << "blocker"; } REQUIRE_THROWS_AS(pusher.EndDataCollection(end_message), JFJochException); - // The pusher must have released its writer despite the failure - otherwise the - // next collection dies with "Image pusher is already writing images". + // Writer released despite the failure: the next collection starts cleanly + // instead of dying with "already writing images". + remove(HDF5Metadata::MasterFileName(start_message).c_str()); REQUIRE_NOTHROW(pusher.StartDataCollection(start_message)); - REQUIRE_THROWS_AS(pusher.EndDataCollection(end_message), JFJochException); + REQUIRE_NOTHROW(pusher.EndDataCollection(end_message)); - remove("pusher_overwrite_repro_master.h5"); - remove(HDF5Metadata::DataFileName(start_message, 0).c_str()); + // The failed finalize intentionally leaves a .tmp behind - sweep the prefix. + for (const auto &e : std::filesystem::directory_iterator(".")) + if (e.path().filename().string().rfind("pusher_finalize_repro", 0) == 0) + std::filesystem::remove(e.path()); + REQUIRE(H5Fget_obj_count(H5F_OBJ_ALL, H5F_OBJ_ALL) == 0); } diff --git a/tests/TCPImagePusherTest.cpp b/tests/TCPImagePusherTest.cpp index 021b49ab..fcc2706e 100644 --- a/tests/TCPImagePusherTest.cpp +++ b/tests/TCPImagePusherTest.cpp @@ -167,6 +167,94 @@ TEST_CASE("TCPImageCommTest_2Writers_WithAck", "[TCP]") { p->Disconnect(); } +// One writer rejects START (as it would on an overwrite conflict) while its sibling +// accepts. The broker must abort the whole collection and cleanly cancel the sibling +// that already started - no half-armed collection, no stuck writer. +TEST_CASE("TCPImageCommTest_StartRejectedByOneWriter_AbortsCleanly", "[TCP]") { + const int64_t npullers = 2; + const int64_t images_per_file = 16; + + TCPStreamPusher pusher("tcp://127.0.0.1:*", npullers); + std::vector > puller; + for (int i = 0; i < npullers; i++) + puller.push_back(std::make_unique(pusher.GetAddress()[0], 8 * 1024 * 1024)); + + for (int attempt = 0; attempt < 100 && pusher.GetConnectedWriters() < static_cast(npullers); ++attempt) + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + REQUIRE(pusher.GetConnectedWriters() == static_cast(npullers)); + + std::atomic start_threw{false}; + std::atomic socket0_cancelled{false}; + + std::thread sender([&] { + StartMessage start{.images_per_file = images_per_file, .write_master_file = true}; + try { + pusher.StartDataCollection(start); + } catch (const JFJochException &) { + start_threw = true; + } + }); + + std::vector receivers; + receivers.reserve(npullers); + for (int w = 0; w < npullers; w++) { + receivers.emplace_back([&, w] { + std::optional my_socket; + for (int polls = 0; polls < 200; polls++) { + auto out = puller[w]->PollImage(std::chrono::milliseconds(100)); + if (!out.has_value()) { + if (my_socket.has_value() && *my_socket == 1) + break; // socket 1 is done once it has rejected + continue; + } + const auto &h = out->tcp_msg->header; + + if (static_cast(h.type) == TCPFrameType::CANCEL) { + PullerAckMessage ack; + ack.ack_for = TCPFrameType::CANCEL; + ack.ok = true; + ack.run_number = h.run_number; + ack.socket_number = h.socket_number; + ack.error_code = TCPAckCode::None; + puller[w]->SendAck(ack); + if (h.socket_number == 0) + socket0_cancelled = true; + break; + } + + if (out->cbor && out->cbor->start_message) { + my_socket = h.socket_number; + PullerAckMessage ack; + ack.ack_for = TCPFrameType::START; + ack.run_number = h.run_number; + ack.socket_number = h.socket_number; + if (h.socket_number == 1) { // this writer refuses (e.g. output exists) + ack.ok = false; + ack.fatal = true; + ack.error_code = TCPAckCode::StartFailed; + ack.error_text = "output file already exists"; + } else { // sibling starts fine + ack.ok = true; + ack.error_code = TCPAckCode::None; + } + puller[w]->SendAck(ack); + if (h.socket_number == 1) + break; + } + } + }); + } + + sender.join(); + for (auto &t: receivers) t.join(); + + REQUIRE(start_threw); // collection aborted + REQUIRE(socket0_cancelled); // the already-started sibling was cleanly cancelled + + for (auto &p: puller) + p->Disconnect(); +} + TEST_CASE("TCPImageCommTest_DataFatalAck_PropagatesToPusher", "[TCP]") { const size_t nframes = 64; const int64_t npullers = 2; diff --git a/writer/FileWriter.cpp b/writer/FileWriter.cpp index 77a6aa70..ff4ebf1e 100644 --- a/writer/FileWriter.cpp +++ b/writer/FileWriter.cpp @@ -2,6 +2,7 @@ // SPDX-License-Identifier: GPL-3.0-only #include "FileWriter.h" +#include #include #include "MakeDirectory.h" #include "../common/CheckPath.h" @@ -10,7 +11,7 @@ #include "../preview/JFJochTIFF.h" #include "JFJochDecompress.h" -FileWriter::FileWriter(const StartMessage &request) +FileWriter::FileWriter(const StartMessage &request, bool check_overwrite_at_start) : start_message(request) { if (start_message.file_format) format = start_message.file_format.value(); @@ -20,6 +21,8 @@ FileWriter::FileWriter(const StartMessage &request) CheckPath(start_message.file_prefix); MakeDirectory(start_message.file_prefix); + if (check_overwrite_at_start) + CheckOutputFilesAvailable(); if (start_message.write_master_file && start_message.write_master_file.value()) { switch (format) { case FileWriterFormat::NXmxLegacy: @@ -224,6 +227,27 @@ void FileWriter::CreateHDF5MasterFile(const StartMessage &msg) { master_file = std::make_unique(msg); } +void FileWriter::CheckOutputFilesAvailable() const { + if (start_message.overwrite.value_or(false)) + return; + + // Only the master file is checked, and only by the single writer that owns it + // (write_master_file - index 0 in a multi-writer TCP/ZMQ setup). Data files are + // staggered across writers by file number, so enumerating them here would make + // every writer stat files it never writes and race sibling writers that are + // already creating them; those conflicts are caught per-writer at finalize. + const bool nxmx = format == FileWriterFormat::NXmxLegacy + || format == FileWriterFormat::NXmxVDS + || format == FileWriterFormat::NXmxIntegrated; + + if (nxmx && start_message.write_master_file.value_or(false)) { + const std::string name = HDF5Metadata::MasterFileName(start_message); + if (std::filesystem::exists(name)) + throw JFJochException(JFJochExceptionCategory::FileWriteError, + "Output file already exists and overwrite is off: " + name); + } +} + void FileWriter::WriteHDF5(const CompressedImage &msg) { if (master_file) { std::lock_guard lock(hdf5_mutex); diff --git a/writer/FileWriter.h b/writer/FileWriter.h index d73c72f1..f4855921 100644 --- a/writer/FileWriter.h +++ b/writer/FileWriter.h @@ -26,12 +26,19 @@ class FileWriter { constexpr static uint64_t close_file_lag_images = 1000; constexpr static uint64_t default_images_per_file = 1000; void CreateHDF5MasterFile(const StartMessage& msg); + void CheckOutputFilesAvailable() const; void AddStats(const std::optional& s); void CloseFile(uint64_t file_number); void CloseOldFiles(uint64_t current_image_number); public: - explicit FileWriter(const StartMessage &request); + // check_overwrite_at_start: fail the constructor if an output file already + // exists and overwrite is off. Only safe for transports with a back-channel + // (direct HDF5 pusher, TCP) that can report the failure to the broker before + // acquisition arms. The ZeroMQ pusher is fire-and-forget with no back-channel, + // so it must pass false: the writer then falls back to writing .tmp files and + // failing at the final rename (see docs/JFJOCH_WRITER.md). + explicit FileWriter(const StartMessage &request, bool check_overwrite_at_start = true); void Write(const DataMessage& msg); void WriteTIFF(const DataMessage& msg); void WriteHDF5(const DataMessage& msg); diff --git a/writer/StreamWriter.cpp b/writer/StreamWriter.cpp index 9245ea50..438de0e9 100644 --- a/writer/StreamWriter.cpp +++ b/writer/StreamWriter.cpp @@ -66,7 +66,11 @@ void StreamWriter::ProcessStartMessage() { writer_notification_zmq_addr = image_puller_output.cbor->start_message->writer_notification_zmq_addr; try { - file_writer = std::make_unique(*image_puller_output.cbor->start_message); + // Fail fast on an overwrite conflict only when the transport can report it + // back to the broker (TCP ACK). The ZeroMQ path has no back-channel, so it + // keeps writing .tmp files and fails at the final rename instead. + file_writer = std::make_unique(*image_puller_output.cbor->start_message, + image_puller.SupportsAck()); if (!file_done_address.empty()) file_writer->SetupFinalizedFileSocket(file_done_address); logger.Info("Starting writing for dataset {} of {} images",