From da5acca2946edb7237291acea1e88e93aca609a8 Mon Sep 17 00:00:00 2001 From: Jeff Rhoades Date: Tue, 11 Aug 2026 20:13:49 +0000 Subject: [PATCH 1/2] fix(v1_compat): wrap check_function like process_function MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task.__new__ on the compat surface wrapped process_function via _wrap_block_fn (native block -> funlib-typed compat Block) but never check_function. A v1-style check that does funlib arithmetic on block.write_roi.offset therefore got a raw native block and raised TypeError — and PyCheckBlock::check turns every Python exception into 'not done' (.unwrap_or(false)), so every done block silently re-ran on resume. Wrap check_function identically, in both the kwarg and the positional (arg index 5) forms. Test: a compat task whose check_function does funlib Coordinate arithmetic on the block's write_roi, with a per-block marker file as the done state; run_blockwise twice must skip all blocks on run 2 (skipped_count == 4). Fails on v2.0 (skipped_count == 0: every block re-ran), passes with the fix. Parametrized over kwarg and positional construction. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01ELENZ8uo6Pc1qshkep6iXH --- daisy-py/python/daisy/v1_compat.py | 9 +++++ tests/test_funlib_interop.py | 60 ++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+) diff --git a/daisy-py/python/daisy/v1_compat.py b/daisy-py/python/daisy/v1_compat.py index 54940a96..b65efdfd 100644 --- a/daisy-py/python/daisy/v1_compat.py +++ b/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" diff --git a/tests/test_funlib_interop.py b/tests/test_funlib_interop.py index ebe261af..e5949c5c 100644 --- a/tests/test_funlib_interop.py +++ b/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 From 9af4f1fa23e34f482026196a387a1631e5f2cf88 Mon Sep 17 00:00:00 2001 From: Jeff Rhoades Date: Tue, 11 Aug 2026 20:18:43 +0000 Subject: [PATCH 2/2] fix(py): log swallowed check_function exceptions instead of hiding them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PyCheckBlock::check turned every Python exception into 'not done' via a bare .unwrap_or(false): a raising check re-ran every done block on resume with no trace anywhere. Keep the fail-open semantics (a broken check must not kill the run) but log the exception as a WARNING on the 'daisy' Python logger with the capped formatted traceback (same formatting as PyProcessBlock::process / PySpawnWorker's wrap_err), best-effort like PyProgressObserver. Once per task via an AtomicBool on the struct — PyCheckBlock is built once per task in py_task.rs, and a broken check raises identically for every block. Python logging rather than tracing because the extension installs no tracing subscriber; logging integration is the Python-side carve-out. Test: an always-raising check_function still processes every block, and exactly one WARNING carrying the exception text lands on the 'daisy' logger. (Import sort in the touched test file fixed to make it lint-clean; it was one of the repo's pre-existing I001s.) Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01ELENZ8uo6Pc1qshkep6iXH --- daisy-py/src/py_callbacks.rs | 56 ++++++++++++++++++++++++++-- tests/test_check_function_warning.py | 41 +++++++++++++++++++- 2 files changed, 92 insertions(+), 5 deletions(-) diff --git a/daisy-py/src/py_callbacks.rs b/daisy-py/src/py_callbacks.rs index 8ced2080..def6becb 100644 --- a/daisy-py/src/py_callbacks.rs +++ b/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,11 +60,53 @@ pub fn take_last_process_pyerr() -> Option { /// Acquires the GIL on each call to invoke the Python function. pub struct PyCheckBlock { py_fn: Py, + /// 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) -> 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(); } } @@ -71,10 +114,17 @@ 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| r.extract::(py)) - .unwrap_or(false) + { + Ok(done) => done, + Err(e) => { + self.warn_check_failed(py, block, &e); + false + } + } }) } } diff --git a/tests/test_check_function_warning.py b/tests/test_check_function_warning.py index 8d8eac94..8610698a 100644 --- a/tests/test_check_function_warning.py +++ b/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()