fix(v1_compat): stop silently neutralizing check_function — wrap it like process_function, and log swallowed check exceptions

daisy-py/python/daisy/v1_compat.py

@@ -226,6 +226,15 @@ def __new__(
             kwargs["process_function"] = _wrap_block_fn(kwargs["process_function"])
         elif len(args) >= 5 and args[4] is not None:
             args = (*args[:4], _wrap_block_fn(args[4]), *args[5:])
+        # check_function gets the same compat view: v1.x checks do funlib
+        # arithmetic on the block's ROIs just like process functions, and
+        # the Rust precheck treats any exception as "not done" — so an
+        # unwrapped check raises TypeError on the native block and every
+        # done block silently re-runs on resume.
+        if "check_function" in kwargs:
+            kwargs["check_function"] = _wrap_block_fn(kwargs["check_function"])
+        elif len(args) >= 6 and args[5] is not None:
+            args = (*args[:5], _wrap_block_fn(args[5]), *args[6:])
         if num_workers is not None and max_workers is not None:
             raise TypeError(
                 "pass either max_workers (v2) or num_workers (v1.x), not both"

daisy-py/src/py_callbacks.rs

@@ -33,6 +33,7 @@ pub(crate) fn cap_traceback(s: &str) -> String {
 use pyo3::types::PyDict;
 use std::cell::RefCell;
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, Ordering};
 use crate::py_block::PyBlock;
 use crate::py_task_state::PyTaskState;
 
@@ -59,22 +60,71 @@ pub fn take_last_process_pyerr() -> Option<PyErr> {
 /// Acquires the GIL on each call to invoke the Python function.
 pub struct PyCheckBlock {
     py_fn: Py<PyAny>,
+    /// One warning per task: a broken check raises the same way for
+    /// every block, and one traceback says everything N copies would.
+    /// (`PyCheckBlock` is constructed once per task in `py_task.rs`.)
+    warned: AtomicBool,
 }
 
 impl PyCheckBlock {
     pub fn new(py_fn: Py<PyAny>) -> Self {
-        Self { py_fn }
+        Self {
+            py_fn,
+            warned: AtomicBool::new(false),
+        }
+    }
+
+    /// Route a `check_function` exception to the `daisy` Python logger.
+    /// The scheduler treats a raising check as "not done", so the block
+    /// (re)runs — safe, but it must not be silent: before this warning, a
+    /// check that raised on every block quietly re-ran entire resumed
+    /// runs. Logging goes through Python's `logging` (the extension has
+    /// no tracing subscriber; logging integration is the Python-side
+    /// carve-out) and is best-effort like `PyProgressObserver`: a busted
+    /// logger must not break the run loop.
+    fn warn_check_failed(&self, py: Python<'_>, block: &Block, e: &PyErr) {
+        if self.warned.swap(true, Ordering::Relaxed) {
+            return;
+        }
+        // include the formatted python traceback (capped), as for block
+        // functions — see PyProcessBlock::process
+        let tb = e
+            .traceback(py)
+            .and_then(|t| t.format().ok())
+            .unwrap_or_default();
+        let msg = format!(
+            "check_function for task {:?} raised; treating block {} as not \
+             done, so it will be (re)processed. Further check failures for \
+             this task will not be logged:\n{}",
+            block.task_id(),
+            block.block_id,
+            cap_traceback(&format!("{tb}{e}")),
+        );
+        (|| -> PyResult<()> {
+            let logging = py.import("logging")?;
+            let logger = logging.call_method1("getLogger", ("daisy",))?;
+            logger.call_method1("warning", (msg,))?;
+            Ok(())
+        })()
+        .ok();
     }
 }
 
 impl CheckBlock for PyCheckBlock {
     fn check(&self, block: &Block) -> bool {
         Python::attach(|py| {
             let py_block = PyBlock::from_core(block.clone());
-            self.py_fn
+            match self
+                .py_fn
                 .call1(py, (py_block,))
                 .and_then(|r: Py<PyAny>| r.extract::<bool>(py))
-                .unwrap_or(false)
+            {
+                Ok(done) => done,
+                Err(e) => {
+                    self.warn_check_failed(py, block, &e);
+                    false
+                }
+            }
         })
     }
 }

tests/test_check_function_warning.py

@@ -2,9 +2,8 @@
 
 import warnings
 
-import pytest
-
 import daisy
+import pytest
 
 
 def _mk(**kw):
@@ -29,3 +28,41 @@ def test_no_check_function_no_warning():
     with warnings.catch_warnings():
         warnings.simplefilter("error")
         _mk()
+
+
+def test_raising_check_function_is_logged_not_swallowed(caplog):
+    """A raising check_function means "not done", so the block (re)runs —
+    but that must not be silent: an always-raising check re-runs every done
+    block on resume. The exception is logged as a WARNING on the `daisy`
+    logger, once per task (not once per block)."""
+    import logging
+
+    def broken_check(block):
+        raise RuntimeError("simulated broken check")
+
+    calls = []
+    task = daisy.Task(
+        task_id="warn_demo",
+        total_roi=daisy.Roi([0], [20]),
+        read_roi=daisy.Roi([0], [10]),
+        write_roi=daisy.Roi([0], [10]),
+        process_function=lambda b: calls.append(b.block_id),
+        check_function=broken_check,
+        read_write_conflict=False,
+        max_workers=1,
+    )
+    with caplog.at_level(logging.WARNING, logger="daisy"):
+        states = daisy.run_blockwise(
+            task, multiprocessing=False, progress=False, return_states=True
+        )
+    # every exception is "not done": both blocks still process
+    assert len(calls) == 2
+    assert states["warn_demo"].completed_count == 2
+    assert states["warn_demo"].skipped_count == 0
+    logged = [
+        r
+        for r in caplog.records
+        if "check_function" in r.getMessage() and "raised" in r.getMessage()
+    ]
+    assert len(logged) == 1, [r.getMessage() for r in logged]
+    assert "simulated broken check" in logged[0].getMessage()

tests/test_funlib_interop.py

@@ -83,6 +83,66 @@ def process(block):
     assert seen == {"read_is_funlib": True, "write_is_funlib": True, "eq": True}
 
 
+@pytest.mark.parametrize("style", ["kwargs", "positional"])
+def test_compat_check_fn_receives_funlib_rois_and_skips_done_blocks(tmp_path, style):
+    """v1.x check functions do funlib arithmetic on the block's ROIs — the
+    same idiom as process functions (volara's resume path). They must get
+    the same compat `Block` view: an unwrapped check raises TypeError on
+    the native block, the Rust precheck swallows every exception into
+    "not done", and every done block silently re-runs on resume."""
+    import daisy.v1_compat as compat
+
+    def marker(block):
+        # funlib Coordinate arithmetic on the offset — raises TypeError on
+        # a native block (daisy._daisy.Coordinate has no arithmetic dunders)
+        end = block.write_roi.offset + block.write_roi.shape
+        return tmp_path / ("done_" + "_".join(str(c) for c in end))
+
+    def process(block):
+        marker(block).touch()
+
+    def check(block):
+        return marker(block).exists()
+
+    def make_task():
+        if style == "kwargs":
+            return compat.Task(
+                "compat-check",
+                total_roi=fg.Roi((0,), (40,)),
+                read_roi=fg.Roi((0,), (10,)),
+                write_roi=fg.Roi((0,), (10,)),
+                process_function=process,
+                check_function=check,
+                num_workers=1,
+                tracking_path=False,
+            )
+        # the v1 positional signature: (task_id, total_roi, read_roi,
+        # write_roi, process_function, check_function, ...)
+        return compat.Task(
+            "compat-check",
+            d2.Roi((0,), (40,)),
+            d2.Roi((0,), (10,)),
+            d2.Roi((0,), (10,)),
+            process,
+            check,
+            num_workers=1,
+            tracking_path=False,
+        )
+
+    # run 1: no markers yet — every block processes and writes its marker
+    states = compat.run_blockwise(
+        [make_task()], multiprocessing=False, progress=False, return_states=True
+    )
+    assert states["compat-check"].completed_count == 4
+    assert states["compat-check"].skipped_count == 0
+
+    # run 2 (the resume): the check sees every marker → all blocks skipped
+    states = compat.run_blockwise(
+        [make_task()], multiprocessing=False, progress=False, return_states=True
+    )
+    assert states["compat-check"].skipped_count == 4
+
+
 def test_compat_spawn_fn_not_wrapped():
     """0-arg spawn functions must pass through unwrapped (arity preserved)."""
     from daisy.v1_compat import _wrap_block_fn