From 6c37f9f8bdebd9b9eb1d79e4c498fdf35e586990 Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Mon, 24 Aug 2026 23:58:09 -0700 Subject: [PATCH 01/15] fix: Stream issues --- pr312-followups.md | 108 +++++++++++++++++++++++++++++++++++++++++++++ src/c2pa/c2pa.py | 23 ++++++++-- 2 files changed, 128 insertions(+), 3 deletions(-) create mode 100644 pr312-followups.md diff --git a/pr312-followups.md b/pr312-followups.md new file mode 100644 index 00000000..9dceafee --- /dev/null +++ b/pr312-followups.md @@ -0,0 +1,108 @@ +# c2pa-python pull request 312: additional fixes + +Sentinel in the native thread-local error slot. Scratch note, not for the +repository. + +The change is correct as written. Three items, the first of which quietly +disables the diagnostic the sentinel exists to provide. + +--- + +## 1. Match on a substring, not on exact equality + +**The problem.** The comparison is + +```python +if error == ManagedResource._NO_NATIVE_ERROR.decode('utf-8'): +``` + +`c2pa_error_set_last` does not store the string verbatim. It runs it through +`Error::from` and then `CimplError::from`, and its own documentation states that +a missing or invalid error type is replaced with `Other` and the message includes +the original string. The sentinel already carries an `Other: ` prefix, so it may +round-trip unchanged, or it may come back re-prefixed or otherwise normalised. + +**Why it matters, and why it is easy to miss.** The failure is silent rather than +dangerous. If the comparison never matches, control falls through to the final +branch, which now performs the identical `_teardown(free_handle=False)`. Same +action, no crash, all tests that check behaviour still pass. + +What is lost is the distinct log line. That log line is the entire reason for +planting a sentinel rather than simply clearing the slot: it separates "the +native side reported nothing" from "the native side reported a real error", and +it is the only field evidence available for how often the ambiguous case occurs. +Losing it costs nothing today and costs the whole diagnostic tomorrow. + +**Fix.** Match on the distinctive part only: + +```python +_NO_NATIVE_ERROR_MARKER = "c2pa-python-no-native-error" +... +if _NO_NATIVE_ERROR_MARKER in error: +``` + +**And pin the round-trip in a test regardless**, since it is a property of the +native side that can change without notice: + +```python +def test_sentinel_round_trips_through_native_error_slot(self): + c2pa_module._lib.c2pa_error_set_last(ManagedResource._NO_NATIVE_ERROR) + self.assertIn(_NO_NATIVE_ERROR_MARKER, c2pa_module._read_native_error()) +``` + +That test fails loudly if the normalisation ever changes, which is exactly the +kind of upstream invariant worth pinning rather than assuming. + +## 2. Restore the ordering rationale that was deleted + +The removed paragraph explained that `c2pa_free` on a handle the registry no +longer tracks returns minus one and overwrites the slot with its own +untracked-pointer message, so the error must be read before any free or the +substitute carries a pre-consume tag and inverts the retain decision. + +That constraint is still true, and the current code still depends on it. The +replacement text explains the sentinel but says nothing about why the read comes +first. A later edit that moves the read after a free would reintroduce the +inversion with no warning anywhere in the file. + +One sentence is enough: + +> The read must precede any free: `c2pa_free` on an untracked handle overwrites +> the slot with its own untracked-pointer message, which carries a pre-consume +> tag and would invert the decision below. + +## 3. Confirm the changed test is green for the right reason + +`test_context_build_null_return_frees_builder` loses its explicit +`c2pa_error_set_last(b"UntrackedPointer: ...")` line. That test needs the +retained branch to fire, which needs a pre-consume tag present at the moment the +failure is read. + +With the sentinel now planted inside `_invoke_consume`, a mock that merely +returns `None` leaves the sentinel in place, the sentinel branch fires, +`_teardown(free_handle=False)` runs, and no free happens. The assertion should +then fail. + +Presumably the mock is now built with `_fail_with_native_error(b"UntrackedPointer: ...")`, +which restores the tag from inside the call rather than before it. Worth +confirming that is what landed, because a test that passes for the wrong reason +here is worse than one that fails: it would be asserting the retained branch +while actually exercising the consumed one. + +--- + +## What already holds + +The sentinel is planted immediately before `ffi_call`, inside `_invoke_consume`, +with nothing between them, on the thread that makes the call. That is the correct +placement and the thread-local slot means it cannot disturb any other worker. + +`_setup_function(_lib.c2pa_error_set_last, [ctypes.c_char_p], ctypes.c_int)` +supplies the explicit argument and return types, which was the one open check. +The return value itself needs no guard: minus one is returned only for a null +pointer, so any non-null sentinel returns zero. + +Changing the final fallback from `_release_handle()` to +`_teardown(free_handle=False)` removes the guarded free from the ambiguous path +entirely. That is the more important half of this pull request, and it holds even +if the sentinel comparison in item 1 never matches. diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 5f3dfa61..4ef1ee6d 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -2507,9 +2507,7 @@ def __init__( else: # format_or_path is a format string, stream is a stream object - with Stream(stream) as stream_obj: - self._create_reader( - format_bytes, stream_obj, manifest_data) + self._init_from_stream(stream, format_bytes, manifest_data) @staticmethod def _resolve_format_bytes(format_or_path, stream) -> Optional[bytes]: @@ -2580,6 +2578,25 @@ def _init_from_file(self, path, format_bytes, raise C2paError.Io( Reader._ERROR_MESSAGES['io_error'].format(str(e))) + def _init_from_stream(self, stream, format_bytes, + manifest_data=None): + """Create a reader from a caller-supplied stream object. + The native reader reads through this stream for as long as it is + alive, so the wrapper is stored on the instance and released by + _release(). + + Args: + stream: A stream-like object owned by the caller + format_bytes: UTF-8 encoded format/MIME type + manifest_data: Optional manifest bytes + """ + try: + self._own_stream = Stream(stream) + self._create_reader(format_bytes, self._own_stream, manifest_data) + except Exception: + self._close_streams() + raise + def _init_from_context(self, context, format_or_path, stream, manifest_data=None): """Initialize Reader from a Context object implementing From a41ddadf3407bc23189b7f1f06715b9854990a6b Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 00:25:38 -0700 Subject: [PATCH 02/15] fix: Lock managed resource --- src/c2pa/c2pa.py | 232 ++++++++++------- tests/test_unit_tests_threaded.py | 411 ++++++++++++++++++++++++++++++ 2 files changed, 553 insertions(+), 90 deletions(-) diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 4ef1ee6d..0255e6db 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -19,6 +19,7 @@ import logging import sys import os +import threading import warnings import weakref from abc import ABC, abstractmethod @@ -264,8 +265,37 @@ def _init_attrs(self): def __init__(self): self._lifecycle_state = LifecycleState.UNINITIALIZED self._handle = None + self._op_lock = threading.RLock() record_owner_pid(self) + def _lock(self): + """Return this resource's operation lock. + + Reentrant because CPython can run a finalizer at any bytecode + boundary, including inside a region this thread has already locked, + and because a consuming call tears the handle down from inside the + locked region (_invoke_consume, _raise_consume_failure). + + Falls back to a fresh lock when the attribute is missing: an object + whose __init__ raised before the assignment is still finalized, and + __del__ must not raise. + + Never hold this across a native call that drives stream callbacks + (construction, resource_to_stream, the Builder stream methods, + signing). Those calls release the GIL and re-enter caller-supplied + Python, which may call back into this API on another thread; holding + the lock across them deadlocks. Only calls that touch no callbacks + are serialized here, and no path holds two of these locks at once. + """ + lock = getattr(self, '_op_lock', None) + if lock is None: + lock = threading.RLock() + try: + self._op_lock = lock + except Exception: + pass + return lock + @staticmethod def _free_native_ptr(ptr): """Free a native pointer by passing it to c2pa_free. @@ -321,22 +351,26 @@ def _safe_release(self): def _teardown(self, free_handle: bool): """Close the object: run _release, optionally free the handle, null it. free_handle=False (consumed) frees nothing, the new owner needs to free. + + Holds the operation lock so the free cannot land between another + thread's state check and its use of the handle in a native call. """ - if is_foreign_process(self): - self._handle = None - self._lifecycle_state = LifecycleState.CLOSED - return + with self._lock(): + if is_foreign_process(self): + self._handle = None + self._lifecycle_state = LifecycleState.CLOSED + return - self._lifecycle_state = LifecycleState.CLOSED - self._safe_release() + self._lifecycle_state = LifecycleState.CLOSED + self._safe_release() - handle, self._handle = self._handle, None - if free_handle and handle: - try: - ManagedResource._free_native_ptr(handle) - except Exception: - logger.error("Failed to free native %s resources", - type(self).__name__, exc_info=True) + handle, self._handle = self._handle, None + if free_handle and handle: + try: + ManagedResource._free_native_ptr(handle) + except Exception: + logger.error("Failed to free native %s resources", + type(self).__name__, exc_info=True) def _release_handle(self): """Free this handle, then close the object. Used only where ownership is @@ -1548,16 +1582,17 @@ def set(self, path: str, value: str) -> 'Settings': Returns: self, for method chaining. """ - self._ensure_valid_state() - path_bytes = _to_utf8_bytes(path, "settings path") value_bytes = _to_utf8_bytes(value, "settings value") - _check_ffi_operation_result( - _lib.c2pa_settings_set_value( - self._handle, path_bytes, value_bytes), - "Failed to set settings value", - check=lambda r: r != 0) + with self._lock(): + self._ensure_valid_state() + + _check_ffi_operation_result( + _lib.c2pa_settings_set_value( + self._handle, path_bytes, value_bytes), + "Failed to set settings value", + check=lambda r: r != 0) return self @@ -1574,15 +1609,16 @@ def update( Returns: self, for method chaining. """ - self._ensure_valid_state() - data_bytes = _to_utf8_bytes(data, "settings data") - _check_ffi_operation_result( - _lib.c2pa_settings_update_from_string( - self._handle, data_bytes, b"json"), - "Failed to update settings", - check=lambda r: r != 0) + with self._lock(): + self._ensure_valid_state() + + _check_ffi_operation_result( + _lib.c2pa_settings_update_from_string( + self._handle, data_bytes, b"json"), + "Failed to update settings", + check=lambda r: r != 0) return self @@ -2784,19 +2820,24 @@ def json(self) -> str: C2paError: If there was an error getting the JSON """ - self._ensure_valid_state() + # The state check and the handle read are one critical section: a + # finalizer on another thread frees the handle while it is still + # non-null, so a check made outside the lock says nothing about the + # handle this call goes on to pass to native code. + with self._lock(): + self._ensure_valid_state() - # Return cached result if available - if self._manifest_json_str_cache is not None: - return self._manifest_json_str_cache + # Return cached result if available + if self._manifest_json_str_cache is not None: + return self._manifest_json_str_cache - result = _lib.c2pa_reader_json(self._handle) - _check_ffi_operation_result(result, - "Error during manifest parsing in Reader") + result = _lib.c2pa_reader_json(self._handle) + _check_ffi_operation_result( + result, "Error during manifest parsing in Reader") - # Cache the result and return it - self._manifest_json_str_cache = _convert_to_py_string(result) - return self._manifest_json_str_cache + # Cache the result and return it + self._manifest_json_str_cache = _convert_to_py_string(result) + return self._manifest_json_str_cache def detailed_json(self) -> str: """Get the detailed JSON representation of the C2PA manifest store. @@ -2814,13 +2855,14 @@ def detailed_json(self) -> str: the Reader has been closed. """ - self._ensure_valid_state() + with self._lock(): + self._ensure_valid_state() - result = _lib.c2pa_reader_detailed_json(self._handle) - _check_ffi_operation_result( - result, "Error during detailed manifest parsing in Reader") + result = _lib.c2pa_reader_detailed_json(self._handle) + _check_ffi_operation_result( + result, "Error during detailed manifest parsing in Reader") - return _convert_to_py_string(result) + return _convert_to_py_string(result) def crjson(self) -> str: """Get the manifest store as a crJSON string. @@ -2836,12 +2878,13 @@ def crjson(self) -> str: call returns null. """ - self._ensure_valid_state() + with self._lock(): + self._ensure_valid_state() - result = _lib.c2pa_reader_crjson(self._handle) - _check_ffi_operation_result(result, "Error parsing crJSON") + result = _lib.c2pa_reader_crjson(self._handle) + _check_ffi_operation_result(result, "Error parsing crJSON") - return _convert_to_py_string(result) + return _convert_to_py_string(result) def _get_manifest_field(self, extractor): """Extract a field from (cached) manifest data, or None if unavailable. @@ -2981,11 +3024,12 @@ def is_embedded(self) -> bool: Raises: C2paError: If there was an error checking the embedded status """ - self._ensure_valid_state() + with self._lock(): + self._ensure_valid_state() - result = _lib.c2pa_reader_is_embedded(self._handle) + result = _lib.c2pa_reader_is_embedded(self._handle) - return bool(result) + return bool(result) def get_remote_url(self) -> Optional[str]: """Get the remote URL of the manifest if it was obtained remotely. @@ -2998,17 +3042,18 @@ def get_remote_url(self) -> Optional[str]: Raises: C2paError: If there was an error getting the remote URL """ - self._ensure_valid_state() + with self._lock(): + self._ensure_valid_state() - result = _lib.c2pa_reader_remote_url(self._handle) + result = _lib.c2pa_reader_remote_url(self._handle) - if result is None: - # No remote URL set (manifest is embedded) - return None + if result is None: + # No remote URL set (manifest is embedded) + return None - # Convert the C string to Python string - url_str = _convert_to_py_string(result) - return url_str + # Convert the C string to Python string + url_str = _convert_to_py_string(result) + return url_str class Signer(ManagedResource): @@ -3226,16 +3271,17 @@ def reserve_size(self) -> int: Raises: C2paError: If there was an error getting the size """ - self._ensure_valid_state() + with self._lock(): + self._ensure_valid_state() - result = _lib.c2pa_signer_reserve_size(self._handle) + result = _lib.c2pa_signer_reserve_size(self._handle) - _check_ffi_operation_result( - result, - "Failed to get reserve size", - check=lambda r: r < 0) + _check_ffi_operation_result( + result, + "Failed to get reserve size", + check=lambda r: r < 0) - return result + return result class Builder(ManagedResource): @@ -3433,8 +3479,9 @@ def set_no_embed(self): into the asset when signing. This is useful when creating cloud or sidecar manifests. """ - self._ensure_valid_state() - _lib.c2pa_builder_set_no_embed(self._handle) + with self._lock(): + self._ensure_valid_state() + _lib.c2pa_builder_set_no_embed(self._handle) def set_remote_url(self, remote_url: str): """Set the remote URL. @@ -3448,15 +3495,17 @@ def set_remote_url(self, remote_url: str): Raises: C2paError: If there was an error setting the remote URL """ - self._ensure_valid_state() - url_bytes = _to_utf8_bytes(remote_url, "remote URL") - result = _lib.c2pa_builder_set_remote_url(self._handle, url_bytes) - _check_ffi_operation_result( - result, - Builder._ERROR_MESSAGES['url_error'], - check=lambda r: r != 0) + with self._lock(): + self._ensure_valid_state() + + result = _lib.c2pa_builder_set_remote_url(self._handle, url_bytes) + + _check_ffi_operation_result( + result, + Builder._ERROR_MESSAGES['url_error'], + check=lambda r: r != 0) def set_intent( self, @@ -3484,18 +3533,19 @@ def set_intent( Raises: C2paError: If there was an error setting the intent """ - self._ensure_valid_state() + with self._lock(): + self._ensure_valid_state() - result = _lib.c2pa_builder_set_intent( - self._handle, - ctypes.c_uint(intent), - ctypes.c_uint(digital_source_type), - ) + result = _lib.c2pa_builder_set_intent( + self._handle, + ctypes.c_uint(intent), + ctypes.c_uint(digital_source_type), + ) - _check_ffi_operation_result( - result, - Builder._ERROR_MESSAGES['intent_error'], - check=lambda r: r != 0) + _check_ffi_operation_result( + result, + Builder._ERROR_MESSAGES['intent_error'], + check=lambda r: r != 0) def add_resource(self, uri: str, stream: Any): """Add a resource to the builder. @@ -3599,15 +3649,17 @@ def add_action(self, action_json: Union[str, dict]) -> None: C2paError: If there was an error adding the action C2paError.Encoding: If the action JSON contains invalid UTF-8 chars """ - self._ensure_valid_state() - action_str = _to_utf8_bytes(action_json, "action JSON") - result = _lib.c2pa_builder_add_action(self._handle, action_str) - _check_ffi_operation_result( - result, - Builder._ERROR_MESSAGES['action_error'], - check=lambda r: r != 0) + with self._lock(): + self._ensure_valid_state() + + result = _lib.c2pa_builder_add_action(self._handle, action_str) + + _check_ffi_operation_result( + result, + Builder._ERROR_MESSAGES['action_error'], + check=lambda r: r != 0) def to_archive(self, stream: Any) -> None: """Write an archive of the builder to a stream. diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index d0d0b2c1..f2b2a643 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -16,6 +16,9 @@ import os import io import json +import subprocess +import sys +import textwrap import unittest import threading import concurrent.futures @@ -3035,5 +3038,413 @@ def build_context_and_builder(): self.assertEqual(settings._owner_pid, pid) +class TestManagedResourceLockDeadlock(unittest.TestCase): + """Tests for the operation lock that serializes native calls against + teardown. + + Every join here is bounded: a deadlock must fail the test, not hang the + suite. + """ + + JOIN_TIMEOUT = 30 + + def _join_all(self, threads, what): + for thread in threads: + thread.join(self.JOIN_TIMEOUT) + stuck = [t for t in threads if t.is_alive()] + self.assertEqual( + stuck, [], + "{} did not finish within {}s: deadlock".format( + what, self.JOIN_TIMEOUT)) + + def _run_isolated(self, body, timeout=180): + """Run body in a subprocess and return it. + + A segfault kills the interpreter, so a crash cannot be asserted on + in-process: it would take the test runner with it. + """ + source = textwrap.dedent(body) + return subprocess.run( + [sys.executable, "-c", source], + cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + capture_output=True, + timeout=timeout, + ) + + def test_json_racing_finalizer_does_not_crash(self): + """Readers used on one thread while others are collected. + + Without the lock this segfaults inside c2pa_reader_json: the + finalizer frees the handle between the state check and the call. + """ + result = self._run_isolated(""" + import sys, io, gc, random, threading, time + sys.path.insert(0, "src") + from c2pa import Reader + + data = open("tests/fixtures/C.jpg", "rb").read() + stop = threading.Event() + pool, lock = [], threading.Lock() + + def worker(): + while not stop.is_set(): + choice = random.random() + try: + if choice < 0.40: + reader = Reader("image/jpeg", io.BytesIO(data)) + with lock: + pool.append(reader) + elif choice < 0.75: + with lock: + snapshot = list(pool) + if snapshot: + reader = random.choice(snapshot) + reader._manifest_json_str_cache = None + reader.json() + elif choice < 0.90: + with lock: + reader = pool.pop(0) if pool else None + if reader: + reader.close() + else: + with lock: + if len(pool) > 20: + del pool[0:5] + gc.collect() + except Exception: + pass + + threads = [threading.Thread(target=worker) for _ in range(12)] + for thread in threads: + thread.start() + deadline = time.time() + 10 + while time.time() < deadline: + time.sleep(0.05) + stop.set() + for thread in threads: + thread.join(30) + """) + self.assertEqual( + result.returncode, 0, + "reader churn crashed with {} " + "(139=SIGSEGV, 134=SIGABRT): {}".format( + result.returncode, result.stderr.decode()[-800:])) + + def test_finalizer_inside_locked_operation(self): + """A finalizer can run at any bytecode boundary, including inside a + region this same thread has locked. A non-reentrant lock deadlocks + here; RLock does not. + """ + resource = _ConcreteResource() + resource._activate(0x51000) + observed = [] + + class Dropped: + def __del__(self): + # Runs on this thread, inside the locked region below. + with resource._lock(): + observed.append(True) + + def body(): + with resource._lock(): + dropped = Dropped() + del dropped + gc.collect() + + thread = threading.Thread(target=body) + thread.start() + self._join_all([thread], "finalizer inside locked region") + self.assertEqual(observed, [True], + "finalizer did not re-enter the lock") + resource.close() + + def test_close_racing_json_does_not_deadlock(self): + """close() on one thread against json() on another.""" + data = open(DEFAULT_TEST_FILE, 'rb').read() + errors = [] + + def rounds(): + try: + for _ in range(40): + reader = Reader("image/jpeg", io.BytesIO(data)) + closer = threading.Thread(target=reader.close) + closer.start() + try: + reader._manifest_json_str_cache = None + reader.json() + except Error: + pass + closer.join(self.JOIN_TIMEOUT) + if closer.is_alive(): + errors.append("closer stuck") + return + except Exception as exc: + errors.append(repr(exc)) + + threads = [threading.Thread(target=rounds) for _ in range(4)] + for thread in threads: + thread.start() + self._join_all(threads, "close/json race") + self.assertEqual(errors, []) + + def test_context_manager_exit_racing_json_does_not_deadlock(self): + """__exit__ closes while another thread is calling json().""" + data = open(DEFAULT_TEST_FILE, 'rb').read() + errors = [] + + def body(): + try: + for _ in range(40): + reader = Reader("image/jpeg", io.BytesIO(data)) + + def use(): + for _ in range(5): + try: + reader._manifest_json_str_cache = None + reader.json() + except Error: + pass + + user = threading.Thread(target=use) + user.start() + with reader: + pass + user.join(self.JOIN_TIMEOUT) + if user.is_alive(): + errors.append("user stuck") + return + except Exception as exc: + errors.append(repr(exc)) + + thread = threading.Thread(target=body) + thread.start() + self._join_all([thread], "__exit__/json race") + self.assertEqual(errors, []) + + def test_consume_failure_teardown_does_not_deadlock(self): + """A failing consuming call tears the handle down from inside the + operation, re-entering the lock on the same thread. + + with_fragment on a JPEG returns NotSupported, which routes through + _raise_consume_failure. + """ + data = open(DEFAULT_TEST_FILE, 'rb').read() + errors = [] + + def body(): + try: + for _ in range(20): + reader = Reader("image/jpeg", io.BytesIO(data)) + try: + reader.with_fragment( + "image/jpeg", io.BytesIO(data), io.BytesIO(data)) + except Error: + pass + reader.close() + except Exception as exc: + errors.append(repr(exc)) + + thread = threading.Thread(target=body) + thread.start() + self._join_all([thread], "consume-failure teardown") + self.assertEqual(errors, []) + + def test_close_during_sign_does_not_deadlock(self): + """_sign_internal calls self.close() inside its own try block, so + signing re-enters the lock on the signing thread. + """ + certs = open(os.path.join(FIXTURES_FOLDER, + "es256_certs.pem"), 'rb').read() + key = open(os.path.join(FIXTURES_FOLDER, + "es256_private.key"), 'rb').read() + data = open(DEFAULT_TEST_FILE, 'rb').read() + signer_info = C2paSignerInfo( + alg=b"es256", + sign_cert=certs, + private_key=key, + ta_url=b"http://timestamp.digicert.com", + ) + manifest = { + "claim_generator": "python_test", + "claim_generator_info": [ + {"name": "python_test", "version": "0.0.1"}], + "format": "image/jpeg", + "assertions": [], + } + errors = [] + + def body(): + try: + for _ in range(3): + signer = Signer.from_info(signer_info) + builder = Builder(manifest) + builder.sign(signer, "image/jpeg", + io.BytesIO(data), io.BytesIO()) + except Exception as exc: + errors.append(repr(exc)) + + threads = [threading.Thread(target=body) for _ in range(4)] + for thread in threads: + thread.start() + self._join_all(threads, "sign with internal close") + self.assertEqual(errors, []) + + def test_stream_callback_reentering_api_does_not_deadlock(self): + """Construction drives caller-supplied stream callbacks, and a caller + may legitimately call back into the API from one. + + This passes only because construction does not hold the lock. + """ + data = open(DEFAULT_TEST_FILE, 'rb').read() + other = Reader("image/jpeg", io.BytesIO(data)) + errors = [] + + class ReentrantStream(io.BytesIO): + def readinto(self, buffer): + try: + other.json() + except Exception: + pass + return super().readinto(buffer) + + def body(): + try: + for _ in range(10): + Reader("image/jpeg", ReentrantStream(data)) + except Exception as exc: + errors.append(repr(exc)) + + thread = threading.Thread(target=body) + thread.start() + self._join_all([thread], "callback re-entering API") + self.assertEqual(errors, []) + other.close() + + def test_stream_callback_blocking_on_other_thread_does_not_deadlock(self): + """The adversarial case: a stream callback that blocks on another + thread which touches the same object. + + A lock held across construction deadlocks here, whether it is global + or per-object. This is the test that pins the scoping decision. + """ + data = open(DEFAULT_TEST_FILE, 'rb').read() + target = Reader("image/jpeg", io.BytesIO(data)) + errors = [] + + class BlockingStream(io.BytesIO): + def readinto(self, buffer): + def use(): + try: + target._manifest_json_str_cache = None + target.json() + except Exception: + pass + + helper = threading.Thread(target=use) + helper.start() + helper.join(10) + if helper.is_alive(): + errors.append("helper stuck inside stream callback") + return super().readinto(buffer) + + def body(): + try: + for _ in range(5): + Reader("image/jpeg", BlockingStream(data)) + except Exception as exc: + errors.append(repr(exc)) + + thread = threading.Thread(target=body) + thread.start() + self._join_all([thread], "callback blocking on another thread") + self.assertEqual(errors, []) + target.close() + + def test_no_nested_op_locks(self): + """No code path may hold two resources' operation locks at once. + + That property, not the tests above, is what makes the design + deadlock-free: with only one lock ever held, no cycle can form. + """ + data = open(DEFAULT_TEST_FILE, 'rb').read() + held = threading.local() + violations = [] + real_lock = ManagedResource._lock + + def tracking_lock(resource): + lock = real_lock(resource) + depth = getattr(held, 'stack', None) + if depth is None: + depth = held.stack = [] + + class Tracked: + def __enter__(self): + others = [r for r in depth if r is not resource] + if others: + violations.append( + "{} while holding {}".format( + type(resource).__name__, + [type(o).__name__ for o in others])) + depth.append(resource) + return lock.__enter__() + + def __exit__(self, *exc): + depth.pop() + return lock.__exit__(*exc) + + return Tracked() + + ManagedResource._lock = tracking_lock + try: + reader = Reader("image/jpeg", io.BytesIO(data)) + reader.json() + reader.detailed_json() + reader.is_embedded() + reader.get_remote_url() + reader.close() + finally: + ManagedResource._lock = real_lock + + self.assertEqual(violations, [], + "a thread held two operation locks at once") + + def test_concurrent_storm_terminates(self): + """Readers, closers and collection running together must all finish.""" + data = open(DEFAULT_TEST_FILE, 'rb').read() + stop = threading.Event() + shared = [Reader("image/jpeg", io.BytesIO(data))] + errors = [] + + def reader_worker(): + while not stop.is_set(): + try: + current = shared[0] + current._manifest_json_str_cache = None + current.json() + except Exception: + pass + + def closer_worker(): + while not stop.is_set(): + try: + shared[0].close() + shared[0] = Reader("image/jpeg", io.BytesIO(data)) + gc.collect() + except Exception as exc: + errors.append(repr(exc)) + return + + threads = [threading.Thread(target=reader_worker) for _ in range(6)] + threads += [threading.Thread(target=closer_worker) for _ in range(2)] + for thread in threads: + thread.start() + deadline = time.time() + 5 + while time.time() < deadline: + time.sleep(0.05) + stop.set() + self._join_all(threads, "concurrent storm") + self.assertEqual(errors, []) + + if __name__ == '__main__': unittest.main() From 101655c6471b74f8d56fb01194ddbf2a82593eba Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 00:36:47 -0700 Subject: [PATCH 03/15] fix: Warnings in tests --- tests/test_unit_tests_threaded.py | 33 ++++++++++++++++++++----------- 1 file changed, 21 insertions(+), 12 deletions(-) diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index f2b2a643..c1118534 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -3048,6 +3048,17 @@ class TestManagedResourceLockDeadlock(unittest.TestCase): JOIN_TIMEOUT = 30 + @classmethod + def setUpClass(cls): + with open(DEFAULT_TEST_FILE, 'rb') as handle: + cls.image_bytes = handle.read() + with open(os.path.join(FIXTURES_FOLDER, + "es256_certs.pem"), 'rb') as handle: + cls.certs = handle.read() + with open(os.path.join(FIXTURES_FOLDER, + "es256_private.key"), 'rb') as handle: + cls.private_key = handle.read() + def _join_all(self, threads, what): for thread in threads: thread.join(self.JOIN_TIMEOUT) @@ -3160,7 +3171,7 @@ def body(): def test_close_racing_json_does_not_deadlock(self): """close() on one thread against json() on another.""" - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes errors = [] def rounds(): @@ -3189,7 +3200,7 @@ def rounds(): def test_context_manager_exit_racing_json_does_not_deadlock(self): """__exit__ closes while another thread is calling json().""" - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes errors = [] def body(): @@ -3228,7 +3239,7 @@ def test_consume_failure_teardown_does_not_deadlock(self): with_fragment on a JPEG returns NotSupported, which routes through _raise_consume_failure. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes errors = [] def body(): @@ -3253,11 +3264,9 @@ def test_close_during_sign_does_not_deadlock(self): """_sign_internal calls self.close() inside its own try block, so signing re-enters the lock on the signing thread. """ - certs = open(os.path.join(FIXTURES_FOLDER, - "es256_certs.pem"), 'rb').read() - key = open(os.path.join(FIXTURES_FOLDER, - "es256_private.key"), 'rb').read() - data = open(DEFAULT_TEST_FILE, 'rb').read() + certs = self.certs + key = self.private_key + data = self.image_bytes signer_info = C2paSignerInfo( alg=b"es256", sign_cert=certs, @@ -3295,7 +3304,7 @@ def test_stream_callback_reentering_api_does_not_deadlock(self): This passes only because construction does not hold the lock. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes other = Reader("image/jpeg", io.BytesIO(data)) errors = [] @@ -3327,7 +3336,7 @@ def test_stream_callback_blocking_on_other_thread_does_not_deadlock(self): A lock held across construction deadlocks here, whether it is global or per-object. This is the test that pins the scoping decision. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes target = Reader("image/jpeg", io.BytesIO(data)) errors = [] @@ -3366,7 +3375,7 @@ def test_no_nested_op_locks(self): That property, not the tests above, is what makes the design deadlock-free: with only one lock ever held, no cycle can form. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes held = threading.local() violations = [] real_lock = ManagedResource._lock @@ -3410,7 +3419,7 @@ def __exit__(self, *exc): def test_concurrent_storm_terminates(self): """Readers, closers and collection running together must all finish.""" - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes stop = threading.Event() shared = [Reader("image/jpeg", io.BytesIO(data))] errors = [] From 78e5d862b8b457078ef402f3f7917d3840dd9f3e Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 08:32:45 -0700 Subject: [PATCH 04/15] WIP 2 (#313) * fix: Warnings in tests * fix: Deferred teardown * fix: Protect signer * fix: Borrow test * fix: Borrow test 2 --- src/c2pa/c2pa.py | 224 ++++++++---- tests/test_unit_tests_threaded.py | 581 +++++++++++++++++++++++++++++- 2 files changed, 732 insertions(+), 73 deletions(-) diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 0255e6db..6e0fe836 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -13,6 +13,7 @@ # Version: 0.37.8 +import contextlib import ctypes import enum import json @@ -266,6 +267,8 @@ def __init__(self): self._lifecycle_state = LifecycleState.UNINITIALIZED self._handle = None self._op_lock = threading.RLock() + self._inflight = 0 + self._pending_teardown = None record_owner_pid(self) def _lock(self): @@ -296,6 +299,39 @@ def _lock(self): pass return lock + @contextlib.contextmanager + def _native_call(self): + """Hold the handle valid across a native call that goes back + and forth to native layers. + + Calls that pass a Stream to the native library run caller-supplied + callbacks, so the lock cannot be held across them: the callback may + re-enter this API on another thread and deadlock. Instead the call is + counted as in flight, and a teardown arriving meanwhile records its + intent rather than freeing. The last caller out performs the free. + + The resource is marked closed as soon as the teardown is recorded, so + a caller that closed it cannot keep using it while the free is + pending. + """ + with self._lock(): + self._ensure_valid_state() + self._inflight = getattr(self, '_inflight', 0) + 1 + try: + yield + finally: + with self._lock(): + self._inflight -= 1 + pending = (self._pending_teardown + if self._inflight == 0 else None) + if pending is not None: + self._pending_teardown = None + # Released the lock before the free: _teardown takes it again, + # and keeping the two acquisitions separate means the counter + # update is never held across the release work. + if pending is not None: + self._teardown(pending) + @staticmethod def _free_native_ptr(ptr): """Free a native pointer by passing it to c2pa_free. @@ -356,6 +392,16 @@ def _teardown(self, free_handle: bool): thread's state check and its use of the handle in a native call. """ with self._lock(): + if getattr(self, '_inflight', 0) > 0: + # A native call is running that re-enters caller Python and + # is still using this handle. Record the intent; whichever + # caller leaves _native_call last performs the free. Mark the + # resource closed now so it cannot be used while the free is + # pending. + self._pending_teardown = free_handle + self._lifecycle_state = LifecycleState.CLOSED + return + if is_foreign_process(self): self._handle = None self._lifecycle_state = LifecycleState.CLOSED @@ -1735,13 +1781,22 @@ def __init__( check=lambda r: r != 0) if signer is not None: - signer._ensure_valid_state() - # A rejected signer is retained, not closed and leaked. - self._signer_callback_cb = signer._callback_cb - signer._consume_no_replacement( - lambda h: _lib.c2pa_context_builder_set_signer( - nb._handle, h), - "Failed to set signer on Context: {}") + # The signer's own in-flight guard: this hands its handle + # to native, so a signer.close() on another thread must + # not free it between the state check and the call. The + # guard also makes the check and the consume atomic. + # + # _consume_no_replacement tears the signer down from + # inside this region. A teardown recorded while the guard + # is held is deferred and performed as the guard unwinds, + # which is still before __init__ returns. + with signer._native_call(): + # A rejected signer is retained, not closed and leaked. + self._signer_callback_cb = signer._callback_cb + signer._consume_no_replacement( + lambda h: _lib.c2pa_context_builder_set_signer( + nb._handle, h), + "Failed to set signer on Context: {}") self._has_signer = True context_ptr = nb._consume_into( @@ -2658,12 +2713,19 @@ def _init_from_context(self, context, format_or_path, self._own_stream = Stream(stream) try: - # Adopt before the consuming call: _consume_and_swap needs an - # active resource, and cleanup then owns the pointer either way. - self._create_and_activate( - lambda: _lib.c2pa_reader_from_context( - context.execution_context), - Reader._ERROR_MESSAGES['reader_error']) + # The Context is caller-supplied and may be shared, so its handle + # needs its own in-flight guard across the native call: the + # execution_context property validates and returns the handle, and + # without the guard a context.close() on another thread could free + # it before c2pa_reader_from_context reads it. + with context._native_call(): + # Adopt before the consuming call: _consume_and_swap needs an + # active resource, and cleanup then owns the pointer either + # way. + self._create_and_activate( + lambda: _lib.c2pa_reader_from_context( + context.execution_context), + Reader._ERROR_MESSAGES['reader_error']) if manifest_data is not None: manifest_array = ( @@ -2702,6 +2764,10 @@ def _init_attrs(self): # Tracks a file we opened ourselves and must close later. self._backing_file = None + # Fragment streams handed to the native reader by with_fragment, + # which it keeps reading from for the rest of its life. + self._fragment_streams = [] + # Caches for manifest JSON string and parsed data. # These are invalidated when with_fragment() is called. self._manifest_json_str_cache = None @@ -2725,6 +2791,12 @@ def _close_streams(self): logger.warning("Failed to close Reader backing file") finally: self._backing_file = None + for fragment in getattr(self, '_fragment_streams', []): + try: + fragment.close() + except Exception: + logger.warning("Failed to close Reader fragment stream") + self._fragment_streams = [] def _release(self): """Release Reader-specific resources (caches, stream, backing file). @@ -2787,19 +2859,38 @@ def with_fragment(self, format: Optional[str], stream, cannot be retried: create a new one instead of reusing this instance. """ - self._ensure_valid_state() - format_arg = _format_ffi_arg(_encode_format(format, "Reader")) - with Stream(stream) as main_obj, Stream(fragment_stream) as frag_obj: - self._consume_and_swap( - lambda handle: _lib.c2pa_reader_with_fragment( - handle, - format_arg, - main_obj._stream, - frag_obj._stream, - ), - Reader._ERROR_MESSAGES['fragment_error']) + # The native reader keeps reading through both streams after this + # returns, so they are owned here and released by _release() rather + # than at the end of a with block. + main_obj = Stream(stream) + frag_obj = Stream(fragment_stream) + try: + with self._native_call(): + self._consume_and_swap( + lambda handle: _lib.c2pa_reader_with_fragment( + handle, + format_arg, + main_obj._stream, + frag_obj._stream, + ), + Reader._ERROR_MESSAGES['fragment_error']) + except Exception: + main_obj.close() + frag_obj.close() + raise + + # Replace the streams this reader owned, closing the previous ones so + # repeated calls do not accumulate them. + previous = self._own_stream + self._own_stream = main_obj + self._fragment_streams.append(frag_obj) + if previous is not None and previous is not main_obj: + try: + previous.close() + except Exception: + logger.warning("Failed to close previous Reader stream") # Invalidate caches: processing a new BMFF fragment updates the native # reader's state, which can change the manifest data it returns. @@ -3000,10 +3091,8 @@ def resource_to_stream(self, uri: str, stream: Any) -> int: Raises: C2paError: If there was an error writing the resource to stream """ - self._ensure_valid_state() - uri_str = uri.encode('utf-8') - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: result = _lib.c2pa_reader_resource_to_stream( self._handle, uri_str, stream_obj._stream) @@ -3451,11 +3540,17 @@ def _init_from_context(self, context, json_str): if not context.is_valid: raise C2paError("Context is not valid") - # Adopt before the consuming call: _consume_and_swap needs an - # active resource, and cleanup then owns the pointer either way. - self._create_and_activate( - lambda: _lib.c2pa_builder_from_context(context.execution_context), - Builder._ERROR_MESSAGES['builder_error']) + # The Context is caller-supplied and may be shared, so its handle + # needs its own in-flight guard across the native call: without it a + # context.close() on another thread frees the handle between the + # is_valid check and c2pa_builder_from_context reading it. + with context._native_call(): + # Adopt before the consuming call: _consume_and_swap needs an + # active resource, and cleanup then owns the pointer either way. + self._create_and_activate( + lambda: _lib.c2pa_builder_from_context( + context.execution_context), + Builder._ERROR_MESSAGES['builder_error']) self._consume_and_swap( lambda handle: _lib.c2pa_builder_with_definition( @@ -3558,10 +3653,8 @@ def add_resource(self, uri: str, stream: Any): Raises: C2paError: If there was an error adding the resource """ - self._ensure_valid_state() - uri_bytes = _to_utf8_bytes(uri, "resource URI") - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: result = _lib.c2pa_builder_add_resource( self._handle, uri_bytes, stream_obj._stream) @@ -3622,7 +3715,7 @@ def add_ingredient_from_stream( ingredient_str = _to_utf8_bytes(ingredient_json, "ingredient JSON") format_str = _to_utf8_bytes(format, "ingredient format") - with Stream(source) as source_stream: + with self._native_call(), Stream(source) as source_stream: result = ( _lib.c2pa_builder_add_ingredient_from_stream( self._handle, @@ -3671,9 +3764,7 @@ def to_archive(self, stream: Any) -> None: Raises: C2paError: If there was an error writing the archive """ - self._ensure_valid_state() - - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: result = _lib.c2pa_builder_to_archive( self._handle, stream_obj._stream) @@ -3698,7 +3789,7 @@ def write_ingredient_archive(self, ingredient_id: str, stream: Any) -> None: ingredient_id_str = _to_utf8_bytes(ingredient_id, "ingredient_id") - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: result = _lib.c2pa_builder_write_ingredient_archive( self._handle, ingredient_id_str, stream_obj._stream) @@ -3718,9 +3809,7 @@ def add_ingredient_from_archive(self, stream: Any) -> None: Raises: C2paError: If there was an error reading the archive """ - self._ensure_valid_state() - - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: result = _lib.c2pa_builder_add_ingredient_from_archive( self._handle, stream_obj._stream) @@ -3749,7 +3838,7 @@ def with_archive(self, stream: Any) -> 'Builder': """ self._ensure_valid_state() - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: self._consume_and_swap( lambda handle: _lib.c2pa_builder_with_archive( handle, stream_obj._stream), @@ -3797,23 +3886,36 @@ def _sign_internal( manifest_bytes_ptr = ctypes.POINTER(ctypes.c_ubyte)() try: - if signer is not None: - result = _lib.c2pa_builder_sign( - self._handle, - format_arg, - source_stream._stream, - dest_stream._stream, - signer._handle, - ctypes.byref(manifest_bytes_ptr) - ) - else: - result = _lib.c2pa_builder_sign_context( - self._handle, - format_arg, - source_stream._stream, - dest_stream._stream, - ctypes.byref(manifest_bytes_ptr), - ) + # _native_call covers the signing call only. The close() below is + # deliberately outside it, so the deferred teardown it records is + # performed on the way out rather than being deferred forever. + with self._native_call(): + if signer is not None: + # c2pa_builder_sign borrows the signer's handle, so the + # signer needs its own in-flight guard: the Builder's + # guard holds only the Builder's handle valid, and a + # signer.close() on another thread would otherwise free + # this handle mid-call. Entered inside self's guard so + # concurrent signs sharing objects acquire in one order. + # The check above is a fast-fail; this re-check inside + # the guard is the one that makes check-then-use atomic. + with signer._native_call(): + result = _lib.c2pa_builder_sign( + self._handle, + format_arg, + source_stream._stream, + dest_stream._stream, + signer._handle, + ctypes.byref(manifest_bytes_ptr) + ) + else: + result = _lib.c2pa_builder_sign_context( + self._handle, + format_arg, + source_stream._stream, + dest_stream._stream, + ctypes.byref(manifest_bytes_ptr), + ) # Sign borrows the Builder without taking ownership. # Closing here ensures resources clean up, # and single use/single sign done by a Builder. diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index f2b2a643..0b9b4f95 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -11,9 +11,12 @@ # specific language governing permissions and limitations under # each license. +import ast import ctypes import gc import os +import re +import inspect import io import json import subprocess @@ -3048,6 +3051,17 @@ class TestManagedResourceLockDeadlock(unittest.TestCase): JOIN_TIMEOUT = 30 + @classmethod + def setUpClass(cls): + with open(DEFAULT_TEST_FILE, 'rb') as handle: + cls.image_bytes = handle.read() + with open(os.path.join(FIXTURES_FOLDER, + "es256_certs.pem"), 'rb') as handle: + cls.certs = handle.read() + with open(os.path.join(FIXTURES_FOLDER, + "es256_private.key"), 'rb') as handle: + cls.private_key = handle.read() + def _join_all(self, threads, what): for thread in threads: thread.join(self.JOIN_TIMEOUT) @@ -3160,7 +3174,7 @@ def body(): def test_close_racing_json_does_not_deadlock(self): """close() on one thread against json() on another.""" - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes errors = [] def rounds(): @@ -3189,7 +3203,7 @@ def rounds(): def test_context_manager_exit_racing_json_does_not_deadlock(self): """__exit__ closes while another thread is calling json().""" - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes errors = [] def body(): @@ -3228,7 +3242,7 @@ def test_consume_failure_teardown_does_not_deadlock(self): with_fragment on a JPEG returns NotSupported, which routes through _raise_consume_failure. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes errors = [] def body(): @@ -3253,11 +3267,9 @@ def test_close_during_sign_does_not_deadlock(self): """_sign_internal calls self.close() inside its own try block, so signing re-enters the lock on the signing thread. """ - certs = open(os.path.join(FIXTURES_FOLDER, - "es256_certs.pem"), 'rb').read() - key = open(os.path.join(FIXTURES_FOLDER, - "es256_private.key"), 'rb').read() - data = open(DEFAULT_TEST_FILE, 'rb').read() + certs = self.certs + key = self.private_key + data = self.image_bytes signer_info = C2paSignerInfo( alg=b"es256", sign_cert=certs, @@ -3295,7 +3307,7 @@ def test_stream_callback_reentering_api_does_not_deadlock(self): This passes only because construction does not hold the lock. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes other = Reader("image/jpeg", io.BytesIO(data)) errors = [] @@ -3327,7 +3339,7 @@ def test_stream_callback_blocking_on_other_thread_does_not_deadlock(self): A lock held across construction deadlocks here, whether it is global or per-object. This is the test that pins the scoping decision. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes target = Reader("image/jpeg", io.BytesIO(data)) errors = [] @@ -3366,7 +3378,7 @@ def test_no_nested_op_locks(self): That property, not the tests above, is what makes the design deadlock-free: with only one lock ever held, no cycle can form. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes held = threading.local() violations = [] real_lock = ManagedResource._lock @@ -3410,7 +3422,7 @@ def __exit__(self, *exc): def test_concurrent_storm_terminates(self): """Readers, closers and collection running together must all finish.""" - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes stop = threading.Event() shared = [Reader("image/jpeg", io.BytesIO(data))] errors = [] @@ -3445,6 +3457,551 @@ def closer_worker(): self._join_all(threads, "concurrent storm") self.assertEqual(errors, []) + def _counted_free(self): + """Patch _free_native_ptr to count frees; returns the list.""" + freed = [] + real = ManagedResource._free_native_ptr + + def counting(ptr): + freed.append(ptr) + return real(ptr) + + ManagedResource._free_native_ptr = staticmethod(counting) + self.addCleanup( + lambda: setattr(ManagedResource, '_free_native_ptr', real)) + return freed + + def _thumbnail_uri(self, reader): + manifests = json.loads(reader.json()).get("manifests", {}) + for manifest in manifests.values(): + thumbnail = manifest.get("thumbnail") + if thumbnail and thumbnail.get("identifier"): + return thumbnail["identifier"] + self.skipTest("fixture has no thumbnail resource to stream") + + def test_close_inside_callback_defers_free(self): + """A close() from inside a stream callback must not free the handle + the native call is still using.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + during = [] + + class Closer(io.BytesIO): + def write(self, buffer): + reader.close() + during.append(len(freed)) + return super().write(buffer) + + try: + reader.resource_to_stream(uri, Closer()) + except Error: + pass + + self.assertEqual(during, [0], "handle was freed mid-call") + self.assertEqual(len(freed), 1, "deferred free did not run once") + self.assertEqual(reader._inflight, 0) + self.assertIsNone(reader._pending_teardown) + self.assertEqual(reader._lifecycle_state, LifecycleState.CLOSED) + + def test_cross_thread_close_during_callback_defers_free(self): + """Same race, with the close arriving from another thread.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + during = [] + started = threading.Event() + + class Slow(io.BytesIO): + def write(self, buffer): + started.set() + time.sleep(0.3) + during.append(len(freed)) + return super().write(buffer) + + def closer(): + started.wait(self.JOIN_TIMEOUT) + reader.close() + + thread = threading.Thread(target=closer) + thread.start() + try: + reader.resource_to_stream(uri, Slow()) + except Error: + pass + self._join_all([thread], "cross-thread closer") + + self.assertEqual(during, [0], "handle was freed mid-call") + self.assertEqual(len(freed), 1) + self.assertEqual(reader._inflight, 0) + + def test_deferred_teardown_still_closes(self): + """After a deferred free the resource is closed and a later close() + is a no-op rather than a second free.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + + class Closer(io.BytesIO): + def write(self, buffer): + reader.close() + return super().write(buffer) + + try: + reader.resource_to_stream(uri, Closer()) + except Error: + pass + + self.assertEqual(len(freed), 1) + reader.close() + self.assertEqual(len(freed), 1, "second close() freed again") + self.assertIsNone(reader._handle) + + def test_use_after_deferred_close_is_rejected(self): + """Deferring must not leave the resource usable: the free is pending, + so the handle is about to go away.""" + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + states = [] + + class Closer(io.BytesIO): + def write(self, buffer): + reader.close() + states.append(reader._lifecycle_state) + try: + reader.json() + states.append("json succeeded") + except Error: + states.append("json rejected") + return super().write(buffer) + + try: + reader.resource_to_stream(uri, Closer()) + except Error: + pass + + self.assertEqual(states[0], LifecycleState.CLOSED) + self.assertEqual(states[1], "json rejected") + + def test_exception_from_callback_still_frees(self): + """An exception unwinding through the native call must not strand the + in-flight counter, or the handle is never freed.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + + class Exploding(io.BytesIO): + def write(self, buffer): + reader.close() + raise RuntimeError("callback failure") + + try: + reader.resource_to_stream(uri, Exploding()) + except Exception: + pass + + self.assertEqual(reader._inflight, 0, "in-flight counter stranded") + self.assertEqual(len(freed), 1, "deferred free did not run") + + def test_inflight_cleared_before_deferred_free(self): + """The counter must reach zero before the deferred free runs. + + _teardown defers whenever _inflight is above zero, so performing the + free while the counter is still raised would defer it a second time + and the handle would never be released. + """ + seen = [] + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + real_release = Reader._release + + def probing_release(self): + seen.append(self._inflight) + return real_release(self) + + class Closer(io.BytesIO): + def write(self, buffer): + reader.close() + return super().write(buffer) + + with patch.object(Reader, '_release', probing_release): + try: + reader.resource_to_stream(uri, Closer()) + except Error: + pass + + self.assertEqual(seen, [0], + "deferred free ran while still counted in flight") + self.assertIsNone(reader._handle) + + def test_release_raising_during_deferred_teardown_does_not_leak(self): + """The deferred free survives a failing _release: the handle must + still be freed.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + + def boom(self): + raise RuntimeError("release failure") + + class Closer(io.BytesIO): + def write(self, buffer): + reader.close() + return super().write(buffer) + + with patch.object(Reader, '_release', boom): + try: + reader.resource_to_stream(uri, Closer()) + except Error: + pass + + self.assertEqual(reader._inflight, 0) + self.assertEqual(len(freed), 1, "handle leaked when _release raised") + + def test_concurrent_closes_during_callback_free_once(self): + """Many threads closing while one native call is in flight must + produce exactly one free.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + started = threading.Event() + closers = [] + + class Slow(io.BytesIO): + def write(self, buffer): + started.set() + time.sleep(0.3) + return super().write(buffer) + + def closer(): + started.wait(self.JOIN_TIMEOUT) + reader.close() + + for _ in range(8): + thread = threading.Thread(target=closer) + closers.append(thread) + thread.start() + try: + reader.resource_to_stream(uri, Slow()) + except Error: + pass + self._join_all(closers, "concurrent closers") + + self.assertEqual(len(freed), 1, + "racing closers freed {} times".format(len(freed))) + self.assertEqual(reader._inflight, 0) + + def test_sign_with_internal_close_frees_once(self): + """_sign_internal closes the Builder inside its own try, so the close + defers and the free happens on the way out.""" + freed = self._counted_free() + signer_info = C2paSignerInfo( + alg=b"es256", + sign_cert=self.certs, + private_key=self.private_key, + ta_url=b"http://timestamp.digicert.com", + ) + manifest = { + "claim_generator": "python_test", + "claim_generator_info": [ + {"name": "python_test", "version": "0.0.1"}], + "format": "image/jpeg", + "assertions": [], + } + signer = Signer.from_info(signer_info) + builder = Builder(manifest) + builder.sign(signer, "image/jpeg", + io.BytesIO(self.image_bytes), io.BytesIO()) + + self.assertEqual(builder._lifecycle_state, LifecycleState.CLOSED) + self.assertEqual(builder._inflight, 0) + builder_frees = [f for f in freed if f is not None] + self.assertGreaterEqual(len(builder_frees), 1) + with self.assertRaises(Error): + builder.sign(signer, "image/jpeg", + io.BytesIO(self.image_bytes), io.BytesIO()) + + def test_class_a_construction_is_not_guarded(self): + """Construction is deliberately unguarded: no external caller holds a + reference yet, and guarding it would reintroduce the deadlock where a + stream callback re-enters the API.""" + entered = [] + real = ManagedResource._native_call + + def recording(resource): + entered.append(type(resource).__name__) + return real(resource) + + ManagedResource._native_call = recording + try: + Reader("image/jpeg", io.BytesIO(self.image_bytes)) + finally: + ManagedResource._native_call = real + + self.assertEqual(entered, [], + "construction entered _native_call: guarding it " + "reintroduces the callback deadlock") + + def test_every_callback_running_method_is_guarded(self): + """Coverage check: every method that hands a Stream to the native + library must be guarded, except the three construction paths. + + A method missed here keeps the use-after-free, and the symptom is a + rare segfault rather than a failing test, so this is checked + mechanically rather than by eye. + """ + source = inspect.getsource(sys.modules[Reader.__module__]) + lines = source.split("\n") + class_a = { + ("Reader", "_create_reader"), + ("Reader", "_init_from_context"), + ("Builder", "from_archive"), + } + stream_use = re.compile( + r"(_stream|stream_obj|source_stream|dest_stream|main_obj" + r"|frag_obj)\._stream") + + bodies = {} + current_class = current_method = None + start = None + for index, line in enumerate(lines): + if re.match(r"^class ", line): + current_class = line.split("(")[0].replace( + "class ", "").strip(":") + if re.match(r"^def ", line): + current_class = None + match = re.match(r"^ def (\w+)", line) + if match: + if current_class and current_method and start is not None: + bodies[(current_class, current_method)] = "\n".join( + lines[start:index]) + current_method = match.group(1) + start = index + if current_class and current_method and start is not None: + bodies[(current_class, current_method)] = "\n".join(lines[start:]) + + unguarded = [] + checked = 0 + for key, body in bodies.items(): + if not stream_use.search(body): + continue + checked += 1 + if key in class_a: + continue + if "_native_call()" not in body: + unguarded.append("{}.{}".format(*key)) + + self.assertGreater(checked, 0, "coverage scan found no methods") + self.assertEqual( + unguarded, [], + "these hand a Stream to native without _native_call(): {}".format( + unguarded)) + + def test_every_borrowed_handle_is_guarded(self): + """Coverage check: when a method hands a *second* object's handle to + the native library, that object needs its own _native_call() guard. + + test_every_callback_running_method_is_guarded only asks whether the + string "_native_call()" appears in the method body, which cannot + express *whose* handle is guarded. A method that guards self while + passing signer._handle to native passes that check and still has the + use-after-free, so the ownership is checked structurally here. + """ + module = sys.modules[Reader.__module__] + tree = ast.parse(inspect.getsource(module)) + + # Attributes that carry a native handle out of an object. + handle_attrs = {"_handle", "execution_context"} + + def guarded_names(node): + """Names X with an active `with X._native_call():` at this node.""" + found = set() + for item in getattr(node, "items", []): + call = item.context_expr + if (isinstance(call, ast.Call) + and isinstance(call.func, ast.Attribute) + and call.func.attr == "_native_call" + and isinstance(call.func.value, ast.Name)): + found.add(call.func.value.id) + return found + + def borrowed_in_call(call): + """Names X whose handle this _lib.* call receives, X not self.""" + if not (isinstance(call.func, ast.Attribute) + and isinstance(call.func.value, ast.Name) + and call.func.value.id == "_lib"): + return set() + names = set() + for arg in ast.walk(call): + if (isinstance(arg, ast.Attribute) + and arg.attr in handle_attrs + and isinstance(arg.value, ast.Name) + and arg.value.id != "self"): + names.add(arg.value.id) + return names + + def locally_owned(method): + """Names bound to an object this method itself constructed. + + A resource created inside the method never escapes to another + thread, so nothing can close it mid-call and it needs no guard. + Only handles reaching the method from outside (parameters, + attributes) are exposed to a concurrent teardown. + """ + owned = set() + for node in ast.walk(method): + # `with self._NativeBuilder() as nb:` / `x = Foo()` + if isinstance(node, (ast.With, ast.AsyncWith)): + for item in node.items: + if (isinstance(item.context_expr, ast.Call) + and isinstance(item.optional_vars, ast.Name)): + owned.add(item.optional_vars.id) + elif isinstance(node, ast.Assign): + if isinstance(node.value, ast.Call): + for target in node.targets: + if isinstance(target, ast.Name): + owned.add(target.id) + return owned + + unguarded = [] + checked = 0 + + for cls in ast.walk(tree): + if not isinstance(cls, ast.ClassDef): + continue + for method in cls.body: + if not isinstance(method, (ast.FunctionDef, + ast.AsyncFunctionDef)): + continue + owned = locally_owned(method) + + # Walk the body tracking which guards are open, so a borrowed + # handle is only accepted when its own guard encloses the use. + def visit(node, active): + nonlocal checked + if isinstance(node, (ast.With, ast.AsyncWith)): + active = active | guarded_names(node) + if isinstance(node, ast.Call): + for name in borrowed_in_call(node) - owned: + checked += 1 + if name not in active: + unguarded.append( + "{}.{} passes {}._handle to native " + "without {}._native_call()".format( + cls.name, method.name, name, name)) + for child in ast.iter_child_nodes(node): + visit(child, active) + + visit(method, frozenset()) + + self.assertGreater( + checked, 0, + "ownership scan found no borrowed handles: the scan is broken") + self.assertEqual( + unguarded, [], + "borrowed handles used without their own guard:\n " + + "\n ".join(unguarded)) + + +class TestSharedSignerTeardownRace(unittest.TestCase): + """A Signer shared across threads must not be freed mid-sign. + + Builder.sign borrows the signer's handle for the duration of the native + call. Without a guard on the signer itself, a close() on another thread + frees that handle while c2pa_builder_sign is using it, and the process + dies with SIGSEGV instead of raising. + """ + + def setUp(self): + self.data_dir = os.path.join(os.path.dirname(__file__), "fixtures") + with open(os.path.join(self.data_dir, "C.jpg"), "rb") as f: + self.image_bytes = f.read() + with open(os.path.join(self.data_dir, "es256_certs.pem"), "rb") as f: + self.certs = f.read() + with open(os.path.join(self.data_dir, "es256_private.key"), "rb") as f: + self.key = f.read() + self.manifest = { + "claim_generator_info": [{"name": "test", "version": "0.1"}], + "assertions": [], + } + + def _make_signer(self): + return Signer.from_info(C2paSignerInfo( + SigningAlg.ES256, self.certs, self.key, None)) + + def test_close_during_concurrent_sign_does_not_crash(self): + """Rotate a shared signer while other threads sign with it. + + Runs in a subprocess: the failure mode is a segfault, which would + take the test runner down with it rather than reporting a failure. + """ + source = textwrap.dedent(""" + import io, os, sys, threading + from c2pa import (Builder, Signer, C2paSignerInfo, + C2paSigningAlg as SigningAlg) + + data_dir = sys.argv[1] + certs = open(os.path.join(data_dir, "es256_certs.pem"), "rb").read() + key = open(os.path.join(data_dir, "es256_private.key"), "rb").read() + img = open(os.path.join(data_dir, "C.jpg"), "rb").read() + manifest = {"claim_generator_info": + [{"name": "test", "version": "0.1"}], + "assertions": []} + + def make(): + return Signer.from_info(C2paSignerInfo( + SigningAlg.ES256, certs, key, None)) + + box = {"signer": make(), "stop": False} + + def rotate(): + while not box["stop"]: + old = box["signer"] + try: + box["signer"] = make() + old.close() + except Exception: + pass + + def sign(): + for _ in range(120): + if box["stop"]: + return + try: + b = Builder(manifest) + b.sign(box["signer"], "image/jpeg", + io.BytesIO(img), io.BytesIO()) + b.close() + except Exception: + # A closed signer may legitimately be rejected; + # only a crash is a failure here. + pass + + rot = threading.Thread(target=rotate, daemon=True) + rot.start() + threads = [threading.Thread(target=sign) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + box["stop"] = True + rot.join(timeout=5) + print("OK") + """) + + result = subprocess.run( + [sys.executable, "-c", source, self.data_dir], + capture_output=True, text=True, timeout=300) + + self.assertNotEqual( + result.returncode, -11, + "SIGSEGV: a signer was freed while a sign was using its handle") + self.assertEqual( + result.returncode, 0, + "shared-signer teardown race failed (rc={}):\n{}".format( + result.returncode, result.stderr[-2000:])) + self.assertIn("OK", result.stdout) + if __name__ == '__main__': unittest.main() From 2eaeec0e9390d980eb28a22fec436ea2abd3d8b0 Mon Sep 17 00:00:00 2001 From: Tania Mathern Date: Tue, 25 Aug 2026 09:43:36 -0700 Subject: [PATCH 05/15] fix: Notes clean up --- pr312-followups.md | 108 --------------------------------------------- 1 file changed, 108 deletions(-) delete mode 100644 pr312-followups.md diff --git a/pr312-followups.md b/pr312-followups.md deleted file mode 100644 index 9dceafee..00000000 --- a/pr312-followups.md +++ /dev/null @@ -1,108 +0,0 @@ -# c2pa-python pull request 312: additional fixes - -Sentinel in the native thread-local error slot. Scratch note, not for the -repository. - -The change is correct as written. Three items, the first of which quietly -disables the diagnostic the sentinel exists to provide. - ---- - -## 1. Match on a substring, not on exact equality - -**The problem.** The comparison is - -```python -if error == ManagedResource._NO_NATIVE_ERROR.decode('utf-8'): -``` - -`c2pa_error_set_last` does not store the string verbatim. It runs it through -`Error::from` and then `CimplError::from`, and its own documentation states that -a missing or invalid error type is replaced with `Other` and the message includes -the original string. The sentinel already carries an `Other: ` prefix, so it may -round-trip unchanged, or it may come back re-prefixed or otherwise normalised. - -**Why it matters, and why it is easy to miss.** The failure is silent rather than -dangerous. If the comparison never matches, control falls through to the final -branch, which now performs the identical `_teardown(free_handle=False)`. Same -action, no crash, all tests that check behaviour still pass. - -What is lost is the distinct log line. That log line is the entire reason for -planting a sentinel rather than simply clearing the slot: it separates "the -native side reported nothing" from "the native side reported a real error", and -it is the only field evidence available for how often the ambiguous case occurs. -Losing it costs nothing today and costs the whole diagnostic tomorrow. - -**Fix.** Match on the distinctive part only: - -```python -_NO_NATIVE_ERROR_MARKER = "c2pa-python-no-native-error" -... -if _NO_NATIVE_ERROR_MARKER in error: -``` - -**And pin the round-trip in a test regardless**, since it is a property of the -native side that can change without notice: - -```python -def test_sentinel_round_trips_through_native_error_slot(self): - c2pa_module._lib.c2pa_error_set_last(ManagedResource._NO_NATIVE_ERROR) - self.assertIn(_NO_NATIVE_ERROR_MARKER, c2pa_module._read_native_error()) -``` - -That test fails loudly if the normalisation ever changes, which is exactly the -kind of upstream invariant worth pinning rather than assuming. - -## 2. Restore the ordering rationale that was deleted - -The removed paragraph explained that `c2pa_free` on a handle the registry no -longer tracks returns minus one and overwrites the slot with its own -untracked-pointer message, so the error must be read before any free or the -substitute carries a pre-consume tag and inverts the retain decision. - -That constraint is still true, and the current code still depends on it. The -replacement text explains the sentinel but says nothing about why the read comes -first. A later edit that moves the read after a free would reintroduce the -inversion with no warning anywhere in the file. - -One sentence is enough: - -> The read must precede any free: `c2pa_free` on an untracked handle overwrites -> the slot with its own untracked-pointer message, which carries a pre-consume -> tag and would invert the decision below. - -## 3. Confirm the changed test is green for the right reason - -`test_context_build_null_return_frees_builder` loses its explicit -`c2pa_error_set_last(b"UntrackedPointer: ...")` line. That test needs the -retained branch to fire, which needs a pre-consume tag present at the moment the -failure is read. - -With the sentinel now planted inside `_invoke_consume`, a mock that merely -returns `None` leaves the sentinel in place, the sentinel branch fires, -`_teardown(free_handle=False)` runs, and no free happens. The assertion should -then fail. - -Presumably the mock is now built with `_fail_with_native_error(b"UntrackedPointer: ...")`, -which restores the tag from inside the call rather than before it. Worth -confirming that is what landed, because a test that passes for the wrong reason -here is worse than one that fails: it would be asserting the retained branch -while actually exercising the consumed one. - ---- - -## What already holds - -The sentinel is planted immediately before `ffi_call`, inside `_invoke_consume`, -with nothing between them, on the thread that makes the call. That is the correct -placement and the thread-local slot means it cannot disturb any other worker. - -`_setup_function(_lib.c2pa_error_set_last, [ctypes.c_char_p], ctypes.c_int)` -supplies the explicit argument and return types, which was the one open check. -The return value itself needs no guard: minus one is returned only for a null -pointer, so any non-null sentinel returns zero. - -Changing the final fallback from `_release_handle()` to -`_teardown(free_handle=False)` removes the guarded free from the ambiguous path -entirely. That is the more important half of this pull request, and it holds even -if the sentinel comparison in item 1 never matches. From 0edcb35fff348089fc7564d815d22084caaadc99 Mon Sep 17 00:00:00 2001 From: Tania Mathern Date: Tue, 25 Aug 2026 10:35:44 -0700 Subject: [PATCH 06/15] fix: Clean up comemnts --- src/c2pa/c2pa.py | 98 ++++++++++++++++++++---------------------------- 1 file changed, 41 insertions(+), 57 deletions(-) diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 6e0fe836..035335c9 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -274,21 +274,18 @@ def __init__(self): def _lock(self): """Return this resource's operation lock. - Reentrant because CPython can run a finalizer at any bytecode + Reentrant because it is possible to run a finalizer at any bytecode boundary, including inside a region this thread has already locked, and because a consuming call tears the handle down from inside the - locked region (_invoke_consume, _raise_consume_failure). + locked region. - Falls back to a fresh lock when the attribute is missing: an object - whose __init__ raised before the assignment is still finalized, and - __del__ must not raise. + Falls back to a fresh lock when the attribute is missing. Never hold this across a native call that drives stream callbacks (construction, resource_to_stream, the Builder stream methods, signing). Those calls release the GIL and re-enter caller-supplied - Python, which may call back into this API on another thread; holding - the lock across them deadlocks. Only calls that touch no callbacks - are serialized here, and no path holds two of these locks at once. + Python, which may call back into this API on another thread. + Only calls that touch no callbacks are serialized here. """ lock = getattr(self, '_op_lock', None) if lock is None: @@ -305,10 +302,9 @@ def _native_call(self): and forth to native layers. Calls that pass a Stream to the native library run caller-supplied - callbacks, so the lock cannot be held across them: the callback may - re-enter this API on another thread and deadlock. Instead the call is - counted as in flight, and a teardown arriving meanwhile records its - intent rather than freeing. The last caller out performs the free. + callbacks, so the lock cannot be held across them. Instead the call + is counted as in flight, and a teardown arriving meanwhile records + its intent rather than freeing. The last caller out performs the free. The resource is marked closed as soon as the teardown is recorded, so a caller that closed it cannot keep using it while the free is @@ -326,9 +322,10 @@ def _native_call(self): if self._inflight == 0 else None) if pending is not None: self._pending_teardown = None - # Released the lock before the free: _teardown takes it again, - # and keeping the two acquisitions separate means the counter - # update is never held across the release work. + # Released the lock before the free: + # _teardown takes it again, and keeping the two acquisitions + # separate means the counter update is never held across + # the release work. if pending is not None: self._teardown(pending) @@ -388,16 +385,16 @@ def _teardown(self, free_handle: bool): """Close the object: run _release, optionally free the handle, null it. free_handle=False (consumed) frees nothing, the new owner needs to free. - Holds the operation lock so the free cannot land between another + Holds the operation lock so the free cannot happen between another thread's state check and its use of the handle in a native call. """ with self._lock(): if getattr(self, '_inflight', 0) > 0: # A native call is running that re-enters caller Python and - # is still using this handle. Record the intent; whichever - # caller leaves _native_call last performs the free. Mark the - # resource closed now so it cannot be used while the free is - # pending. + # is still using this handle. Record the intent and whichever + # caller leaves _native_call last performs the free. + # Mark the resource closed now so it cannot be used + # while the free is pending. self._pending_teardown = free_handle self._lifecycle_state = LifecycleState.CLOSED return @@ -1781,10 +1778,10 @@ def __init__( check=lambda r: r != 0) if signer is not None: - # The signer's own in-flight guard: this hands its handle - # to native, so a signer.close() on another thread must - # not free it between the state check and the call. The - # guard also makes the check and the consume atomic. + # The signer's in-flight guard: + # this hands its handle to native, + # so a signer.close() on another thread must not + # free it between the state check and the call. # # _consume_no_replacement tears the signer down from # inside this region. A teardown recorded while the guard @@ -2672,9 +2669,8 @@ def _init_from_file(self, path, format_bytes, def _init_from_stream(self, stream, format_bytes, manifest_data=None): """Create a reader from a caller-supplied stream object. - The native reader reads through this stream for as long as it is - alive, so the wrapper is stored on the instance and released by - _release(). + The native reader reads through this stream as long as it's alive, + so the wrapper is stored on the instance and released by _release(). Args: stream: A stream-like object owned by the caller @@ -2714,10 +2710,7 @@ def _init_from_context(self, context, format_or_path, try: # The Context is caller-supplied and may be shared, so its handle - # needs its own in-flight guard across the native call: the - # execution_context property validates and returns the handle, and - # without the guard a context.close() on another thread could free - # it before c2pa_reader_from_context reads it. + # needs its own in-flight guard across the native call. with context._native_call(): # Adopt before the consuming call: _consume_and_swap needs an # active resource, and cleanup then owns the pointer either @@ -2765,7 +2758,7 @@ def _init_attrs(self): self._backing_file = None # Fragment streams handed to the native reader by with_fragment, - # which it keeps reading from for the rest of its life. + # which it keeps reading from for the rest of its lifecycle. self._fragment_streams = [] # Caches for manifest JSON string and parsed data. @@ -2861,8 +2854,8 @@ def with_fragment(self, format: Optional[str], stream, """ format_arg = _format_ffi_arg(_encode_format(format, "Reader")) - # The native reader keeps reading through both streams after this - # returns, so they are owned here and released by _release() rather + # The native reader keeps reading through both streams after this returns, + # so they are owned here and released by _release() rather # than at the end of a with block. main_obj = Stream(stream) frag_obj = Stream(fragment_stream) @@ -2881,8 +2874,8 @@ def with_fragment(self, format: Optional[str], stream, frag_obj.close() raise - # Replace the streams this reader owned, closing the previous ones so - # repeated calls do not accumulate them. + # Replace the streams this reader owned, + # closing the previous ones so repeated calls do not accumulate them. previous = self._own_stream self._own_stream = main_obj self._fragment_streams.append(frag_obj) @@ -2911,10 +2904,7 @@ def json(self) -> str: C2paError: If there was an error getting the JSON """ - # The state check and the handle read are one critical section: a - # finalizer on another thread frees the handle while it is still - # non-null, so a check made outside the lock says nothing about the - # handle this call goes on to pass to native code. + # Lock due to checks on native handles. with self._lock(): self._ensure_valid_state() @@ -3540,13 +3530,11 @@ def _init_from_context(self, context, json_str): if not context.is_valid: raise C2paError("Context is not valid") - # The Context is caller-supplied and may be shared, so its handle - # needs its own in-flight guard across the native call: without it a - # context.close() on another thread frees the handle between the - # is_valid check and c2pa_builder_from_context reading it. + # The Context is caller-supplied and may be shared, + # so its handle needs its own in-flight guard across + # the native call, especially for state checks. with context._native_call(): - # Adopt before the consuming call: _consume_and_swap needs an - # active resource, and cleanup then owns the pointer either way. + # Adopt before the consuming call. self._create_and_activate( lambda: _lib.c2pa_builder_from_context( context.execution_context), @@ -3886,19 +3874,15 @@ def _sign_internal( manifest_bytes_ptr = ctypes.POINTER(ctypes.c_ubyte)() try: - # _native_call covers the signing call only. The close() below is - # deliberately outside it, so the deferred teardown it records is - # performed on the way out rather than being deferred forever. + # _native_call covers the signing call only. + # The close() below is deliberately outside it, + # so the deferred teardown it records is performed + # on the way out rather than being deferred forever. with self._native_call(): if signer is not None: - # c2pa_builder_sign borrows the signer's handle, so the - # signer needs its own in-flight guard: the Builder's - # guard holds only the Builder's handle valid, and a - # signer.close() on another thread would otherwise free - # this handle mid-call. Entered inside self's guard so - # concurrent signs sharing objects acquire in one order. - # The check above is a fast-fail; this re-check inside - # the guard is the one that makes check-then-use atomic. + # Signer needs its own in-flight guard. + # Entered inside self's guard so concurrent signs + # sharing objects (Signers) acquire in one order. with signer._native_call(): result = _lib.c2pa_builder_sign( self._handle, From 9dfaf878f554b06544addd26359c28cf6b2d7c98 Mon Sep 17 00:00:00 2001 From: Tania Mathern Date: Tue, 25 Aug 2026 11:19:09 -0700 Subject: [PATCH 07/15] fix: clean up tests --- tests/test_unit_tests_threaded.py | 97 +++++++++++++------------------ 1 file changed, 40 insertions(+), 57 deletions(-) diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index 0b9b4f95..5cbccea2 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -3045,8 +3045,8 @@ class TestManagedResourceLockDeadlock(unittest.TestCase): """Tests for the operation lock that serializes native calls against teardown. - Every join here is bounded: a deadlock must fail the test, not hang the - suite. + Every join here is bounded: + A deadlock must fail the test when timing out, not hang the suite. """ JOIN_TIMEOUT = 30 @@ -3072,10 +3072,8 @@ def _join_all(self, threads, what): what, self.JOIN_TIMEOUT)) def _run_isolated(self, body, timeout=180): - """Run body in a subprocess and return it. - - A segfault kills the interpreter, so a crash cannot be asserted on - in-process: it would take the test runner with it. + """Run body in a subprocess and return it, + so that crashes can be caught and do not crash the suite itself. """ source = textwrap.dedent(body) return subprocess.run( @@ -3087,9 +3085,6 @@ def _run_isolated(self, body, timeout=180): def test_json_racing_finalizer_does_not_crash(self): """Readers used on one thread while others are collected. - - Without the lock this segfaults inside c2pa_reader_json: the - finalizer frees the handle between the state check and the call. """ result = self._run_isolated(""" import sys, io, gc, random, threading, time @@ -3146,8 +3141,8 @@ def worker(): def test_finalizer_inside_locked_operation(self): """A finalizer can run at any bytecode boundary, including inside a - region this same thread has locked. A non-reentrant lock deadlocks - here; RLock does not. + region this same thread has locked. + A non-reentrant lock deadlocks here, but RLock does not. """ resource = _ConcreteResource() resource._activate(0x51000) @@ -3240,7 +3235,7 @@ def test_consume_failure_teardown_does_not_deadlock(self): operation, re-entering the lock on the same thread. with_fragment on a JPEG returns NotSupported, which routes through - _raise_consume_failure. + _raise_consume_failure (on purpose). """ data = self.image_bytes errors = [] @@ -3264,8 +3259,8 @@ def body(): self.assertEqual(errors, []) def test_close_during_sign_does_not_deadlock(self): - """_sign_internal calls self.close() inside its own try block, so - signing re-enters the lock on the signing thread. + """_sign_internal calls self.close() inside its own try block, + so signing re-enters the lock on the signing thread. """ certs = self.certs key = self.private_key @@ -3302,10 +3297,10 @@ def body(): self.assertEqual(errors, []) def test_stream_callback_reentering_api_does_not_deadlock(self): - """Construction drives caller-supplied stream callbacks, and a caller - may legitimately call back into the API from one. + """Construction drives caller-supplied stream callbacks, + and a caller may call back into the API from one. - This passes only because construction does not hold the lock. + This passes because construction does not hold the lock. """ data = self.image_bytes other = Reader("image/jpeg", io.BytesIO(data)) @@ -3333,11 +3328,11 @@ def body(): other.close() def test_stream_callback_blocking_on_other_thread_does_not_deadlock(self): - """The adversarial case: a stream callback that blocks on another - thread which touches the same object. + """A stream callback that blocks on another thread + which touches the same object. A lock held across construction deadlocks here, whether it is global - or per-object. This is the test that pins the scoping decision. + or per-object. """ data = self.image_bytes target = Reader("image/jpeg", io.BytesIO(data)) @@ -3374,9 +3369,7 @@ def body(): def test_no_nested_op_locks(self): """No code path may hold two resources' operation locks at once. - - That property, not the tests above, is what makes the design - deadlock-free: with only one lock ever held, no cycle can form. + With only one lock ever held, no cycle can form here. """ data = self.image_bytes held = threading.local() @@ -3505,7 +3498,8 @@ def write(self, buffer): self.assertEqual(reader._lifecycle_state, LifecycleState.CLOSED) def test_cross_thread_close_during_callback_defers_free(self): - """Same race, with the close arriving from another thread.""" + """A close() from inside a stream callback must not free the handle + the native call is still using.""" freed = self._counted_free() reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) uri = self._thumbnail_uri(reader) @@ -3558,8 +3552,8 @@ def write(self, buffer): self.assertIsNone(reader._handle) def test_use_after_deferred_close_is_rejected(self): - """Deferring must not leave the resource usable: the free is pending, - so the handle is about to go away.""" + """Deferring must not leave the resource usable: + the free is pending, so the handle is about to go away.""" reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) uri = self._thumbnail_uri(reader) states = [] @@ -3584,8 +3578,8 @@ def write(self, buffer): self.assertEqual(states[1], "json rejected") def test_exception_from_callback_still_frees(self): - """An exception unwinding through the native call must not strand the - in-flight counter, or the handle is never freed.""" + """An exception unwinding through the native call must not + leave the inflight-handler hanging.""" freed = self._counted_free() reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) uri = self._thumbnail_uri(reader) @@ -3635,8 +3629,8 @@ def write(self, buffer): self.assertIsNone(reader._handle) def test_release_raising_during_deferred_teardown_does_not_leak(self): - """The deferred free survives a failing _release: the handle must - still be freed.""" + """The deferred free survives a failing _release: + the handle must still be freed.""" freed = self._counted_free() reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) uri = self._thumbnail_uri(reader) @@ -3659,8 +3653,9 @@ def write(self, buffer): self.assertEqual(len(freed), 1, "handle leaked when _release raised") def test_concurrent_closes_during_callback_free_once(self): - """Many threads closing while one native call is in flight must - produce exactly one free.""" + """Many threads closing while one native call is in flight + must produce exactly one free (avoid double-frees, + or freeing something the object wouldn't own).""" freed = self._counted_free() reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) uri = self._thumbnail_uri(reader) @@ -3692,8 +3687,8 @@ def closer(): self.assertEqual(reader._inflight, 0) def test_sign_with_internal_close_frees_once(self): - """_sign_internal closes the Builder inside its own try, so the close - defers and the free happens on the way out.""" + """_sign_internal closes the Builder inside its own try, + so the close defers and the free happens on the way out.""" freed = self._counted_free() signer_info = C2paSignerInfo( alg=b"es256", @@ -3722,9 +3717,8 @@ def test_sign_with_internal_close_frees_once(self): io.BytesIO(self.image_bytes), io.BytesIO()) def test_class_a_construction_is_not_guarded(self): - """Construction is deliberately unguarded: no external caller holds a - reference yet, and guarding it would reintroduce the deadlock where a - stream callback re-enters the API.""" + """Construction is unguarded: no external caller holds a reference yet. + """ entered = [] real = ManagedResource._native_call @@ -3743,12 +3737,8 @@ def recording(resource): "reintroduces the callback deadlock") def test_every_callback_running_method_is_guarded(self): - """Coverage check: every method that hands a Stream to the native - library must be guarded, except the three construction paths. - - A method missed here keeps the use-after-free, and the symptom is a - rare segfault rather than a failing test, so this is checked - mechanically rather than by eye. + """Every method that hands a Stream to the native lib must be guarded, + except the construction paths. """ source = inspect.getsource(sys.modules[Reader.__module__]) lines = source.split("\n") @@ -3798,14 +3788,11 @@ def test_every_callback_running_method_is_guarded(self): unguarded)) def test_every_borrowed_handle_is_guarded(self): - """Coverage check: when a method hands a *second* object's handle to - the native library, that object needs its own _native_call() guard. - - test_every_callback_running_method_is_guarded only asks whether the - string "_native_call()" appears in the method body, which cannot - express *whose* handle is guarded. A method that guards self while - passing signer._handle to native passes that check and still has the - use-after-free, so the ownership is checked structurally here. + """When a method hands a second object's handle to the native library, + that object needs its own _native_call() guard. + + This can happen in callbacks, where you can't express whose handle + is the one needing attention. """ module = sys.modules[Reader.__module__] tree = ast.parse(inspect.getsource(module)) @@ -3841,12 +3828,8 @@ def borrowed_in_call(call): return names def locally_owned(method): - """Names bound to an object this method itself constructed. - - A resource created inside the method never escapes to another + """A resource created inside the method never escapes to another thread, so nothing can close it mid-call and it needs no guard. - Only handles reaching the method from outside (parameters, - attributes) are exposed to a concurrent teardown. """ owned = set() for node in ast.walk(method): @@ -3933,7 +3916,7 @@ def test_close_during_concurrent_sign_does_not_crash(self): """Rotate a shared signer while other threads sign with it. Runs in a subprocess: the failure mode is a segfault, which would - take the test runner down with it rather than reporting a failure. + take the test runner down with it otherwise. """ source = textwrap.dedent(""" import io, os, sys, threading From e80b0397c4ba56829618284f6fafba7cb0ff20a5 Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 18:40:53 -0700 Subject: [PATCH 08/15] fix: with_fragment has issues too --- src/c2pa/c2pa.py | 13 ++++++++++++- tests/perf/baseline.json | 9 +++++++-- tests/perf/scenarios.py | 25 +++++++++++++++++++++++++ tests/test_unit_tests.py | 34 ++++++++++++++++++++++++++++++++++ 4 files changed, 78 insertions(+), 3 deletions(-) diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 035335c9..d762e285 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -2876,14 +2876,25 @@ def with_fragment(self, format: Optional[str], stream, # Replace the streams this reader owned, # closing the previous ones so repeated calls do not accumulate them. + # Only the current fragment is retained: the native reader does not + # read a superseded one back, and each wrapper held open pins a native + # stream, its callbacks and the caller's buffer. previous = self._own_stream + previous_fragments = self._fragment_streams self._own_stream = main_obj - self._fragment_streams.append(frag_obj) + self._fragment_streams = [frag_obj] if previous is not None and previous is not main_obj: try: previous.close() except Exception: logger.warning("Failed to close previous Reader stream") + for fragment in previous_fragments: + if fragment is frag_obj: + continue + try: + fragment.close() + except Exception: + logger.warning("Failed to close Reader fragment stream") # Invalidate caches: processing a new BMFF fragment updates the native # reader's state, which can change the manifest data it returns. diff --git a/tests/perf/baseline.json b/tests/perf/baseline.json index c151efe5..feb51bd0 100644 --- a/tests/perf/baseline.json +++ b/tests/perf/baseline.json @@ -2,8 +2,8 @@ "_meta": { "memray_version": "1.19.3", "python_version": "3.12.13", - "c2pa_native_version": "c2pa-v0.90.0", - "iterations": 200, + "c2pa_native_version": "c2pa-v0.90.15", + "iterations": 100, "perf_env": "python-3.12-slim", "arch": "aarch64" }, @@ -296,5 +296,10 @@ "peak_bytes": 3681161, "leaked_bytes": 3350287, "total_allocations": 672537 + }, + "reader_with_fragment_repeated": { + "peak_bytes": 3803564, + "leaked_bytes": 3381191, + "total_allocations": 966288 } } \ No newline at end of file diff --git a/tests/perf/scenarios.py b/tests/perf/scenarios.py index 23300aed..518c8f97 100644 --- a/tests/perf/scenarios.py +++ b/tests/perf/scenarios.py @@ -524,6 +524,30 @@ def scenario_reader_with_fragment_swap(iterations: int = 100) -> None: reader.close() +def scenario_reader_with_fragment_repeated(iterations: int = 100) -> None: + """Loop Reader.with_fragment() against a SINGLE long-lived Reader. + + The Reader is built outside the loop on purpose. Every other fragment + scenario constructs one per iteration and closes it, which releases the + streams each time round and so cannot show anything retained across calls. + Only repeated calls on one instance expose a fragment stream that is kept + instead of released, and each one held open pins a native C2paStream, its + four ctypes callbacks and the caller's buffer. + """ + init_bytes = DASH_INIT_MP4.read_bytes() + fragment_bytes = DASH_FRAGMENT.read_bytes() + reader = Reader("video/mp4", io.BytesIO(init_bytes)) + try: + for _ in _iterate(iterations): + reader.with_fragment( + "video/mp4", + io.BytesIO(init_bytes), + io.BytesIO(fragment_bytes), + ) + finally: + reader.close() + + def scenario_builder_from_archive_roundtrip(iterations: int = 100) -> None: """Loop Builder.from_archive() itself (context-less alternate constructor), then sign. Regression guard for the classmethod's native-handle wrapping. @@ -1370,6 +1394,7 @@ def scenario_fork_stream_cleanup(iterations: int = 100) -> None: "builder_from_archive_roundtrip": scenario_builder_from_archive_roundtrip, "builder_with_archive_swap": scenario_builder_with_archive_swap, "reader_with_fragment_swap": scenario_reader_with_fragment_swap, + "reader_with_fragment_repeated": scenario_reader_with_fragment_repeated, "with_fragment_pre_consume_rejection": scenario_reader_with_fragment_pre_consume_rejection, "with_archive_post_consume_failure": diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index 4bff6dbb..5bf091d5 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -8786,6 +8786,40 @@ def test_with_fragment_pre_consume_rejection_does_not_leak(self): self.assertTrue(reader.json()) reader.close() + def test_repeated_with_fragment_does_not_accumulate_streams(self): + """Repeated calls on one Reader must not pile up fragment streams. + + Each retained wrapper pins a native C2paStream, four ctypes callback + trampolines and the caller's buffer, so an unbounded list grows the + process by tens of megabytes over a long-lived Reader. Every other + fragment test builds a fresh Reader per call, which never accumulates. + """ + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + self.addCleanup(reader.close) + + superseded = [] + for _ in range(25): + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + reader.with_fragment("video/mp4", init, frag) + self.assertLessEqual( + len(reader._fragment_streams), 1, + "fragment streams accumulated across repeated calls") + superseded.append(reader._fragment_streams[-1]) + + # Dropping the reference is not enough: the native stream is only + # released by close(), so every superseded wrapper must be closed. + self.assertTrue( + all(s.closed for s in superseded[:-1]), + "a superseded fragment stream was dropped without being closed") + + # The reader still works on the fragment it currently holds. + self.assertTrue(reader.json()) + def test_with_archive_post_consume_failure_consumes_handle(self): # Ownership taken, then the operation failed: # The handle is gone, so close() must not free it again. From 099c82ec2dee50a4a888d12acf7a05b742977b22 Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:09:01 -0700 Subject: [PATCH 09/15] fix: Reorder to avoid potential deadlock --- src/c2pa/c2pa.py | 33 ++++++--- tests/test_unit_tests_threaded.py | 108 ++++++++++++++++++++++++++++++ 2 files changed, 133 insertions(+), 8 deletions(-) diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index d762e285..4ef44d12 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -286,7 +286,17 @@ def _lock(self): signing). Those calls release the GIL and re-enter caller-supplied Python, which may call back into this API on another thread. Only calls that touch no callbacks are serialized here. + + Raises in a forked child rather than returning the lock. + A child inherits this lock in whatever state it had at fork(), + and a thread holding it does not exist in the child to release it, + so acquiring it there waits and waits and waits. + The child's copy is unusable for the same reason a closed resource is, + and reports the same error. """ + if is_foreign_process(self): + raise C2paError(f"{type(self).__name__} is closed") + lock = getattr(self, '_op_lock', None) if lock is None: lock = threading.RLock() @@ -387,23 +397,30 @@ def _teardown(self, free_handle: bool): Holds the operation lock so the free cannot happen between another thread's state check and its use of the handle in a native call. + + The forked-child case is handled before the lock is taken, because + _lock() refuses in a child: this path has to finish rather than report + an error, so it cannot rely on acquiring. """ + if is_foreign_process(self): + # The parent owns the handle and frees its own copy. Mark this one + # closed and drop the pointer so the child cannot use or free it. + self._handle = None + self._lifecycle_state = LifecycleState.CLOSED + return + with self._lock(): if getattr(self, '_inflight', 0) > 0: - # A native call is running that re-enters caller Python and - # is still using this handle. Record the intent and whichever - # caller leaves _native_call last performs the free. + # A native call is running that re-enters calling non-native code + # and is still using this handle. + # Record the intent and whichever caller leaves + # _native_call last performs the free. # Mark the resource closed now so it cannot be used # while the free is pending. self._pending_teardown = free_handle self._lifecycle_state = LifecycleState.CLOSED return - if is_foreign_process(self): - self._handle = None - self._lifecycle_state = LifecycleState.CLOSED - return - self._lifecycle_state = LifecycleState.CLOSED self._safe_release() diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index 5cbccea2..c7dca601 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -184,6 +184,114 @@ def test_foreign_pid_close_marks_closed(self): self.assertFalse(obj._initialized) +class TestForkedChildDoesNotDeadlock(unittest.TestCase): + """A forked child must never block on a lock the parent held at fork(). + + Locking a resource for the duration of an operation means a child that + forks while some thread holds that lock inherits it locked, with the owner + thread gone. Anything in the child that acquires it waits forever. + + The failure mode is a hang: each operation runs on a worker thread and + is joined with a timeout: a test that called it directly would hang the + runner instead of failing. + """ + + _TIMEOUT = 5.0 + + def _foreign_reader_with_lock_held(self): + """A Reader in the state a forked child inherits: + lock held by another thread, and stamped with a PID other than this process's. + """ + with open(DEFAULT_TEST_FILE, "rb") as asset: + reader = Reader("image/jpeg", asset) + holding = threading.Event() + release = threading.Event() + + def hold_the_lock(): + with reader._lock(): + holding.set() + release.wait(30) + + holder = threading.Thread(target=hold_the_lock, daemon=True) + holder.start() + self.assertTrue(holding.wait(self._TIMEOUT), + "helper thread never acquired the lock") + self.addCleanup(holder.join, self._TIMEOUT) + self.addCleanup(release.set) + + reader._owner_pid = os.getpid() + 1 + return reader + + def _run_with_timeout(self, operation): + """Run operation on a worker; return 'ok', the exception, or None if it + was still running when the timeout expired.""" + result = {} + + def run(): + try: + operation() + result["outcome"] = "ok" + except BaseException as e: # noqa: BLE001 - asserted on below + result["outcome"] = e + + worker = threading.Thread(target=run, daemon=True) + worker.start() + worker.join(self._TIMEOUT) + return result.get("outcome") + + def test_locked_read_raises_instead_of_blocking(self): + reader = self._foreign_reader_with_lock_held() + outcome = self._run_with_timeout(reader.json) + self.assertIsNotNone( + outcome, "json() blocked on a lock inherited from the parent") + self.assertIsInstance(outcome, Error) + + def test_native_call_path_raises_instead_of_blocking(self): + reader = self._foreign_reader_with_lock_held() + outcome = self._run_with_timeout( + lambda: reader.resource_to_stream("any-uri", io.BytesIO())) + self.assertIsNotNone( + outcome, + "resource_to_stream() blocked on a lock inherited from the parent") + self.assertIsInstance(outcome, Error) + + def test_close_still_completes(self): + reader = self._foreign_reader_with_lock_held() + self.assertEqual(self._run_with_timeout(reader.close), "ok", + "close() must neither block nor raise") + + def test_teardown_still_completes(self): + # Cleanup has to finish, not report an error. + reader = self._foreign_reader_with_lock_held() + self.assertEqual( + self._run_with_timeout( + lambda: reader._teardown(free_handle=True)), "ok", + "_teardown() must neither block nor raise") + self.assertEqual(reader._lifecycle_state, LifecycleState.CLOSED) + self.assertIsNone(reader._handle) + + def test_parent_copy_unaffected(self): + """The child closing its copy must leave the parent's usable. + """ + with open(DEFAULT_TEST_FILE, "rb") as asset: + reader = Reader("image/jpeg", asset) + self.addCleanup(reader.close) + before = reader.json() + + pid = os.fork() + if pid == 0: + try: + reader.close() + os._exit(0) + except BaseException: + os._exit(1) + _, status = os.waitpid(pid, 0) + + self.assertEqual(status >> 8, 0, "child could not close its own copy") + self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) + self.assertEqual(reader.json(), before) + + class TestHelpers(unittest.TestCase): def test_record_and_detect_own_pid(self): From a59f06bb4d17e14caabb2c683a286eabc23e7bbd Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:19:19 -0700 Subject: [PATCH 10/15] Update iterations count in baseline.json --- tests/perf/baseline.json | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/perf/baseline.json b/tests/perf/baseline.json index feb51bd0..7ce38df4 100644 --- a/tests/perf/baseline.json +++ b/tests/perf/baseline.json @@ -3,7 +3,7 @@ "memray_version": "1.19.3", "python_version": "3.12.13", "c2pa_native_version": "c2pa-v0.90.15", - "iterations": 100, + "iterations": 200, "perf_env": "python-3.12-slim", "arch": "aarch64" }, @@ -302,4 +302,4 @@ "leaked_bytes": 3381191, "total_allocations": 966288 } -} \ No newline at end of file +} From aa605af0c6c929d016f4d9fae466ed7247b76845 Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:24:03 -0700 Subject: [PATCH 11/15] fix: Error handling --- src/c2pa/c2pa.py | 7 ++- tests/test_unit_tests.py | 93 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 99 insertions(+), 1 deletion(-) diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 4ef44d12..83d9593b 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -517,7 +517,12 @@ def _swap_handle(self, new_handle): # Errors set by native lib, hinting at the cause of the error # These errors here means the pointer got somehow rejected by the lib, # so it is still ours to deal with. - _PRE_CONSUME_ERROR_TAGS = ("UntrackedPointer:", "WrongPointerType:") + _PRE_CONSUME_ERROR_TAGS = ( + "UntrackedPointer:", + "WrongPointerType:", + "NullParameter:", + "InvalidBufferSize:", + ) def _invoke_consume(self, ffi_call, error_message): """Run an FFI call that consumes this handle, returning its raw result. diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index 5bf091d5..d8c09bdd 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -8786,6 +8786,99 @@ def test_with_fragment_pre_consume_rejection_does_not_leak(self): self.assertTrue(reader.json()) reader.close() + def _reader_from_context(self): + """A Reader holding a fresh native handle and nothing else. + + Built through the FFI so the consuming call can be + set up with one deliberately invalid argument. + """ + context = Context() + self.addCleanup(context.close) + reader = Reader.__new__(Reader) + ManagedResource.__init__(reader) + reader._init_attrs() + with context._native_call(): + reader._create_and_activate( + lambda: c2pa_module._lib.c2pa_reader_from_context( + context.execution_context), + "Failed to create reader: {}") + return reader + + def test_null_parameter_rejection_retains_the_handle(self): + """A null argument is rejected before the reader is untracked. + Ownership never transferred, so the handle is still ours to free. + Treating it as consumed leaks one reader per call. + """ + reader = self._reader_from_context() + handle = reader._handle + freed = self._instrument_frees() + + with self.assertRaises(Error) as caught: + with reader._native_call(): + reader._consume_and_swap( + lambda h: c2pa_module._lib.c2pa_reader_with_stream( + h, b"image/jpeg", None), + "Failed to configure reader: {}") + + self.assertIn("NullParameter", str(caught.exception)) + self.assertIsNotNone(reader._handle, "the retained handle was dropped") + self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) + + reader.close() + self.assertEqual( + self._free_count(freed, handle), 1, + "a handle the native side never took was leaked") + + def test_invalid_buffer_size_rejection_retains_the_handle(self): + """A zero-length manifest buffer is rejected before the untrack.. + """ + reader = self._reader_from_context() + handle = reader._handle + freed = self._instrument_frees() + empty = (ctypes.c_ubyte * 4)() + + with Stream(io.BytesIO(b"abc")) as stream_obj: + with self.assertRaises(Error) as caught: + with reader._native_call(): + reader._consume_and_swap( + lambda h: ( + c2pa_module._lib + .c2pa_reader_with_manifest_data_and_stream( + h, b"image/jpeg", stream_obj._stream, + empty, 0) + ), + "Failed to configure reader: {}") + + self.assertIn("InvalidBufferSize", str(caught.exception)) + self.assertIsNotNone(reader._handle, "the retained handle was dropped") + self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) + + reader.close() + self.assertEqual( + self._free_count(freed, handle), 1, + "a handle the native side never took was leaked") + + def test_repeated_rejections_do_not_accumulate_handles(self): + """Every rejected call must give its handle back, not just the first. + """ + handles = [] + freed = self._instrument_frees() + + for _ in range(10): + reader = self._reader_from_context() + handles.append(reader._handle) + with self.assertRaises(Error): + with reader._native_call(): + reader._consume_and_swap( + lambda h: c2pa_module._lib.c2pa_reader_with_stream( + h, b"image/jpeg", None), + "Failed to configure reader: {}") + reader.close() + + leaked = [h for h in handles if self._free_count(freed, h) == 0] + self.assertEqual( + leaked, [], f"{len(leaked)} of {len(handles)} handles leaked") + def test_repeated_with_fragment_does_not_accumulate_streams(self): """Repeated calls on one Reader must not pile up fragment streams. From ab110f73386a1883273d9676ba33ccf49bd07f13 Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:49:04 -0700 Subject: [PATCH 12/15] fix: Rewrite some threaded tests to avoid multifork issues --- tests/test_unit_tests_threaded.py | 63 +++++++++++++++++++++---------- 1 file changed, 44 insertions(+), 19 deletions(-) diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index c7dca601..85097321 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -272,24 +272,49 @@ def test_teardown_still_completes(self): def test_parent_copy_unaffected(self): """The child closing its copy must leave the parent's usable. + + Runs in a subprocess so that the fork happens in a single-threaded + process. Operations that reach the network, such as reading an asset + with a remote manifest, start background native threads that outlive + the object that triggered them, and forking a multi-threaded process + can lead to issues. """ - with open(DEFAULT_TEST_FILE, "rb") as asset: - reader = Reader("image/jpeg", asset) - self.addCleanup(reader.close) - before = reader.json() + source = textwrap.dedent(""" + import os, sys + from c2pa import Reader + from c2pa.c2pa import LifecycleState - pid = os.fork() - if pid == 0: - try: - reader.close() - os._exit(0) - except BaseException: - os._exit(1) - _, status = os.waitpid(pid, 0) + asset_path = sys.argv[1] + with open(asset_path, "rb") as asset: + reader = Reader("image/jpeg", asset) + before = reader.json() - self.assertEqual(status >> 8, 0, "child could not close its own copy") - self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) - self.assertEqual(reader.json(), before) + pid = os.fork() + if pid == 0: + try: + reader.close() + os._exit(0) + except BaseException: + os._exit(1) + _, status = os.waitpid(pid, 0) + + assert status >> 8 == 0, "child could not close its own copy" + assert reader._lifecycle_state == LifecycleState.ACTIVE + assert reader.json() == before + reader.close() + print("OK") + """) + + result = subprocess.run( + [sys.executable, "-c", source, DEFAULT_TEST_FILE], + capture_output=True, text=True, timeout=120) + + self.assertEqual( + result.returncode, 0, + "parent copy was affected by the child (rc={}):\n{}".format( + result.returncode, result.stderr[-2000:])) + self.assertIn("OK", result.stdout) + self.assertNotIn("DeprecationWarning", result.stderr) class TestHelpers(unittest.TestCase): @@ -813,12 +838,12 @@ def setUp(self): with open(os.path.join(self.data_dir, "es256_private.key"), "rb") as key_file: self.key = key_file.read() - # Create a local Es256 signer with certs and a timestamp server + # Create a local Es256 signer with certs and no timestamp server. self.signer_info = C2paSignerInfo( alg=b"es256", sign_cert=self.certs, private_key=self.key, - ta_url=b"http://timestamp.digicert.com" + ta_url=None ) self.signer = Signer.from_info(self.signer_info) @@ -3377,7 +3402,7 @@ def test_close_during_sign_does_not_deadlock(self): alg=b"es256", sign_cert=certs, private_key=key, - ta_url=b"http://timestamp.digicert.com", + ta_url=None, ) manifest = { "claim_generator": "python_test", @@ -3802,7 +3827,7 @@ def test_sign_with_internal_close_frees_once(self): alg=b"es256", sign_cert=self.certs, private_key=self.private_key, - ta_url=b"http://timestamp.digicert.com", + ta_url=None, ) manifest = { "claim_generator": "python_test", From fc630b80c5cf7c9cae7292f40a70dabd78a4523b Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 20:59:26 -0700 Subject: [PATCH 13/15] fix: Docs --- docs/native-resources-management.md | 44 +++++++++++++++++++++-------- 1 file changed, 33 insertions(+), 11 deletions(-) diff --git a/docs/native-resources-management.md b/docs/native-resources-management.md index 1cf057f0..2ab550a6 100644 --- a/docs/native-resources-management.md +++ b/docs/native-resources-management.md @@ -106,15 +106,27 @@ Therefore, the managed resources have the following principles: ### Double-free risk mitigations -Three distinct risks. Two have a mechanism in this layer; the third is the caller's to synchronize: +Three distinct risks, each with its own mechanism in this layer: | Hazard | Covered by | How | | --- | --- | --- | -| Freeing a pointer a consuming call already took (single flow) | `_swap_handle` / `_teardown(free_handle=False)` triage | The consumed pointer is abandoned, never freed. The retained-vs-consumed decision reads the native error tag (`UntrackedPointer:` / `WrongPointerType:` mean not taken). | +| Freeing a pointer a consuming call already took (single flow) | `_swap_handle` / `_teardown(free_handle=False)` triage | The consumed pointer is abandoned, never freed. The retained-vs-consumed decision reads the native error tag (`UntrackedPointer:` / `WrongPointerType:` / `NullParameter:` / `InvalidBufferSize:` mean not taken). | | A forked child freeing a pointer its parent owns | PID stamp (`record_owner_pid` / `is_foreign_process`) | Cleanup in a process that did not allocate the pointer nulls the handle and marks `CLOSED` without freeing (see [Fork safety](#fork-safety)). | -| Two **threads** in one process racing frees on distinct objects, where the allocator recycles a just-freed address | Not covered here | `ManagedResource` has no lock and no thread stamping. The PID stamp cannot see it: sibling threads share a PID. Safety for genuinely shared handles must come from the caller's own synchronization or from the native registry, not this layer. | +| Two **threads** racing a `close()` against an in-flight native call on the same object, where the allocator recycles a just-freed address | `_op_lock` / `_native_call()` / `_pending_teardown` | A close arriving while a native call is in flight is recorded rather than applied. The last caller to leave `_native_call()` performs the deferred free (see [Locking and in-flight tracking](#locking-and-in-flight-tracking)). | -The PID stamp is fork-only: it compares process IDs, and two threads in the same process always match. Sharing one `ManagedResource` instance across threads without external synchronization is outside what this layer protects against. +The PID stamp is fork-only: it compares process IDs, and two threads in the same process always match on that PID. Sharing one `ManagedResource` instance across threads still needs locks: nothing here protects two threads racing on genuinely distinct objects that happen to share an allocator. + +## Locking and in-flight tracking + +Each `ManagedResource` holds a reentrant lock, `_op_lock`, and a counter, `_inflight`, that together serialize teardown against concurrent use from other threads. + +A lock (Python's `threading.Lock`) can be acquired once, and a second `acquire()` from the same thread on that lock blocks forever, waiting on a lock that thread itself is holding. A reentrant lock (`threading.RLock`) tracks which thread holds it and how many times: the owning thread can acquire it again without blocking, and the lock is only released once that thread has released it the same number of times it acquired it. A different thread still blocks until the owner releases fully. + +`_op_lock` is an `RLock` rather than a plain `Lock` for two reasons specific to this code. First, a finalizer (`__del__`) can run at any bytecode boundary — including one in the middle of a method that has already acquired the lock on this same thread — so `__del__` calling back into locked code must not deadlock against itself. Second, a consuming call tears the handle down from inside the locked region it is already holding: `_teardown()` is called while `_op_lock` is held, and it needs to acquire the same lock again rather than re-entering as a different, blocked acquisition. `_lock()` returns it, except in a forked child: there it raises `C2paError` immediately rather than blocking, because the thread that might hold the lock at fork time does not exist in the child to release it, and waiting on it would hang forever (see [Fork safety](#fork-safety)). + +The lock is never held across a native call that drives a stream callback: construction, `resource_to_stream`, the Builder stream methods, and signing all release the GIL and call back into caller-supplied Python, which may itself call into this API on another thread. Holding `_op_lock` there would deadlock against that reentry. Those calls go through `_native_call()` instead: a context manager that increments `_inflight` under the lock, yields to run the native call unlocked, then decrements `_inflight` on the way out. If `_teardown()` runs while a call is in flight, it records the requested `free_handle` value in `_pending_teardown` and marks the resource `CLOSED` immediately, so no other caller can start using it, but defers the actual free. The last `_native_call()` to exit picks up `_pending_teardown` and runs `_teardown()` for real. + +`Context.__init__` wraps the signer hand-off in `signer._native_call()`, so a `signer.close()` on another thread cannot free the handle between the state check and the consuming call. `Builder._sign_internal` wraps the sign call in `self._native_call()` and, when an explicit `Signer` is passed, nests `signer._native_call()` inside it in that fixed order, so two concurrent `sign()` calls sharing one `Signer` cannot deadlock by acquiring the two locks in opposite orders. The Builder's `close()` after signing runs outside its own `_native_call()` block, so a teardown deferred during the call still executes once the call returns. ## Guarantees provided by ManagedResource @@ -317,6 +329,8 @@ While `ACTIVE`, callers can use `.add_ingredient()`, `.add_action()`, etc. repea The native sign call borrows the builder's pointer rather than taking ownership of it, so `Builder` never marks it consumed and the pointer is freed normally through `c2pa_free`. The close enforces single use; it is not a memory-management requirement. +The sign call runs inside `self._native_call()`, nesting `signer._native_call()` when the caller passes an explicit `Signer`, and `close()` runs after that block exits (see [Locking and in-flight tracking](#locking-and-in-flight-tracking) for why the order matters and what it protects against). + ## Ownership transfer Some operations transfer a native pointer from one object to another. When this happens, the original object must stop managing the pointer (e.g. so it is not freed twice). @@ -338,6 +352,8 @@ sequenceDiagram C->>X: Context(settings, signer) X->>B: with _NativeBuilder() (owns the builder, close() frees it on any failure) X->>S: _ensure_valid_state() + X->>S: enter _native_call() + Note right of S: Pins the Signer active for the duration:
a close() on another thread now waits
instead of freeing the handle mid-transfer X->>X: copy signer._callback_cb to _signer_callback_cb Note right of X: Pin the callback first:
the Signer is about to be consumed X->>S: _consume_no_replacement(set_signer) @@ -346,12 +362,13 @@ sequenceDiagram alt status 0 (success) S->>S: _teardown(free_handle=False) Note right of S: Consumed: native took the signer - else pre-consume rejection (UntrackedPointer / WrongPointerType) + else pre-consume rejection (one of _PRE_CONSUME_ERROR_TAGS) Note right of S: Rejected before ownership moved:
Signer retained, typed error raised else other error S->>S: _teardown(free_handle=False) Note right of S: Native took it then failed and dropped it end + X->>S: exit _native_call() X->>B: _consume_into(build) B->>N: c2pa_context_builder_build(builder_ptr) @@ -362,7 +379,8 @@ sequenceDiagram Details in that sequence that are easy to get wrong: - The callback is copied to the Context *before* the transfer. A successful consume runs `_release()`, which drops the Signer's reference to the callback; a Context that copied it afterwards would be pointing at a callback nothing keeps alive. -- `set_signer` does not always take the pointer. A pre-consume rejection (`UntrackedPointer:` / `WrongPointerType:`) leaves the Signer `ACTIVE` and retained, so the triage must read the native error before deciding to close it. Treating every failure as "consumed" would close a signer the native side never took. +- The state check and the consuming call both run inside `signer._native_call()`, so a `signer.close()` racing on another thread cannot free the handle in the gap between them. If a close does arrive while the transfer is in flight, it is recorded as a pending teardown and applied once the transfer finishes (see [Locking and in-flight tracking](#locking-and-in-flight-tracking)). +- `set_signer` does not always take the pointer. A pre-consume rejection (one of `_PRE_CONSUME_ERROR_TAGS`) leaves the Signer `ACTIVE` and retained, so the triage must read the native error before deciding to close it. Treating every failure as "consumed" would close a signer the native side never took. - A `ctypes.ArgumentError` from `set_signer` is re-raised untouched by `_invoke_consume`: marshalling failed, the native function never ran, and the Signer still owns its handle. Only calls that reached native go through the consumed/retained triage. - The builder is never held as a raw local across the signer and build calls. `_NativeBuilder`'s `with` block owns it: a settings error, a retained-signer error, a build rejection, or an async interrupt all free it through `close()`, and a successful build consumes it so `close()` is then a no-op. The old raw-pointer recovery block that used to free `builder_ptr` on the un-reached-build path is gone. @@ -396,6 +414,8 @@ stateDiagram-v2 On success the object stays `ACTIVE` because the Python-side object is still valid: it has a live native pointer, its public methods still work, and callers may continue using it (e.g. reading the updated manifest or feeding in another fragment). The lifecycle state does not change because from `ManagedResource`'s perspective nothing has closed. Only the underlying native pointer has been swapped. This is different from a consumed teardown (`_teardown(free_handle=False)`), where the object transitions to `CLOSED` and becomes unusable. On the success path the old pointer must not be freed by `ManagedResource` because the native library already consumed it as part of the FFI call. The failure path is different and is covered by the triage in [`_consume_and_swap()`](#_consume_and_swap). +`Reader.with_fragment()` runs the swap inside `self._native_call()`, and keeps a `_fragment_streams` list holding the `Stream` wrapper for the current fragment. Each call to `with_fragment()` replaces that list rather than appending to it, closing the previous fragment's wrapper immediately: the native reader never reads a superseded fragment back, and each open wrapper pins a native stream, its callbacks, and the caller's buffer. + ### `_consume_and_swap()` Every call of this shape goes through one helper, which takes the FFI call as a callable and handles the outcomes: @@ -428,7 +448,7 @@ The two failure paths are indistinguishable from the return value alone. Only th | Native error | Who owns the handle | What the helper does | | --- | --- | --- | -| `UntrackedPointer:` or `WrongPointerType:` | Still ours: rejected before ownership moved | Handle kept, resource stays `ACTIVE`, typed error raised. Normal cleanup frees it later. | +| One of `_PRE_CONSUME_ERROR_TAGS` | Still ours: rejected before ownership moved | Handle kept, resource stays `ACTIVE`, typed error raised. Normal cleanup frees it later. | | Any other error | Taken, then the operation failed | `_teardown(free_handle=False)`: the native side already dropped the value, so nothing is freed here. Resource goes `CLOSED`, error typed from the native message. | | No error at all | Unknown | `_release_handle()` guarded free, the caller's message is raised with `"Unknown error"` filled in. | @@ -448,17 +468,17 @@ Three consume helpers share this triage; they differ only in what the FFI call r A consuming FFI call can fail. It may reject the borrowed pointer before taking it, or it may take ownership first and then, on a later failure, drop the value itself. -The native error message indicates which of the errors happened. A rejection carries one of the `_PRE_CONSUME_ERROR_TAGS` (`UntrackedPointer:` or `WrongPointerType:`), which means the handle was never taken and is retained. Any other error message means the native side may have taken ownership and already dropped the value. On top of those, preparing the call's own arguments can fail in Python before the native function ever runs (for example, encoding a bad value or a ctypes marshalling error other than `ArgumentError`), and that outcome is handled separately. +The native error message indicates which of the errors happened. A rejection carries one of the `_PRE_CONSUME_ERROR_TAGS`, which means the handle was never taken and is retained. Any other error message means the native side may have taken ownership and already dropped the value. On top of those, preparing the call's own arguments can fail in Python before the native function ever runs (for example, encoding a bad value or a ctypes marshalling error other than `ArgumentError`), and that outcome is handled separately. -The two settled branches each take the exact action their ownership implies. A pre-consume rejection (an error prefixed `UntrackedPointer:` or `WrongPointerType:`) means the handle is still the caller's, so it is retained and freed later by normal cleanup. Any other native error means the value is already gone, so `_teardown(free_handle=False)` runs the Python-side cleanup without freeing anything. +The two settled branches each take the exact action their ownership implies. A pre-consume rejection (one of the `_PRE_CONSUME_ERROR_TAGS`) means the handle is still the caller's, so it is retained and freed later by normal cleanup. Any other native error means the value is already gone, so `_teardown(free_handle=False)` runs the Python-side cleanup without freeing anything. Always calling the guarded free instead, even where the value is known to be gone, is tempting because a stale free looks like a harmless `-1` no-op. It is only harmless while the freed address stays unclaimed. The native registry rejects an address it no longer tracks, but once another thread allocates a fresh tracked object at that recycled address, the registry does track it again — and a stale free aimed at the old value would now find a live entry and destroy a different thread's object. The scenario is unlikely, but not unreachable: it needs a second thread inside its own FFI call, an allocator that hands back the exact address just freed, and that reuse to happen during the (narrow) window between the native drop and this free. But the window is real under concurrent use. The failure is a silent cross-thread corruption rather than a clean error, and the free is not needed in the first place on this branch. So where the value is known to be consumed, the free is skipped rather than issued and left to the registry to reject. The native error slot stays sticky: it holds whatever it last held until the next error overwrites it, and nothing clears it in between. Issuing an unneeded free would set an untracked-pointer error there that a later caller could mistake for the failure it actually asked about, so skipping the free keeps the slot free for the next real error. `_release_handle()` (a guarded free) is reserved for the two branches where ownership is not known for certain: a Python exception raised before native reports anything, and a failure that leaves the error slot empty (which no defined native failure is expected to produce). In both, a guarded free is a good default, since it is a real free when the handle is still ours and a `-1` no-op when the native side already took it. -None of this is protected by a lock on the Python side: `ManagedResource` has no thread-safety mechanism of its own, and the retained-vs-consumed guarantee comes entirely from the native pointer registry and its thread-local error slot. As noted under [Which double-free risks this layer guards](#double-free-risk-mitigations), sharing one instance across threads without external synchronization is the caller's responsibility. This is a different hazard from [Fork safety](#fork-safety), which concerns a forked child process, not a thread within the same process. +The retained-vs-consumed decision itself comes entirely from the native pointer registry and its thread-local error slot, not from a Python-side lock: `_op_lock` guards concurrent teardown against a call still in flight (see [Locking and in-flight tracking](#locking-and-in-flight-tracking)), but it plays no part in reading which rejection prefix the native side set. That is a different hazard from [Fork safety](#fork-safety), which concerns a forked child process, not a thread within the same process. -A consuming C FFI function first removes the pointer from its registry, then reconstructs the owned value from it. `untrack_or_return!` runs ahead of `Box::from_raw` in `c2pa_c_ffi`. If the address is unknown or the wrong type, the untrack step fails before ownership is taken and sets an error whose prefix (`UntrackedPointer:` or `WrongPointerType:`) identifies it as a pre-consume rejection. Once the value has been reconstructed, a later failure simply drops it, the same as any owned value going out of scope. The Python side stays defensive (and as generic as possible) rather than assuming any exact behavior: it retains the handle when it recognizes one of those rejection prefixes, and where the outcome is unclear it falls back to the guarded free. A native side that behaved differently would degrade in one of two bounded ways: If it kept a pointer the Python side treated as consumed, nothing would free that pointer and it would leak. If it had already released a pointer the Python side then tried to free, the registry would not find the address and the free would return `-1` without touching memory. +A consuming C FFI function first removes the pointer from its registry, then reconstructs the owned value from it. `untrack_or_return!` runs ahead of `Box::from_raw` in `c2pa_c_ffi`. If the address is unknown or the wrong type, the untrack step fails before ownership is taken and sets an error whose prefix (one of the `_PRE_CONSUME_ERROR_TAGS`) identifies it as a pre-consume rejection. Once the value has been reconstructed, a later failure simply drops it, the same as any owned value going out of scope. The Python side stays defensive (and as generic as possible) rather than assuming any exact behavior: it retains the handle when it recognizes one of those rejection prefixes, and where the outcome is unclear it falls back to the guarded free. A native side that behaved differently would degrade in one of two bounded ways: a pointer the native side still held but Python treated as consumed would leak, since nothing would free it; a pointer the native side had already released but Python then tried to free would return `-1` from the registry without touching memory. ### Adopting the handle before giving it away @@ -531,6 +551,8 @@ sequenceDiagram Both `_cleanup_resources()` and the consumed teardown take this branch. Neither simply skips the work: they null the handle and mark the object `CLOSED` so the child cannot go on to use it or try to free it later. Mutating the child's copy has no effect on the parent's, which is untouched and still valid. +`_teardown()` checks `is_foreign_process()` before taking `_op_lock`, not after, so the foreign-process branch above never tries to acquire a lock in the child. The lock itself would raise there anyway (see [Locking and in-flight tracking](#locking-and-in-flight-tracking)), but `_teardown()` needs to finish its cleanup rather than raise. Therefore, it settles the fork case first and only reaches for the lock once it knows this process owns the pointer. + The memory the child skips is not lost for good. A child that calls `exec()` replaces its address space; a child that exits has its memory reclaimed by the OS. Even a long-lived child (a `multiprocessing` worker using the fork start method) retains at most the objects it inherited at fork time, which is a bounded, one-off amount rather than a growing leak. Anything the child allocates itself carries the child's own PID and is freed normally. > [!NOTE] From e657a236f985eb865dc18ebc2dea8283fab3fac2 Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 21:33:27 -0700 Subject: [PATCH 14/15] fix: Rebaseline --- tests/perf/baseline.json | 360 ++++++++++++++++++------------------ tests/perf/reports/.gitkeep | 0 2 files changed, 180 insertions(+), 180 deletions(-) delete mode 100644 tests/perf/reports/.gitkeep diff --git a/tests/perf/baseline.json b/tests/perf/baseline.json index 7ce38df4..74a431a9 100644 --- a/tests/perf/baseline.json +++ b/tests/perf/baseline.json @@ -8,298 +8,298 @@ "arch": "aarch64" }, "reader_jpeg_legacy": { - "peak_bytes": 3851610, - "leaked_bytes": 3351823, - "total_allocations": 1362322 + "peak_bytes": 3878176, + "leaked_bytes": 3381657, + "total_allocations": 1307172 }, "reader_jpeg_with_context": { - "peak_bytes": 3845367, - "leaked_bytes": 3345097, - "total_allocations": 1349879 + "peak_bytes": 3872284, + "leaked_bytes": 3374437, + "total_allocations": 1299545 }, "reader_manifest_data_context": { - "peak_bytes": 7636730, - "leaked_bytes": 3468040, - "total_allocations": 1147359 + "peak_bytes": 7658137, + "leaked_bytes": 3491877, + "total_allocations": 1098306 }, "reader_mp4": { - "peak_bytes": 4222601, - "leaked_bytes": 3345724, - "total_allocations": 4095915 + "peak_bytes": 4238670, + "leaked_bytes": 3374080, + "total_allocations": 3984581 }, "reader_wav": { - "peak_bytes": 4523095, - "leaked_bytes": 3355666, - "total_allocations": 742391 + "peak_bytes": 4539135, + "leaked_bytes": 3384038, + "total_allocations": 739057 }, "builder_sign_jpeg_legacy": { - "peak_bytes": 7785129, - "leaked_bytes": 3468507, - "total_allocations": 1041412 + "peak_bytes": 7810402, + "leaked_bytes": 3498186, + "total_allocations": 1019644 }, "builder_sign_jpeg_with_context": { - "peak_bytes": 7779538, - "leaked_bytes": 3463042, - "total_allocations": 1027485 + "peak_bytes": 7802790, + "leaked_bytes": 3490850, + "total_allocations": 1005722 }, "builder_sign_png_legacy": { - "peak_bytes": 8023081, - "leaked_bytes": 3468300, - "total_allocations": 3883115 + "peak_bytes": 8048349, + "leaked_bytes": 3498022, + "total_allocations": 3861729 }, "builder_sign_png_with_context": { - "peak_bytes": 8017008, - "leaked_bytes": 3462829, - "total_allocations": 3869515 + "peak_bytes": 8041266, + "leaked_bytes": 3491795, + "total_allocations": 3847713 }, "builder_sign_jpeg_parallel_split_pool": { - "peak_bytes": 45854797, - "leaked_bytes": 3840928, - "total_allocations": 1035646 + "peak_bytes": 45869711, + "leaked_bytes": 3860295, + "total_allocations": 1009812 }, "builder_sign_jpeg_parallel_split_barrier": { - "peak_bytes": 45844809, - "leaked_bytes": 3861014, - "total_allocations": 1037741 + "peak_bytes": 45838322, + "leaked_bytes": 3859113, + "total_allocations": 1008528 }, "builder_sign_png_parallel_split_pool": { - "peak_bytes": 46586728, - "leaked_bytes": 3868054, - "total_allocations": 3877696 + "peak_bytes": 46107474, + "leaked_bytes": 3877946, + "total_allocations": 3851810 }, "builder_sign_png_parallel_split_barrier": { - "peak_bytes": 46082548, - "leaked_bytes": 3879161, - "total_allocations": 3879780 + "peak_bytes": 46075853, + "leaked_bytes": 3877260, + "total_allocations": 3850542 }, "builder_sign_gif": { - "peak_bytes": 14635465, - "leaked_bytes": 3461270, - "total_allocations": 17017654 + "peak_bytes": 14660656, + "leaked_bytes": 3491475, + "total_allocations": 16995947 }, "builder_sign_heic": { - "peak_bytes": 4698434, - "leaked_bytes": 3469086, - "total_allocations": 1563419 + "peak_bytes": 4723711, + "leaked_bytes": 3499336, + "total_allocations": 1529895 }, "builder_sign_m4a": { - "peak_bytes": 18833496, - "leaked_bytes": 3469085, - "total_allocations": 5194205 + "peak_bytes": 18859208, + "leaked_bytes": 3499290, + "total_allocations": 5160957 }, "builder_sign_webp": { - "peak_bytes": 8991237, - "leaked_bytes": 3461271, - "total_allocations": 916145 + "peak_bytes": 9016473, + "leaked_bytes": 3491521, + "total_allocations": 898326 }, "builder_sign_avi": { - "peak_bytes": 7130933, - "leaked_bytes": 3461270, - "total_allocations": 89982012 + "peak_bytes": 7156127, + "leaked_bytes": 3491475, + "total_allocations": 89959516 }, "builder_sign_mp4": { - "peak_bytes": 6245379, - "leaked_bytes": 3469085, - "total_allocations": 3788717 + "peak_bytes": 6270688, + "leaked_bytes": 3499335, + "total_allocations": 3753347 }, "builder_sign_tiff": { - "peak_bytes": 13213169, - "leaked_bytes": 3461271, - "total_allocations": 10862700 + "peak_bytes": 13238405, + "leaked_bytes": 3491521, + "total_allocations": 10845456 }, "builder_sign_jpeg_parent_of": { - "peak_bytes": 14265295, - "leaked_bytes": 3461665, - "total_allocations": 2506107 + "peak_bytes": 14290485, + "leaked_bytes": 3492132, + "total_allocations": 2434129 }, "builder_sign_jpeg_component_of": { - "peak_bytes": 14266996, - "leaked_bytes": 3462012, - "total_allocations": 2551180 + "peak_bytes": 14291315, + "leaked_bytes": 3491288, + "total_allocations": 2477796 }, "builder_sign_jpeg_parent_and_component": { - "peak_bytes": 14665241, - "leaked_bytes": 3614613, - "total_allocations": 4523960 + "peak_bytes": 14638503, + "leaked_bytes": 3636596, + "total_allocations": 4394837 }, "builder_sign_jpeg_parent_and_component_mixed_mime": { - "peak_bytes": 14568780, - "leaked_bytes": 3462718, - "total_allocations": 5517180 + "peak_bytes": 14593417, + "leaked_bytes": 3492387, + "total_allocations": 5447516 }, "builder_sign_jpeg_two_components_same_mime": { - "peak_bytes": 14559274, - "leaked_bytes": 3564233, - "total_allocations": 4497379 + "peak_bytes": 14631537, + "leaked_bytes": 3636600, + "total_allocations": 4367414 }, "builder_sign_jpeg_two_components_mixed_mime": { - "peak_bytes": 14564839, - "leaked_bytes": 3461873, - "total_allocations": 5490592 + "peak_bytes": 14589983, + "leaked_bytes": 3492082, + "total_allocations": 5419842 }, "builder_sign_jpeg_archive_roundtrip": { - "peak_bytes": 14297571, - "leaked_bytes": 3481212, - "total_allocations": 3467149 + "peak_bytes": 14321971, + "leaked_bytes": 3512017, + "total_allocations": 3343806 }, "builder_from_archive_roundtrip": { - "peak_bytes": 14297349, - "leaked_bytes": 3480475, - "total_allocations": 3101030 + "peak_bytes": 14320730, + "leaked_bytes": 3510869, + "total_allocations": 2987299 }, "builder_with_archive_swap": { - "peak_bytes": 3681081, - "leaked_bytes": 3350198, - "total_allocations": 704373 + "peak_bytes": 3720558, + "leaked_bytes": 3389404, + "total_allocations": 708591 }, "reader_with_fragment_swap": { - "peak_bytes": 3778159, - "leaked_bytes": 3353205, - "total_allocations": 3787587 + "peak_bytes": 3805398, + "leaked_bytes": 3382233, + "total_allocations": 3769246 + }, + "reader_with_fragment_repeated": { + "peak_bytes": 3803023, + "leaked_bytes": 3380589, + "total_allocations": 1842543 }, "with_fragment_pre_consume_rejection": { - "peak_bytes": 3778057, - "leaked_bytes": 3354795, - "total_allocations": 2094004 + "peak_bytes": 3805255, + "leaked_bytes": 3384325, + "total_allocations": 2093470 }, "with_archive_post_consume_failure": { - "peak_bytes": 3350600, - "leaked_bytes": 3308056, - "total_allocations": 175290 + "peak_bytes": 3388641, + "leaked_bytes": 3346670, + "total_allocations": 185458 }, "with_fragment_marshalling_error": { - "peak_bytes": 3708068, - "leaked_bytes": 3352335, - "total_allocations": 2077090 + "peak_bytes": 3734139, + "leaked_bytes": 3381598, + "total_allocations": 2072540 }, "with_fragment_mixed_outcomes": { - "peak_bytes": 3779175, - "leaked_bytes": 3356294, - "total_allocations": 2656787 + "peak_bytes": 3803836, + "leaked_bytes": 3382987, + "total_allocations": 2650818 }, "builder_to_archive_with_ingredient": { - "peak_bytes": 14069232, - "leaked_bytes": 3337316, - "total_allocations": 1830896 + "peak_bytes": 14107950, + "leaked_bytes": 3375874, + "total_allocations": 1766289 }, "builder_sign_jpeg_archive_roundtrip_ingredient_in_archive": { - "peak_bytes": 14287046, - "leaked_bytes": 3481977, - "total_allocations": 5879957 + "peak_bytes": 14311785, + "leaked_bytes": 3511500, + "total_allocations": 5681125 }, "builder_write_ingredient_archive": { - "peak_bytes": 14069289, - "leaked_bytes": 3337377, - "total_allocations": 1805304 + "peak_bytes": 14107944, + "leaked_bytes": 3375872, + "total_allocations": 1742859 }, "builder_sign_jpeg_add_ingredient_from_archive": { - "peak_bytes": 14133742, - "leaked_bytes": 3480831, - "total_allocations": 3415920 + "peak_bytes": 14174086, + "leaked_bytes": 3512155, + "total_allocations": 3320279 }, "builder_ingredient_archive_roundtrip": { - "peak_bytes": 14284443, - "leaked_bytes": 3480809, - "total_allocations": 5132060 + "peak_bytes": 14310577, + "leaked_bytes": 3512018, + "total_allocations": 4973895 }, "builder_sign_jpeg_two_ingredient_archives": { - "peak_bytes": 14134560, - "leaked_bytes": 3481604, - "total_allocations": 4215728 + "peak_bytes": 14173988, + "leaked_bytes": 3512433, + "total_allocations": 4113195 }, "reader_error_no_manifest": { - "peak_bytes": 3564471, - "leaked_bytes": 3323629, - "total_allocations": 276175 + "peak_bytes": 3588164, + "leaked_bytes": 3352030, + "total_allocations": 276214 }, "builder_error_invalid_manifest": { - "peak_bytes": 3352053, - "leaked_bytes": 3297079, - "total_allocations": 113926 + "peak_bytes": 3388827, + "leaked_bytes": 3333835, + "total_allocations": 115678 }, "reader_string_apis": { - "peak_bytes": 3978113, - "leaked_bytes": 3346111, - "total_allocations": 2287335 + "peak_bytes": 4005286, + "leaked_bytes": 3375590, + "total_allocations": 2183974 }, "signer_construction": { - "peak_bytes": 3350893, - "leaked_bytes": 3288137, - "total_allocations": 153245 + "peak_bytes": 3388872, + "leaked_bytes": 3325939, + "total_allocations": 155796 }, "builder_from_context_construction": { - "peak_bytes": 3350600, - "leaked_bytes": 3288582, - "total_allocations": 112688 + "peak_bytes": 3388641, + "leaked_bytes": 3327073, + "total_allocations": 120460 }, "fork_reader_collect": { - "peak_bytes": 3850530, - "leaked_bytes": 3353063, - "total_allocations": 1328122 + "peak_bytes": 3877630, + "leaked_bytes": 3381450, + "total_allocations": 1271971 }, "fork_contended_mutex": { - "peak_bytes": 7679019, - "leaked_bytes": 3482128, - "total_allocations": 67472694 + "peak_bytes": 7646946, + "leaked_bytes": 3477655, + "total_allocations": 65668554 }, "fork_thread_local_orphan": { - "peak_bytes": 3936170, - "leaked_bytes": 3439733, - "total_allocations": 1381055 + "peak_bytes": 3960026, + "leaked_bytes": 3468161, + "total_allocations": 1324309 }, "fork_gc_cycle": { - "peak_bytes": 3850434, - "leaked_bytes": 3353160, - "total_allocations": 1332098 + "peak_bytes": 3875869, + "leaked_bytes": 3379622, + "total_allocations": 1276946 }, "fork_parent_frees_after_fork": { - "peak_bytes": 5447584, - "leaked_bytes": 3350400, - "total_allocations": 24829257 + "peak_bytes": 5561869, + "leaked_bytes": 3389085, + "total_allocations": 23724547 }, "fork_child_closes_then_parent_frees": { - "peak_bytes": 5446620, - "leaked_bytes": 3350407, - "total_allocations": 24829254 + "peak_bytes": 5563403, + "leaked_bytes": 3390349, + "total_allocations": 23724544 }, "fork_child_sys_exit": { - "peak_bytes": 3850546, - "leaked_bytes": 3353234, - "total_allocations": 1335925 + "peak_bytes": 3877646, + "leaked_bytes": 3381666, + "total_allocations": 1282172 }, "fork_stream_cleanup": { - "peak_bytes": 3464063, - "leaked_bytes": 3291969, - "total_allocations": 105340 + "peak_bytes": 3500857, + "leaked_bytes": 3329555, + "total_allocations": 105687 }, "fork_swap_cleanup": { - "peak_bytes": 3681171, - "leaked_bytes": 3350696, - "total_allocations": 714376 + "peak_bytes": 3720613, + "leaked_bytes": 3389867, + "total_allocations": 718596 }, "fork_contended_mutex_swap": { - "peak_bytes": 7302379, - "leaked_bytes": 3475147, - "total_allocations": 35948516 + "peak_bytes": 7306199, + "leaked_bytes": 3492867, + "total_allocations": 35863039 }, "fork_contended_mutex_wrap": { - "peak_bytes": 7288748, - "leaked_bytes": 3463411, - "total_allocations": 34847186 + "peak_bytes": 7295518, + "leaked_bytes": 3491027, + "total_allocations": 35014720 }, "fork_consumed_signer": { - "peak_bytes": 3350894, - "leaked_bytes": 3288906, - "total_allocations": 175055 + "peak_bytes": 3388873, + "leaked_bytes": 3327740, + "total_allocations": 196221 }, "swap_chain_churn": { - "peak_bytes": 3681161, - "leaked_bytes": 3350287, - "total_allocations": 672537 - }, - "reader_with_fragment_repeated": { - "peak_bytes": 3803564, - "leaked_bytes": 3381191, - "total_allocations": 966288 + "peak_bytes": 3720603, + "leaked_bytes": 3389458, + "total_allocations": 669590 } -} +} \ No newline at end of file diff --git a/tests/perf/reports/.gitkeep b/tests/perf/reports/.gitkeep deleted file mode 100644 index e69de29b..00000000 From 63fc5a7ad0ccc1a8e658aec7d922e80111ebc519 Mon Sep 17 00:00:00 2001 From: tmathern <60901087+tmathern@users.noreply.github.com> Date: Tue, 25 Aug 2026 21:57:23 -0700 Subject: [PATCH 15/15] fix: Reorder locking --- src/c2pa/c2pa.py | 150 ++++++++++++++++-------------- tests/test_unit_tests_threaded.py | 60 ++++++++++++ 2 files changed, 141 insertions(+), 69 deletions(-) diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 83d9593b..1ad8dce1 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -1923,6 +1923,8 @@ def __init__(self, file_like_stream): self._closed = False self._initialized = False self._stream = None + # Serializes close() and __del__ against a concurrent double-free. + self._close_lock = threading.Lock() # Generate unique stream ID using object ID and counter stream_counter = next(Stream._stream_id_counter) @@ -2144,22 +2146,22 @@ def __del__(self): try: if is_foreign_process(self): return - # Only cleanup if not already closed and we have a valid stream - if hasattr(self, '_closed') and not self._closed: - stream = self._stream - if hasattr(self, '_stream') and stream: - # Use internal cleanup to avoid calling close() which could - # cause issues - try: - _lib.c2pa_release_stream(stream) - except Exception: - # Destructors shouldn't raise exceptions - logger.error("Failed to release Stream") - pass - finally: - self._stream = None - self._closed = True - self._initialized = False + lock = getattr(self, '_close_lock', None) + with lock if lock is not None else contextlib.nullcontext(): + # Only cleanup if not already closed and we have a valid stream + if hasattr(self, '_closed') and not self._closed: + stream = self._stream + if hasattr(self, '_stream') and stream: + try: + _lib.c2pa_release_stream(stream) + except Exception: + # Destructors shouldn't raise exceptions + logger.error("Failed to release Stream") + pass + finally: + self._stream = None + self._closed = True + self._initialized = False except Exception: # Destructors must not raise exceptions pass @@ -2172,45 +2174,48 @@ def close(self): Errors during cleanup are logged but not raised to ensure cleanup. Multiple calls to close() are handled gracefully. """ - if self._closed: - return - if is_foreign_process(self): - self._closed = True - self._initialized = False - return + # Serializes against __del__ / a concurrent close(). + with self._close_lock: + if self._closed: + return + if is_foreign_process(self): + self._closed = True + self._initialized = False + return - try: - # Clean up stream first as it depends on callbacks - # Note: We don't close self._file_like_stream as we don't own it, - # the opener owns it. - stream = self._stream - if stream: - try: - _lib.c2pa_release_stream(stream) - except Exception as e: - logger.error( - Stream._ERROR_MESSAGES['stream_error'].format( - str(e))) - finally: - self._stream = None - - # Clean up callbacks - for attr in ['_read_cb', '_seek_cb', '_write_cb', '_flush_cb']: - if hasattr(self, attr): + try: + # Clean up stream first as it depends on callbacks + # Note: We don't close self._file_like_stream as we don't + # own it, the opener owns it. + stream = self._stream + if stream: try: - setattr(self, attr, None) + _lib.c2pa_release_stream(stream) except Exception as e: logger.error( - Stream._ERROR_MESSAGES['callback_error'].format( - attr, str(e))) + Stream._ERROR_MESSAGES['stream_error'].format( + str(e))) + finally: + self._stream = None - except Exception as e: - logger.error( - Stream._ERROR_MESSAGES['cleanup_error'].format( - str(e))) - finally: - self._closed = True - self._initialized = False + # Clean up callbacks + for attr in [ + '_read_cb', '_seek_cb', '_write_cb', '_flush_cb']: + if hasattr(self, attr): + try: + setattr(self, attr, None) + except Exception as e: + logger.error( + Stream._ERROR_MESSAGES['callback_error'] + .format(attr, str(e))) + + except Exception as e: + logger.error( + Stream._ERROR_MESSAGES['cleanup_error'].format( + str(e))) + finally: + self._closed = True + self._initialized = False def write_to_target(self, dest_stream): self._file_like_stream.seek(0) @@ -2896,27 +2901,34 @@ def with_fragment(self, format: Optional[str], stream, frag_obj.close() raise - # Replace the streams this reader owned, - # closing the previous ones so repeated calls do not accumulate them. - # Only the current fragment is retained: the native reader does not - # read a superseded one back, and each wrapper held open pins a native - # stream, its callbacks and the caller's buffer. - previous = self._own_stream - previous_fragments = self._fragment_streams - self._own_stream = main_obj - self._fragment_streams = [frag_obj] - if previous is not None and previous is not main_obj: - try: - previous.close() - except Exception: - logger.warning("Failed to close previous Reader stream") - for fragment in previous_fragments: - if fragment is frag_obj: - continue + # Locked so a concurrent close() cannot run _release() + # between the check and the field swap. + with self._lock(): try: - fragment.close() + self._ensure_valid_state() except Exception: - logger.warning("Failed to close Reader fragment stream") + main_obj.close() + frag_obj.close() + raise + + # Replace the streams this reader owned, closing the previous + # ones (only the current fragment is retained). + previous = self._own_stream + previous_fragments = self._fragment_streams + self._own_stream = main_obj + self._fragment_streams = [frag_obj] + if previous is not None and previous is not main_obj: + try: + previous.close() + except Exception: + logger.warning("Failed to close previous Reader stream") + for fragment in previous_fragments: + if fragment is frag_obj: + continue + try: + fragment.close() + except Exception: + logger.warning("Failed to close Reader fragment stream") # Invalidate caches: processing a new BMFF fragment updates the native # reader's state, which can change the manifest data it returns. diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index 85097321..001ab50e 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -12,6 +12,7 @@ # each license. import ast +import contextlib import ctypes import gc import os @@ -64,6 +65,7 @@ def _make_stream(pid_offset): obj._closed = False obj._initialized = True obj._stream = MagicMock() # non-None stream handle + obj._close_lock = threading.Lock() if pid_offset is not None: obj._owner_pid = os.getpid() + pid_offset return obj @@ -317,6 +319,64 @@ def test_parent_copy_unaffected(self): self.assertNotIn("DeprecationWarning", result.stderr) +class TestReaderWithFragmentConcurrentClose(unittest.TestCase): + """with_fragment's post-native-call bookkeeping must not race close().""" + + def test_close_during_with_fragment_does_not_double_close_stream(self): + init_path = os.path.join(FIXTURES_FOLDER, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_FOLDER, "dash1.m4s") + + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + + entered_gap = threading.Event() + release_gap = threading.Event() + + real_native_call = reader._native_call + + @contextlib.contextmanager + def gated_native_call(): + with real_native_call(): + yield + # Pauses right in with_fragment's unlocked window before it reassigns _own_stream/_fragment_streams. + entered_gap.set() + release_gap.wait(5) + + reader._native_call = gated_native_call + + result = {} + + def run_with_fragment(): + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + reader.with_fragment("video/mp4", init, frag) + result["outcome"] = "ok" + except BaseException as e: # noqa: BLE001 - asserted below + result["outcome"] = e + + worker = threading.Thread(target=run_with_fragment, daemon=True) + worker.start() + self.assertTrue( + entered_gap.wait(5), + "with_fragment never reached the post-native-call gap") + + # close() must win the race cleanly, not leave with_fragment hung, crashed, or silently successful. + reader.close() + release_gap.set() + worker.join(5) + self.assertFalse(worker.is_alive(), "with_fragment hung") + self.assertIsInstance( + result.get("outcome"), Error, + "with_fragment must raise C2paError when it loses the race, " + "not hang, crash, or silently succeed") + + self.assertEqual(reader._lifecycle_state, LifecycleState.CLOSED) + # with_fragment must not resurrect these fields on a reader close() already tore down. + self.assertIsNone(reader._own_stream) + self.assertEqual(reader._fragment_streams, []) + + class TestHelpers(unittest.TestCase): def test_record_and_detect_own_pid(self):