Fix five silent-failure bugs in the streaming pipeline

scripts/laue_image_server.py

@@ -232,11 +232,18 @@ def _settled(path):
     send_error: list = []  # shared error flag
 
     frame_mapping: dict = {}
+    # frame_mapping is written by BOTH the sender thread and the consumer thread, and
+    # serialised to JSON by the consumer every save_interval frames. Without a lock,
+    # json.dump() iterates the dict while the other thread inserts into it and raises
+    # "RuntimeError: dictionary changed size during iteration". That killed the consumer
+    # thread mid-run, after which the server reported a normal "Done: N sent, 0 skipped"
+    # having silently processed only part of the scan (observed: 15,955 of 40,401).
+    mapping_lock = threading.Lock()
     sent_count_lock = threading.Lock()
     counters = {"sent": 0, "skip": 0}
 
     def _sender_thread(sock, send_q, send_error, frame_mapping, counters,
-                        labels_h5f):
+                        labels_h5f, mapping_lock):
         """Drain the queue and send frames over TCP."""
         while True:
             item = send_q.get()
@@ -265,7 +272,8 @@ def _sender_thread(sock, send_q, send_error, frame_mapping, counters,
                 with sent_count_lock:
                     counters["skip"] += 1
                 send_error.append(e)
-            frame_mapping[str(image_num)] = mapping_entry
+            with mapping_lock:
+                frame_mapping[str(image_num)] = mapping_entry
             send_q.task_done()
 
     # Open labels HDF5 for writing
@@ -278,7 +286,8 @@ def _sender_thread(sock, send_q, send_error, frame_mapping, counters,
 
     sender = threading.Thread(
         target=_sender_thread,
-        args=(sock, send_q, send_error, frame_mapping, counters, labels_h5f),
+        args=(sock, send_q, send_error, frame_mapping, counters, labels_h5f,
+              mapping_lock),
         daemon=True,
     )
     sender.start()
@@ -368,10 +377,11 @@ def _consumer_thread():
                         reason = result["error"]
                         if reason != "no_spots":
                             logger.error(f"  Frame {fr_idx} of {h5bn}: {reason}")
-                        frame_mapping[str(img_num)] = {
-                            "file": h5bn, "frame": fr_idx,
-                            "skipped": True, "reason": reason,
-                        }
+                        with mapping_lock:
+                            frame_mapping[str(img_num)] = {
+                                "file": h5bn, "frame": fr_idx,
+                                "skipped": True, "reason": reason,
+                            }
                         skip_count += 1
                     else:
                         mapping_entry = {
@@ -403,7 +413,15 @@ def _consumer_thread():
                                 f"({rate:.1f} img/s, ETA {remaining:.0f}s)"
                             )
                     if processed > 0 and processed % save_interval == 0:
-                        lsu.save_frame_mapping(frame_mapping, mapping_file)
+                        with mapping_lock:
+                            snapshot = dict(frame_mapping)
+                        try:
+                            lsu.save_frame_mapping(snapshot, mapping_file)
+                        except Exception as e:
+                            # a periodic save is only for crash resilience; the final
+                            # save after the producer finishes is the authoritative one.
+                            # Never let it take down the consumer thread.
+                            logger.warning(f"  periodic mapping save failed: {e}")
 
                 # Signal sender thread: no more frames
                 send_q.put(None)
@@ -424,8 +442,10 @@ def _consumer_thread():
         send_q.put(None)  # sentinel
         sender.join(timeout=60)
 
-        # Final mapping save
-        lsu.save_frame_mapping(frame_mapping, mapping_file)
+        # Final mapping save (authoritative)
+        with mapping_lock:
+            snapshot = dict(frame_mapping)
+        lsu.save_frame_mapping(snapshot, mapping_file)
         logger.info(f"Final mapping saved to {mapping_file}")
 
         # Close socket

scripts/laue_orchestrator.py

@@ -84,6 +84,67 @@ def _find_daemon_binary() -> str:
     )
 
 
+def _wait_for_daemon_port(
+    proc: subprocess.Popen,
+    port: int,
+    timeout: float,
+    poll_interval: float = 2.0,
+    progress_every: float = 30.0,
+) -> bool:
+    """Wait for the daemon to open *port*, watching the process while we wait.
+
+    ``lsu.wait_for_port()`` is process-blind, which fails in both directions:
+
+    * A daemon that dies immediately (bad params, missing file, GPU error) still
+      holds the orchestrator for the whole timeout before it reports anything.
+    * A daemon that is merely *slow* gets killed. Startup reads a multi-GB
+      orientation database and then initialises a CUDA context; on a cold page
+      cache, a busy GPU, or a loaded machine that legitimately exceeds a short
+      fixed budget, and the run aborts with "Daemon did not open port in time"
+      even though nothing is wrong.
+
+    Polling ``proc`` as well as the port fixes both: we fail fast with the exit
+    code when the daemon is genuinely dead, and we keep waiting (logging
+    progress) as long as it is alive.
+
+    Returns True once the port is open, False if the daemon died or the timeout
+    elapsed while it was still running.
+    """
+    t0 = time.time()
+    last_note = t0
+    while True:
+        if lsu.is_port_open("127.0.0.1", port):
+            logger.info(f"Port {port} ready ({time.time() - t0:.1f}s)")
+            return True
+
+        rc = proc.poll()
+        if rc is not None:
+            logger.error(
+                f"Daemon exited (code {rc}) after {time.time() - t0:.1f}s "
+                f"without opening port {port}. Check daemon log."
+            )
+            return False
+
+        now = time.time()
+        if now - t0 >= timeout:
+            logger.error(
+                f"Daemon is still running but has not opened port {port} after "
+                f"{timeout:.0f}s; giving up. If startup is legitimately this slow "
+                f"(very large orientation database, contended GPU), raise "
+                f"--port-timeout."
+            )
+            return False
+
+        if now - last_note >= progress_every:
+            logger.info(
+                f"  still waiting for port {port} "
+                f"({now - t0:.0f}s elapsed, daemon alive)..."
+            )
+            last_note = now
+
+        time.sleep(poll_interval)
+
+
 def _terminate_process(proc: subprocess.Popen, name: str, timeout: float = 10.0) -> None:
     """Send SIGTERM, wait, then SIGKILL if necessary."""
     if proc.poll() is not None:
@@ -162,7 +223,7 @@ def run_pipeline(
     ncpus: int = 1,
     output_dir: str = "",
     port: int = lsu.LAUE_STREAM_PORT,
-    port_timeout: float = 180.0,
+    port_timeout: float = 900.0,
     flush_time: float = 5.0,
     min_unique: int = 2,
     write_indexfile: bool = True,
@@ -185,7 +246,8 @@ def run_pipeline(
         ncpus:        Number of CPUs (passed to daemon).
         output_dir:   Output directory (auto-generated if empty).
         port:         Daemon TCP port.
-        port_timeout: Max seconds to wait for daemon port.
+        port_timeout: Max seconds to wait for the daemon port while the
+                      daemon is still alive (a dead daemon aborts immediately).
         flush_time:   Seconds to wait after server finishes before killing daemon.
         min_unique:   Minimum unique spots for orientation filtering.
     """
@@ -308,8 +370,7 @@ def run_pipeline(
 
     # --- 3. Wait for daemon port ---
     logger.info(f"Waiting for port {port}...")
-    if not lsu.wait_for_port("127.0.0.1", port, timeout=port_timeout):
-        logger.error("Daemon did not open port in time. Check daemon log.")
+    if not _wait_for_daemon_port(daemon_proc, port, port_timeout):
         _terminate_process(daemon_proc, "daemon")
         daemon_logf.close()
         _print_log_tail(daemon_log)
@@ -369,19 +430,35 @@ def run_pipeline(
     logger.info(f"Image server exited (code {server_proc.returncode}). "
                 f"Waiting for daemon to write results...")
 
-    # Poll for solutions.txt (mirrors integrator_batch_process.py pattern)
-    flush_deadline = time.time() + flush_time + 60
+    # Wait for the daemon to finish WRITING, not merely to have started.
+    #
+    # solutions.txt is appended to as each frame is fitted, so "the file exists and
+    # is non-empty" means the *first* frame landed, not the last. Breaking there and
+    # SIGTERMing the daemon two seconds later silently discards everything still in
+    # its queue — and the daemon is routinely behind (it logs "receive queue was full
+    # N times (backpressure from processing)"). Observed: a 6561-frame batch run lost
+    # the last 31 frames, contiguously, with no error anywhere.
+    #
+    # Instead, wait until the file stops growing: that is the daemon actually draining.
+    quiet_needed = max(flush_time, 10.0)
+    flush_deadline = time.time() + flush_time + 3600
+    last_size, last_change = -1, time.time()
     while time.time() < flush_deadline:
-        if os.path.isfile(solutions_file) and os.path.getsize(solutions_file) > 0:
-            logger.info(f"solutions.txt detected ({os.path.getsize(solutions_file)} bytes)")
-            time.sleep(2.0)  # extra wait for file to be fully flushed
-            break
         if daemon_proc.poll() is not None:
             logger.info("Daemon exited on its own.")
             break
+        size = os.path.getsize(solutions_file) if os.path.isfile(solutions_file) else 0
+        now = time.time()
+        if size != last_size:
+            last_size, last_change = size, now
+        elif size > 0 and (now - last_change) >= quiet_needed:
+            logger.info(f"solutions.txt quiescent at {size} bytes after "
+                        f"{quiet_needed:.0f}s without growth — daemon has drained")
+            break
         time.sleep(1.0)
     else:
-        logger.warning("Timed out waiting for daemon to write solutions.txt")
+        logger.warning("Timed out waiting for the daemon to finish writing "
+                       "solutions.txt; results may be truncated")
 
     # --- 7. Terminate daemon ---
     _terminate_process(daemon_proc, "daemon")
@@ -570,7 +647,7 @@ def main() -> None:
         help=f"Daemon TCP port (default: {lsu.LAUE_STREAM_PORT})"
     )
     parser.add_argument(
-        "--port-timeout", type=float, default=180.0,
+        "--port-timeout", type=float, default=900.0,
         help="Max seconds to wait for daemon port (default: 180)"
     )
     parser.add_argument(

scripts/pipeline/run_laue.sh

@@ -26,7 +26,10 @@ BETA_CONFIG=${BETA_CONFIG:-"$WORK/params/params_beta.txt"}  # set BETA_CONFIG=""
 ALPHA_GPU=${ALPHA_GPU:-0};  ALPHA_PORT=${ALPHA_PORT:-60517}
 BETA_GPU=${BETA_GPU:-1};    BETA_PORT=${BETA_PORT:-60518}
 NCPUS=${NCPUS:-32}                                   # CPU cores for the refinement stage
-WATCH=${WATCH:-"--watch"}                            # set WATCH="" to batch an existing folder
+# NB: "-" not ":-" — ${WATCH:-...} would substitute the default for an *empty*
+# WATCH too, so the documented WATCH="" (batch an existing folder) never took
+# effect and every run silently stayed in watch mode.
+WATCH=${WATCH-"--watch"}                             # set WATCH="" to batch an existing folder
 # -----------------------------------------------------------------------------
 
 FOLDER=${1:?usage: run_laue.sh DATA_FOLDER [H5_LOCATION]}
@@ -40,7 +43,13 @@ launch() {   # launch PHASE CONFIG GPU PORT
   [ -z "$config" ] && return 0
   [ -f "$config" ] || { echo "ERROR: config not found: $config" >&2; exit 1; }
   local out="$WORK/results/${phase}_$TS"
-  CUDA_VISIBLE_DEVICES=$gpu setsid nohup "$PY" "$SCRIPTS/laue_orchestrator.py" \
+  # CUDA_DEVICE_ORDER=PCI_BUS_ID is required for $gpu to mean the same card that
+  # nvidia-smi calls $gpu. CUDA defaults to FASTEST_FIRST, so on a mixed-GPU host
+  # "GPU 0" can land on a completely different physical card -- observed on a shared
+  # machine where the daemon landed on another user's GPU while the intended cards
+  # sat idle.
+  CUDA_DEVICE_ORDER=PCI_BUS_ID CUDA_VISIBLE_DEVICES=$gpu \
+      setsid nohup "$PY" "$SCRIPTS/laue_orchestrator.py" \
       --config "$config" --folder "$FOLDER" --h5-location "$H5LOC" \
       --ncpus "$NCPUS" --port "$port" $WATCH --output-dir "$out" \
       > "$WORK/results/${phase}_$TS.launch.log" 2>&1 < /dev/null &