writer: detect overwrite conflict at start for back-channel transports

Previously an existing output file was only detected at the final rename,
after the whole collection had run. Add an up-front check in the FileWriter
constructor that fails before acquisition arms.

Only the master file is checked, and only by the single writer that owns it
(write_master_file). In a multi-writer TCP/ZMQ setup the per-image data files
are staggered across writers by file number, and a writer does not even know
the total writer count, so it cannot enumerate its own subset; checking all
data files in every writer would make each writer stat files it never writes
and race the writers already creating them. Data-file conflicts stay caught by
their owning writer at the final rename.

Gate the check on whether the transport can report the failure to the broker:
the direct HDF5 pusher throws in-process and the TCP writer returns a
START-failure ACK, so both abort the collection cleanly (verified: an asymmetric
rejection cancels the already-started sibling). The ZeroMQ path is fire-and-forget
with no back-channel, so StreamWriter opts out via image_puller.SupportsAck() and
keeps writing .tmp files, failing only at the final rename (leaving the images on
disk). Documented in docs/JFJOCH_WRITER.md.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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 `.<random>.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:
 

tests/HDF5WritingTest.cpp

@@ -4,6 +4,7 @@
 #include <catch2/catch_all.hpp>
 #include <iostream>
 #include <fstream>
+#include <filesystem>
 
 #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);
 }

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<std::unique_ptr<TCPImagePuller> > puller;
+    for (int i = 0; i < npullers; i++)
+        puller.push_back(std::make_unique<TCPImagePuller>(pusher.GetAddress()[0], 8 * 1024 * 1024));
+
+    for (int attempt = 0; attempt < 100 && pusher.GetConnectedWriters() < static_cast<size_t>(npullers); ++attempt)
+        std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    REQUIRE(pusher.GetConnectedWriters() == static_cast<size_t>(npullers));
+
+    std::atomic<bool> start_threw{false};
+    std::atomic<bool> 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<std::thread> receivers;
+    receivers.reserve(npullers);
+    for (int w = 0; w < npullers; w++) {
+        receivers.emplace_back([&, w] {
+            std::optional<uint32_t> 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<TCPFrameType>(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;

writer/FileWriter.cpp

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: GPL-3.0-only
 
 #include "FileWriter.h"
+#include <filesystem>
 #include <nlohmann/json.hpp>
 #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<NXmx>(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<std::mutex> lock(hdf5_mutex);

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<HDF5DataFileStatistics>& 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);

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<FileWriter>(*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<FileWriter>(*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",