diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 6c697862..e83018ad 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -272,9 +272,13 @@ def __init__(self): self._released = False record_owner_pid(self) - def _lock(self): + def _state_lock(self): """Return this resource's operation lock. + Acquiring it only provides mutual exclusion; unlike _lock(), + it does not mark the thread as being inside a native-error + section. + 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 @@ -320,6 +324,19 @@ def _ensure_not_borrowed(self): f"{name} is in use by another operation and " f"cannot be consumed") + @contextlib.contextmanager + def _lock(self): + """Hold this resource's operation lock its duration, + and mark this thread as inside a native-error section. + + Never hold this across a native call that drives stream callbacks. + Those calls release the GIL and re-enter caller-supplied + code, which may call back into this API on another thread. + Only calls that don't touch callbacks are serialized here. + """ + with self._state_lock(), _native_section(): + yield + @contextlib.contextmanager def _native_call(self): """Hold the handle valid across a native call that goes back @@ -333,25 +350,22 @@ def _native_call(self): 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. + + Also opens a native-error section around the yielded body (see + _lock()): the in-flight guard alone only protects this resource's + own handle, not the shared thread-local error slot a caller inside + the block is about to read. """ - with self._lock(): + with self._state_lock(): self._ensure_valid_state() self._inflight = getattr(self, '_inflight', 0) + 1 try: - yield + with _native_section(): + yield finally: - with self._lock(): + with self._state_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) + self._maybe_flush_pending() @staticmethod def _free_native_ptr(ptr): @@ -412,6 +426,11 @@ 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. + Deferred (instead of run now) when either gate is blocking: + - this resource's own handle is in flight in a native call + - this thread is inside a native-error section for some call, + that may access the native error slot. + The forked-child case is handled before the lock is taken, because _lock() raises in a child: this path has to finish rather than report an error, so it cannot rely on acquiring. @@ -423,22 +442,20 @@ def _teardown(self, free_handle: bool): self._lifecycle_state = LifecycleState.CLOSED return - with self._lock(): + with self._state_lock(): if getattr(self, '_released', False): # A racing close()/__del__ already ran the release branch # under this lock. # Idempotent: nothing left to release or free. # Keyed on the release having happened, not on CLOSED: the # deferred path below sets CLOSED without releasing, and still - # owes a release performed by _native_call()'s finally. + # owes a release performed by _finish_teardown(). return - if getattr(self, '_inflight', 0) > 0: - # 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. + if getattr(self, '_inflight', 0) > 0 or _in_native_section(): + # Mark the resource closed now so it cannot be used while + # the free is pending, but record the intent: + # whichever check is blocking will call + # _maybe_flush_pending() once it clears. # # free_handle=False records that a consuming call handed # ownership to the native library. Ownership does not come @@ -453,20 +470,61 @@ def _teardown(self, free_handle: bool): self._pending_teardown = ( self._pending_teardown and free_handle) self._lifecycle_state = LifecycleState.CLOSED + if _in_native_section(): + _register_for_section_flush(self) return - self._released = True + self._finish_teardown(free_handle) + + def _finish_teardown(self, free_handle: bool): + """The part of _teardown that only runs once nothing is blocking + teardown. Steps: release, null the handle, free if requested. + + Not called directly outside _teardown/_maybe_flush_pending: + callers that want to close a resource still go through _teardown, + which decides whether this can run now or must be deferred. + """ + if is_foreign_process(self): + self._handle = None self._lifecycle_state = LifecycleState.CLOSED - self._safe_release() + return - handle, self._handle = self._handle, None - if free_handle and handle: - try: - # Subclasses may override the deallocator. - type(self)._free_native_ptr(handle) - except Exception: - logger.error("Failed to free native %s resources", - type(self).__name__, exc_info=True) + if getattr(self, '_released', False): + # A concurrent caller already ran this. + return + + self._released = True + 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) + + def _maybe_flush_pending(self): + """Called when a gate that may have been blocking a deferred + teardown clears (this resource's own _inflight dropping to 0, or + this thread's native-error section closing). + """ + if is_foreign_process(self): + return + + with self._state_lock(): + if self._pending_teardown is None: + return + if getattr(self, '_inflight', 0) > 0: + return + if _in_native_section(): + # An enclosing section is still open. + # Re-register, since the deferral is the only path to this free. + _register_for_section_flush(self) + return + free_handle, self._pending_teardown = self._pending_teardown, None + self._finish_teardown(free_handle) def _release_handle(self): """Free this handle, then close the object. Used only where ownership is @@ -516,9 +574,11 @@ def _create_and_activate(self, ffi_call, error_message, *, Raises: C2paError: If the pointer fails validation; it is freed first. """ - ptr = ffi_call() + ptr = None try: - _check_ffi_operation_result(ptr, error_message, check=check) + with self._lock(): + ptr = ffi_call() + _check_ffi_operation_result(ptr, error_message, check=check) self._activate(ptr) except Exception: if ptr: @@ -556,10 +616,24 @@ def _swap_handle(self, new_handle): _PRE_CONSUME_ERROR_TAGS = ( "UntrackedPointer:", "WrongPointerType:", - "NullParameter:", - "InvalidBufferSize:", ) + # An error tag starts the message or follows this one wrapper. + _NATIVE_ERROR_WRAPPER = "Other: " + + @staticmethod + def _is_pre_consume_rejection(error: str) -> bool: + """True when native rejected the handle before taking ownership. + + Anchored, not a substring search: native quotes caller text verbatim, + so a tag mid-message describes the caller's input, not ownership. + """ + body = error + if body.startswith(ManagedResource._NATIVE_ERROR_WRAPPER): + body = body[len(ManagedResource._NATIVE_ERROR_WRAPPER):] + return any(body.startswith(tag) + for tag in ManagedResource._PRE_CONSUME_ERROR_TAGS) + def _invoke_consume(self, ffi_call, error_message, *, reserved=False): """Run an FFI call that consumes this handle, returning its raw result. @@ -582,6 +656,8 @@ def _invoke_consume(self, ffi_call, error_message, *, reserved=False): ctypes.ArgumentError: If marshalling failed; handle untouched. C2paError: If the call raised any other exception. """ + # Same thread that makes the call, same thread-local slot. + _mark_sentinel_no_native_error() try: return ffi_call(self._handle) except ctypes.ArgumentError: @@ -601,16 +677,12 @@ def _raise_consume_failure(self, error_message, previous_state=None): """Raise the error from an FFI handler consuming call. The native error is read before any free so a free's own - pointer-tracking error cannot overwrite it: the native error slot is - sticky and thread-local and the SDK does not clear it before the call, - so this trusts that the failing native path set its own error. - - That ordering is required: - c2pa_free on a handle the registry no longer tracks returns -1 and - overwrites the slot with its own "Other: UntrackedPointer: 0x..." - message. Freeing first would therefore replace the real failure - with another one and, because that substitute carries a pre-consume - tag, invert the retain/consume decision made below. + pointer-tracking error cannot overwrite it. + The native error slot is sticky and thread-local, + and the native SDK does not clear it before the call. + _invoke_consume marks the slot as carrying no error right before a + consuming call, so a failure that sets no error of its own reads back + as no error rather than as a stale one left by an earlier call. A caller that reserved the handle with _begin_consume() passes previous_state and stays reserved until this classification finishes. @@ -631,8 +703,7 @@ def _raise_consume_failure(self, error_message, previous_state=None): """ error = _read_native_error() if error: - if any(tag in error - for tag in ManagedResource._PRE_CONSUME_ERROR_TAGS): + if ManagedResource._is_pre_consume_rejection(error): logger.warning( "%s: native call rejected the handle before taking " "ownership (%s); handle retained", @@ -649,9 +720,13 @@ def _raise_consume_failure(self, error_message, previous_state=None): self._teardown(free_handle=False) _raise_typed_c2pa_error(error) - # No error in the slot: ownership is unknown, so free defensively. + # No error of its own: ownership is unknown, so free defensively. + # c2pa_free returns -1 for an address native already reclaimed. # A reservation leaves the resource CLOSED with the handle set, # which _release_handle() nulls without freeing. + logger.debug( + "%s: consuming call failed without setting error", + type(self).__name__) if previous_state is not None: self._teardown(free_handle=True) else: @@ -668,7 +743,7 @@ def _begin_consume(self): Raises: C2paError: If a native call is in flight on this resource. """ - with self._lock(): + with self._state_lock(): # A consumed or closed resource has no handle left to hand over; # without this the call would pass a null pointer to native. self._ensure_valid_state() @@ -682,8 +757,13 @@ def _abort_consume(self, previous_state): A pre-consume rejection leaves the handle ours, so the resource has to become usable again. + + A deferred free still happens when the section drains, so a resource + with a queued teardown stays closed. """ - with self._lock(): + with self._state_lock(): + if self._pending_teardown is not None: + return if self._lifecycle_state == LifecycleState.CLOSED and self._handle: self._lifecycle_state = previous_state @@ -911,25 +991,136 @@ class C2paStream(ctypes.Structure): ] +# Unaligned address passed to c2pa_free to plant a marker +# in the native error slot. +# Never a real handle: allocations are aligned, and the Python +# layer only passes real handles or this constant to c2pa_free. +_MARKER_ADDR = 1 + +# Exact text the native lib writes for a failed free of _MARKER_ADDR. +# Learned at import by _learn_sentinel_no_native_error_text(). +# The format is a native implementation detail, +# so it is read back rather than hardcoded. +_NATIVE_NO_ERROR_TEXT = None + + +def _mark_sentinel_no_native_error(): + """Write the no-error marker into this thread's native error slot. + + A c2pa_free of an address the registry does not track writes + an expected error message learned at import into the + thread-local error slot and returns -1. + + This marker mechanism exists to distinguish a consuming call that + failed without setting its own error from a stale message left + by an earlier call on the same thread. + """ + _lib.c2pa_free(_MARKER_ADDR) + + +def _is_no_native_error(message: str) -> bool: + """True for the sentinel marker meaning "no current error of our own".""" + return message == _NATIVE_NO_ERROR_TEXT + + def _read_native_error() -> Optional[str]: """Read the last error from the native library, or None if unset. - Peeks: the error stays in the native slot, - until the next error overwrites it. - - With no error set the native side still returns an owned pointer to an - empty string, so the pointer alone does not tell us whether there is an - error. Only a non-empty message counts as one; the empty string still - has to be freed. + The slot is marked as carrying no error before returning, so a + given error is reported once, by the caller that observes it. The native + slot is thread-local and sticky, so a message left in place stays readable + indefinitely and is available to be reported again by a later, + unrelated call that failed without setting an error of its own + (or a missing clear of an error slot). """ error = _lib.c2pa_error() if not error: + # NULL means the message could not be rendered, not that the slot + # is empty, so it still has to be marked. + _mark_sentinel_no_native_error() return None try: message = ctypes.string_at(error).decode('utf-8') finally: _lib.c2pa_string_free(error) - return message or None + if not message: + _mark_sentinel_no_native_error() + return None + + _mark_sentinel_no_native_error() + if _is_no_native_error(message): + return None + return message + + +_native_section_state = threading.local() + + +def _in_native_section() -> bool: + """True while this thread is between an FFI call and reading back the + native error it may have set (see _native_section()).""" + return getattr(_native_section_state, 'depth', 0) > 0 + + +def _register_for_section_flush(resource): + """Record that `resource`'s teardown was deferred only because this + thread's native-error section was open.""" + pending = getattr(_native_section_state, 'pending_resources', None) + if pending is not None: + pending.append(resource) + + +@contextlib.contextmanager +def _native_section(): + """Mark this thread as inside a section where a native call's result is + about to be read back: an error-slot check, or a consuming call's + success/failure classification. + + Reentrant: a call whose own native call triggers another one + recursively (same thread) nests correctly here. Only the outermost + span flushes, so nothing is freed before an inner, still-open span is + done reading its own error. + + Each flush is guarded: the deferral is the only remaining path to that + resource's free, so one raising would strand the rest. The first + exception is re-raised once the queue is drained. A body that raised + keeps its own exception, and the flush failure is logged. + """ + state = _native_section_state + depth = getattr(state, 'depth', 0) + state.depth = depth + 1 + if depth == 0: + state.pending_resources = [] + + def _drain(): + """Flush every deferred resource. Returns the first error raised.""" + pending, state.pending_resources = state.pending_resources, [] + first_error = None + for resource in pending: + try: + resource._maybe_flush_pending() + except BaseException as e: # noqa: BLE001 + if first_error is None: + first_error = e + return first_error + + try: + yield + except BaseException: + state.depth -= 1 + if state.depth == 0: + drain_error = _drain() + if drain_error is not None: + logger.error( + "Deferred teardown failed while unwinding: %s", + drain_error) + raise + else: + state.depth -= 1 + if state.depth == 0: + drain_error = _drain() + if drain_error is not None: + raise drain_error class C2paSignerInfo(ctypes.Structure): @@ -1254,6 +1445,41 @@ def _setup_function(func, argtypes, restype=None): ) _setup_function(_lib.c2pa_free, [ctypes.c_void_p], ctypes.c_int) + +def _learn_sentinel_no_native_error_text(): + """Plant the marker once and read back the exact text the native lib + produces for it, so equality checks match this build of the lib. + + Runs on the importing thread; the text is a format constant, so the + learned value holds for every thread. Raises at import when the read + back text is empty, or when it does not carry the planted address, + because the marker mechanism cannot work in either case. + """ + _mark_sentinel_no_native_error() + raw = _lib.c2pa_error() + if not raw: + raise ImportError( + "c2pa native library did not report an error for a free of " + "an untracked pointer; the error-slot marker cannot work") + try: + text = ctypes.string_at(raw).decode('utf-8') + finally: + _lib.c2pa_string_free(raw) + if not text: + raise ImportError( + "c2pa native library reported an empty error for a free of " + "an untracked pointer; the error-slot marker cannot work") + marker_hex = hex(_MARKER_ADDR) + if marker_hex not in text: + raise ImportError( + "c2pa native library's untracked-pointer error text no longer " + f"includes the planted address {marker_hex}; the error-slot " + "marker assumption no longer holds") + return text + + +_NATIVE_NO_ERROR_TEXT = _learn_sentinel_no_native_error_text() + _setup_function( _lib.c2pa_context_builder_set_signer, [ctypes.POINTER(C2paContextBuilder), ctypes.POINTER(C2paSigner)], @@ -1675,11 +1901,12 @@ def load_settings(settings: Union[str, dict], format: str = "json") -> None: except (AttributeError, UnicodeEncodeError) as e: raise C2paError(f"Failed to encode settings to UTF-8: {e}") - result = _lib.c2pa_load_settings(settings_bytes, format_bytes) - _check_ffi_operation_result( - result, - "Error loading settings", - check=lambda r: r != 0) + with _native_section(): + result = _lib.c2pa_load_settings(settings_bytes, format_bytes) + _check_ffi_operation_result( + result, + "Error loading settings", + check=lambda r: r != 0) @contextlib.contextmanager @@ -1936,11 +2163,12 @@ def __init__( # a successful build consumes it, so close() is then a no-op. with self._NativeBuilder() as nb: if settings is not None: - _check_ffi_operation_result( - _lib.c2pa_context_builder_set_settings( - nb._handle, settings._c_settings), - "Failed to set settings on Context", - check=lambda r: r != 0) + with nb._lock(): + _check_ffi_operation_result( + _lib.c2pa_context_builder_set_settings( + nb._handle, settings._c_settings), + "Failed to set settings on Context", + check=lambda r: r != 0) if signer is not None: # No in-flight guard around the hand-off: the consume @@ -1960,6 +2188,10 @@ def __init__( "Failed to set signer on Context: {}") self._has_signer = True + # No borrow around the build: _ensure_not_borrowed refuses a + # consume nested in a _native_call() on the same resource, + # because the enclosing frame would still expect the handle + # back after it had been handed to native. context_ptr = nb._consume_into( lambda h: _lib.c2pa_context_builder_build(h), "Failed to build Context: {}") @@ -2900,25 +3132,27 @@ def _init_from_context(self, context, format_or_path, # Consume current reader, # with manifest data and stream (C FFI pattern), # to create a new one (switch out) - self._consume_and_swap( - lambda handle: ( - _lib.c2pa_reader_with_manifest_data_and_stream( - handle, - format_arg, - self._own_stream._stream, - manifest_array, - len(manifest_data), - ) - ), - Reader._ERROR_MESSAGES['reader_error']) + with self._native_call(): + self._consume_and_swap( + lambda handle: ( + _lib.c2pa_reader_with_manifest_data_and_stream( + handle, + format_arg, + self._own_stream._stream, + manifest_array, + len(manifest_data), + ) + ), + Reader._ERROR_MESSAGES['reader_error']) else: # Consume reader with stream - self._consume_and_swap( - lambda handle: _lib.c2pa_reader_with_stream( - handle, format_arg, - self._own_stream._stream, - ), - Reader._ERROR_MESSAGES['reader_error']) + with self._native_call(): + self._consume_and_swap( + lambda handle: _lib.c2pa_reader_with_stream( + handle, format_arg, + self._own_stream._stream, + ), + Reader._ERROR_MESSAGES['reader_error']) except Exception: self._close_streams() raise @@ -3379,10 +3613,12 @@ def from_info(cls, signer_info: C2paSignerInfo) -> 'Signer': Raises: C2paError: If there was an error creating the signer """ - signer_ptr = _lib.c2pa_signer_from_info(ctypes.byref(signer_info)) + with _native_section(): + signer_ptr = _lib.c2pa_signer_from_info(ctypes.byref(signer_info)) - _check_ffi_operation_result( - signer_ptr, "Failed to create signer from configured signer_info") + _check_ffi_operation_result( + signer_ptr, + "Failed to create signer from configured signer_info") try: return cls(signer_ptr) @@ -3505,16 +3741,17 @@ def wrapped_callback( callback_cb = SignerCallback(wrapped_callback) # Create the signer with the wrapped callback - signer_ptr = _lib.c2pa_signer_create( - None, - callback_cb, - alg, - certs_bytes, - tsa_url_bytes - ) + with _native_section(): + signer_ptr = _lib.c2pa_signer_create( + None, + callback_cb, + alg, + certs_bytes, + tsa_url_bytes + ) - _check_ffi_operation_result(signer_ptr, - "Failed to create signer") + _check_ffi_operation_result(signer_ptr, + "Failed to create signer") try: # Create and return the signer instance with the callback @@ -3676,11 +3913,11 @@ def from_archive( stream_obj = Stream(stream) try: - handle = _lib.c2pa_builder_from_archive(stream_obj._stream) + with _native_section(): + handle = _lib.c2pa_builder_from_archive(stream_obj._stream) - _check_ffi_operation_result(handle, - "Failed to create builder from archive" - ) + _check_ffi_operation_result( + handle, "Failed to create builder from archive") try: # A builder from an archive here carries no context. @@ -3759,10 +3996,11 @@ def _init_from_context(self, context, json_str): context.execution_context), Builder._ERROR_MESSAGES['builder_error']) - self._consume_and_swap( - lambda handle: _lib.c2pa_builder_with_definition( - handle, json_str), - Builder._ERROR_MESSAGES['builder_error']) + with self._native_call(): + self._consume_and_swap( + lambda handle: _lib.c2pa_builder_with_definition( + handle, json_str), + Builder._ERROR_MESSAGES['builder_error']) def _init_attrs(self): super()._init_attrs() @@ -4094,9 +4332,10 @@ def _sign_internal( 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. + # The result check and the close() are deliberately + # outside of it: the check needs its own, later section, + # and close() runs only once that check has read whatever + # error this call set. with self._native_call(): if signer is not None: # Signer needs its own in-flight guard. @@ -4129,18 +4368,25 @@ def _sign_internal( 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. - self.close() except Exception as e: self.close() raise C2paError(f"Error during signing: {e}") from e - _check_ffi_operation_result( - result, - "Error during signing", - check=lambda r: r < 0) + try: + # Own section (the native_call already closed, so its + # own reads are done): close() can free this Builder, + # and freeing can write to the same thread-local error slot + # this check reads. + with _native_section(): + _check_ffi_operation_result( + result, + "Error during signing", + check=lambda r: r < 0) + finally: + # Sign borrows the Builder without taking ownership. + # Closing here ensures resources clean up, and single + # use/single sign done by a Builder. + self.close() # Capture the manifest bytes if available manifest_bytes = b"" @@ -4372,17 +4618,18 @@ def format_embeddable(format: str, manifest_bytes: bytes) -> tuple[int, bytes]: ) result_bytes_ptr = ctypes.POINTER(ctypes.c_ubyte)() - result = _lib.c2pa_format_embeddable( - format_str, - manifest_array, - len(manifest_bytes), - ctypes.byref(result_bytes_ptr) - ) + with _native_section(): + result = _lib.c2pa_format_embeddable( + format_str, + manifest_array, + len(manifest_bytes), + ctypes.byref(result_bytes_ptr) + ) - _check_ffi_operation_result( - result, - "Failed to format embeddable manifest", - check=lambda r: r < 0) + _check_ffi_operation_result( + result, + "Failed to format embeddable manifest", + check=lambda r: r < 0) size = result try: @@ -4504,14 +4751,15 @@ def ed25519_sign(data: bytes, private_key: str) -> bytes: f"Invalid UTF-8 characters in private key: {str(e)}") # Perform the signing operation - signature_ptr = _lib.c2pa_ed25519_sign( - data_array, - data_size, - key_bytes - ) + with _native_section(): + signature_ptr = _lib.c2pa_ed25519_sign( + data_array, + data_size, + key_bytes + ) - _check_ffi_operation_result(signature_ptr, - "Failed to sign data with Ed25519") + _check_ffi_operation_result(signature_ptr, + "Failed to sign data with Ed25519") try: # Ed25519 signatures are always 64 bytes diff --git a/tests/perf/baseline.json b/tests/perf/baseline.json index c151efe5..5d7017ee 100644 --- a/tests/perf/baseline.json +++ b/tests/perf/baseline.json @@ -2,299 +2,304 @@ "_meta": { "memray_version": "1.19.3", "python_version": "3.12.13", - "c2pa_native_version": "c2pa-v0.90.0", + "c2pa_native_version": "c2pa-v0.90.16", "iterations": 200, "perf_env": "python-3.12-slim", "arch": "aarch64" }, "reader_jpeg_legacy": { - "peak_bytes": 3851610, - "leaked_bytes": 3351823, - "total_allocations": 1362322 + "peak_bytes": 3912724, + "leaked_bytes": 3414162, + "total_allocations": 1324293 }, "reader_jpeg_with_context": { - "peak_bytes": 3845367, - "leaked_bytes": 3345097, - "total_allocations": 1349879 + "peak_bytes": 3907256, + "leaked_bytes": 3407478, + "total_allocations": 1333897 }, "reader_manifest_data_context": { - "peak_bytes": 7636730, - "leaked_bytes": 3468040, - "total_allocations": 1147359 + "peak_bytes": 7692972, + "leaked_bytes": 3524955, + "total_allocations": 1132827 }, "reader_mp4": { - "peak_bytes": 4222601, - "leaked_bytes": 3345724, - "total_allocations": 4095915 + "peak_bytes": 4272788, + "leaked_bytes": 3406355, + "total_allocations": 4018933 }, "reader_wav": { - "peak_bytes": 4523095, - "leaked_bytes": 3355666, - "total_allocations": 742391 + "peak_bytes": 4573253, + "leaked_bytes": 3416313, + "total_allocations": 773409 }, "builder_sign_jpeg_legacy": { - "peak_bytes": 7785129, - "leaked_bytes": 3468507, - "total_allocations": 1041412 + "peak_bytes": 7844930, + "leaked_bytes": 3530986, + "total_allocations": 1046607 }, "builder_sign_jpeg_with_context": { - "peak_bytes": 7779538, - "leaked_bytes": 3463042, - "total_allocations": 1027485 + "peak_bytes": 7839456, + "leaked_bytes": 3524460, + "total_allocations": 1058202 }, "builder_sign_png_legacy": { - "peak_bytes": 8023081, - "leaked_bytes": 3468300, - "total_allocations": 3883115 + "peak_bytes": 8082891, + "leaked_bytes": 3530892, + "total_allocations": 3888499 }, "builder_sign_png_with_context": { - "peak_bytes": 8017008, - "leaked_bytes": 3462829, - "total_allocations": 3869515 + "peak_bytes": 8077349, + "leaked_bytes": 3524774, + "total_allocations": 3900456 }, "builder_sign_jpeg_parallel_split_pool": { - "peak_bytes": 45854797, - "leaked_bytes": 3840928, - "total_allocations": 1035646 + "peak_bytes": 45936892, + "leaked_bytes": 3893837, + "total_allocations": 1062520 }, "builder_sign_jpeg_parallel_split_barrier": { - "peak_bytes": 45844809, - "leaked_bytes": 3861014, - "total_allocations": 1037741 + "peak_bytes": 45905378, + "leaked_bytes": 3892593, + "total_allocations": 1061149 }, "builder_sign_png_parallel_split_pool": { - "peak_bytes": 46586728, - "leaked_bytes": 3868054, - "total_allocations": 3877696 + "peak_bytes": 46673225, + "leaked_bytes": 3929128, + "total_allocations": 3904507 }, "builder_sign_png_parallel_split_barrier": { - "peak_bytes": 46082548, - "leaked_bytes": 3879161, - "total_allocations": 3879780 + "peak_bytes": 46143013, + "leaked_bytes": 3910964, + "total_allocations": 3903125 }, "builder_sign_gif": { - "peak_bytes": 14635465, - "leaked_bytes": 3461270, - "total_allocations": 17017654 + "peak_bytes": 14696646, + "leaked_bytes": 3524513, + "total_allocations": 17048351 }, "builder_sign_heic": { - "peak_bytes": 4698434, - "leaked_bytes": 3469086, - "total_allocations": 1563419 + "peak_bytes": 4759642, + "leaked_bytes": 3532315, + "total_allocations": 1582361 }, "builder_sign_m4a": { - "peak_bytes": 18833496, - "leaked_bytes": 3469085, - "total_allocations": 5194205 + "peak_bytes": 18895243, + "leaked_bytes": 3532373, + "total_allocations": 5213365 }, "builder_sign_webp": { - "peak_bytes": 8991237, - "leaked_bytes": 3461271, - "total_allocations": 916145 + "peak_bytes": 9052463, + "leaked_bytes": 3524559, + "total_allocations": 950737 }, "builder_sign_avi": { - "peak_bytes": 7130933, - "leaked_bytes": 3461270, - "total_allocations": 89982012 + "peak_bytes": 7192106, + "leaked_bytes": 3524502, + "total_allocations": 90011891 }, "builder_sign_mp4": { - "peak_bytes": 6245379, - "leaked_bytes": 3469085, - "total_allocations": 3788717 + "peak_bytes": 6306630, + "leaked_bytes": 3532325, + "total_allocations": 3805992 }, "builder_sign_tiff": { - "peak_bytes": 13213169, - "leaked_bytes": 3461271, - "total_allocations": 10862700 + "peak_bytes": 13274395, + "leaked_bytes": 3524559, + "total_allocations": 10898003 }, "builder_sign_jpeg_parent_of": { - "peak_bytes": 14265295, - "leaked_bytes": 3461665, - "total_allocations": 2506107 + "peak_bytes": 14324563, + "leaked_bytes": 3525074, + "total_allocations": 2495210 }, "builder_sign_jpeg_component_of": { - "peak_bytes": 14266996, - "leaked_bytes": 3462012, - "total_allocations": 2551180 + "peak_bytes": 14325966, + "leaked_bytes": 3524803, + "total_allocations": 2538825 }, "builder_sign_jpeg_parent_and_component": { - "peak_bytes": 14665241, - "leaked_bytes": 3614613, - "total_allocations": 4523960 + "peak_bytes": 14605703, + "leaked_bytes": 3610907, + "total_allocations": 4464450 }, "builder_sign_jpeg_parent_and_component_mixed_mime": { - "peak_bytes": 14568780, - "leaked_bytes": 3462718, - "total_allocations": 5517180 + "peak_bytes": 14627596, + "leaked_bytes": 3525478, + "total_allocations": 5516963 }, "builder_sign_jpeg_two_components_same_mime": { - "peak_bytes": 14559274, - "leaked_bytes": 3564233, - "total_allocations": 4497379 + "peak_bytes": 14602589, + "leaked_bytes": 3610897, + "total_allocations": 4436718 }, "builder_sign_jpeg_two_components_mixed_mime": { - "peak_bytes": 14564839, - "leaked_bytes": 3461873, - "total_allocations": 5490592 + "peak_bytes": 14624276, + "leaked_bytes": 3525287, + "total_allocations": 5489138 }, "builder_sign_jpeg_archive_roundtrip": { - "peak_bytes": 14297571, - "leaked_bytes": 3481212, - "total_allocations": 3467149 + "peak_bytes": 14356429, + "leaked_bytes": 3545507, + "total_allocations": 3432412 }, "builder_from_archive_roundtrip": { - "peak_bytes": 14297349, - "leaked_bytes": 3480475, - "total_allocations": 3101030 + "peak_bytes": 14353258, + "leaked_bytes": 3542427, + "total_allocations": 3024501 }, "builder_with_archive_swap": { - "peak_bytes": 3681081, - "leaked_bytes": 3350198, - "total_allocations": 704373 + "peak_bytes": 3753525, + "leaked_bytes": 3421374, + "total_allocations": 744039 }, "reader_with_fragment_swap": { - "peak_bytes": 3778159, - "leaked_bytes": 3353205, - "total_allocations": 3787587 + "peak_bytes": 3839453, + "leaked_bytes": 3414104, + "total_allocations": 3806570 }, "with_fragment_pre_consume_rejection": { - "peak_bytes": 3778057, - "leaked_bytes": 3354795, - "total_allocations": 2094004 + "peak_bytes": 3841126, + "leaked_bytes": 3417826, + "total_allocations": 2128201 }, "with_archive_post_consume_failure": { - "peak_bytes": 3350600, - "leaked_bytes": 3308056, - "total_allocations": 175290 + "peak_bytes": 3423615, + "leaked_bytes": 3380332, + "total_allocations": 208578 }, "with_fragment_marshalling_error": { - "peak_bytes": 3708068, - "leaked_bytes": 3352335, - "total_allocations": 2077090 + "peak_bytes": 3767911, + "leaked_bytes": 3413267, + "total_allocations": 2094661 }, "with_fragment_mixed_outcomes": { - "peak_bytes": 3779175, - "leaked_bytes": 3356294, - "total_allocations": 2656787 + "peak_bytes": 3840297, + "leaked_bytes": 3417238, + "total_allocations": 2686269 }, "builder_to_archive_with_ingredient": { - "peak_bytes": 14069232, - "leaked_bytes": 3337316, - "total_allocations": 1830896 + "peak_bytes": 14142488, + "leaked_bytes": 3409388, + "total_allocations": 1790801 }, "builder_sign_jpeg_archive_roundtrip_ingredient_in_archive": { - "peak_bytes": 14287046, - "leaked_bytes": 3481977, - "total_allocations": 5879957 + "peak_bytes": 14345054, + "leaked_bytes": 3543673, + "total_allocations": 5769615 }, "builder_write_ingredient_archive": { - "peak_bytes": 14069289, - "leaked_bytes": 3337377, - "total_allocations": 1805304 + "peak_bytes": 14142437, + "leaked_bytes": 3409341, + "total_allocations": 1767419 }, "builder_sign_jpeg_add_ingredient_from_archive": { - "peak_bytes": 14133742, - "leaked_bytes": 3480831, - "total_allocations": 3415920 + "peak_bytes": 14207929, + "leaked_bytes": 3544945, + "total_allocations": 3383511 }, "builder_ingredient_archive_roundtrip": { - "peak_bytes": 14284443, - "leaked_bytes": 3480809, - "total_allocations": 5132060 + "peak_bytes": 14345163, + "leaked_bytes": 3545508, + "total_allocations": 5061203 }, "builder_sign_jpeg_two_ingredient_archives": { - "peak_bytes": 14134560, - "leaked_bytes": 3481604, - "total_allocations": 4215728 + "peak_bytes": 14208534, + "leaked_bytes": 3545923, + "total_allocations": 4185124 }, "reader_error_no_manifest": { - "peak_bytes": 3564471, - "leaked_bytes": 3323629, - "total_allocations": 276175 + "peak_bytes": 3622191, + "leaked_bytes": 3383889, + "total_allocations": 291735 }, "builder_error_invalid_manifest": { - "peak_bytes": 3352053, - "leaked_bytes": 3297079, - "total_allocations": 113926 + "peak_bytes": 3421406, + "leaked_bytes": 3365544, + "total_allocations": 126199 }, "reader_string_apis": { - "peak_bytes": 3978113, - "leaked_bytes": 3346111, - "total_allocations": 2287335 + "peak_bytes": 4039136, + "leaked_bytes": 3407512, + "total_allocations": 2238705 }, "signer_construction": { - "peak_bytes": 3350893, - "leaked_bytes": 3288137, - "total_allocations": 153245 + "peak_bytes": 3421644, + "leaked_bytes": 3358098, + "total_allocations": 159717 }, "builder_from_context_construction": { - "peak_bytes": 3350600, - "leaked_bytes": 3288582, - "total_allocations": 112688 + "peak_bytes": 3423152, + "leaked_bytes": 3360708, + "total_allocations": 146014 }, "fork_reader_collect": { - "peak_bytes": 3850530, - "leaked_bytes": 3353063, - "total_allocations": 1328122 + "peak_bytes": 3911936, + "leaked_bytes": 3413740, + "total_allocations": 1284294 }, "fork_contended_mutex": { - "peak_bytes": 7679019, - "leaked_bytes": 3482128, - "total_allocations": 67472694 + "peak_bytes": 7700288, + "leaked_bytes": 3510073, + "total_allocations": 66621221 }, "fork_thread_local_orphan": { - "peak_bytes": 3936170, - "leaked_bytes": 3439733, - "total_allocations": 1381055 + "peak_bytes": 4073730, + "leaked_bytes": 3581333, + "total_allocations": 1339630 }, "fork_gc_cycle": { - "peak_bytes": 3850434, - "leaked_bytes": 3353160, - "total_allocations": 1332098 + "peak_bytes": 3913068, + "leaked_bytes": 3414664, + "total_allocations": 1289268 }, "fork_parent_frees_after_fork": { - "peak_bytes": 5447584, - "leaked_bytes": 3350400, - "total_allocations": 24829257 + "peak_bytes": 5602620, + "leaked_bytes": 3423572, + "total_allocations": 23965279 }, "fork_child_closes_then_parent_frees": { - "peak_bytes": 5446620, - "leaked_bytes": 3350407, - "total_allocations": 24829254 + "peak_bytes": 5603711, + "leaked_bytes": 3424393, + "total_allocations": 23965271 }, "fork_child_sys_exit": { - "peak_bytes": 3850546, - "leaked_bytes": 3353234, - "total_allocations": 1335925 + "peak_bytes": 3911952, + "leaked_bytes": 3413956, + "total_allocations": 1301497 }, "fork_stream_cleanup": { - "peak_bytes": 3464063, - "leaked_bytes": 3291969, - "total_allocations": 105340 + "peak_bytes": 3532824, + "leaked_bytes": 3361106, + "total_allocations": 110397 }, "fork_swap_cleanup": { - "peak_bytes": 3681171, - "leaked_bytes": 3350696, - "total_allocations": 714376 + "peak_bytes": 3753679, + "leaked_bytes": 3421936, + "total_allocations": 754042 }, "fork_contended_mutex_swap": { - "peak_bytes": 7302379, - "leaked_bytes": 3475147, - "total_allocations": 35948516 + "peak_bytes": 7360409, + "leaked_bytes": 3525035, + "total_allocations": 37359891 }, "fork_contended_mutex_wrap": { - "peak_bytes": 7288748, - "leaked_bytes": 3463411, - "total_allocations": 34847186 + "peak_bytes": 7140341, + "leaked_bytes": 3522965, + "total_allocations": 34204380 }, "fork_consumed_signer": { - "peak_bytes": 3350894, - "leaked_bytes": 3288906, - "total_allocations": 175055 + "peak_bytes": 3421645, + "leaked_bytes": 3359803, + "total_allocations": 206540 }, "swap_chain_churn": { - "peak_bytes": 3681161, - "leaked_bytes": 3350287, - "total_allocations": 672537 + "peak_bytes": 3753669, + "leaked_bytes": 3421527, + "total_allocations": 679964 + }, + "deferred_teardown_flush_queue": { + "peak_bytes": 4103247, + "leaked_bytes": 3412690, + "total_allocations": 2448668 } } \ No newline at end of file diff --git a/tests/perf/scenarios.py b/tests/perf/scenarios.py index 23300aed..87dd512e 100644 --- a/tests/perf/scenarios.py +++ b/tests/perf/scenarios.py @@ -587,8 +587,8 @@ def scenario_reader_with_fragment_pre_consume_rejection( # Fail loudly: without these the scenario still runs when the # ownership logic regresses, and a rejection that stops being # recognised looks identical to a pass. - if not any(tag in str(e) for tag in - c2pa_module.ManagedResource._PRE_CONSUME_ERROR_TAGS): + if not c2pa_module.ManagedResource._is_pre_consume_rejection( + str(e)): raise AssertionError( f"expected a pre-consume rejection, got: {e}") from e if reader._handle is None: @@ -1297,6 +1297,40 @@ def scenario_swap_chain_churn(iterations: int = 100) -> None: context.close() +def scenario_deferred_teardown_flush_queue(iterations: int = 100) -> None: + """Close resources from inside an open native-error section, so their + teardowns defer onto one pending list and are drained together when the + section closes. + + Two resources per iteration rather than one: a single-element queue cannot + show a resource stranded behind its predecessor. + """ + signed_bytes = SIGNED_JPEG.read_bytes() + real_free = c2pa_module.ManagedResource._free_native_ptr + for _ in _iterate(iterations): + first = Reader("image/jpeg", io.BytesIO(signed_bytes)) + second = Reader("image/jpeg", io.BytesIO(signed_bytes)) + + freed = [] + c2pa_module.ManagedResource._free_native_ptr = staticmethod( + lambda ptr: (freed.append(ptr), real_free(ptr))[1]) + try: + with c2pa_module._native_section(): + first.close() + second.close() + # Fail loudly: a free here means the teardown was not deferred. + if freed: + raise AssertionError( + "teardown inside a section freed immediately " + "instead of deferring") + if len(freed) != 2: + raise AssertionError( + f"drain freed {len(freed)} of 2 deferred handles; " + f"the rest leak") + finally: + c2pa_module.ManagedResource._free_native_ptr = real_free + + def scenario_fork_swap_cleanup(iterations: int = 100) -> None: """Fork safety benchmark scenario: the handle a Builder owns at fork time came from with_archive(), which @@ -1403,6 +1437,7 @@ def scenario_fork_stream_cleanup(iterations: int = 100) -> None: "fork_contended_mutex_wrap": scenario_fork_contended_mutex_wrap, "fork_consumed_signer": scenario_fork_consumed_signer, "swap_chain_churn": scenario_swap_chain_churn, + "deferred_teardown_flush_queue": scenario_deferred_teardown_flush_queue, } diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index 2c1fb42e..a4e565ae 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -31,6 +31,7 @@ import shutil import ctypes import threading +import concurrent.futures # Suppress deprecation warnings warnings.simplefilter("ignore", category=DeprecationWarning) @@ -51,6 +52,15 @@ ALTERNATIVE_INGREDIENT_TEST_FILE = os.path.join(FIXTURES_DIR, "cloud.jpg") +def _fail_with_native_error(tag_bytes): + """Build a mock FFI callable that sets a native error and returns None. + """ + def _mock(*args): + c2pa_module._lib.c2pa_error_set_last(tag_bytes) + return None + return _mock + + def load_test_settings_json(): """ Load default (legacy) trust configuration test settings from a @@ -1362,7 +1372,6 @@ def test_sign_and_read_is_not_embedded(self): # Direct the Builder not to embed the manifest into the asset builder.set_no_embed() - with open(temp_file_path, "wb") as temp_file: manifest_data = builder.sign( signer, "image/jpeg", file, temp_file) @@ -8277,12 +8286,11 @@ def test_construction_failure_leaves_nothing_to_free(self): c2pa_module._lib.c2pa_builder_from_json = real_json def test_context_build_null_return_frees_builder(self): - # Set a pre-consume tag in the error slot to mock a pointer rejection. + # Mock a pointer rejection. settings = Settings() - c2pa_module._lib.c2pa_error_set_last( - b"UntrackedPointer: mocked pre-consume rejection") real_build = c2pa_module._lib.c2pa_context_builder_build - c2pa_module._lib.c2pa_context_builder_build = lambda ptr: None + c2pa_module._lib.c2pa_context_builder_build = _fail_with_native_error( + b"UntrackedPointer: mocked pre-consume rejection") try: with self.assertRaises(Error): Context(settings=settings) @@ -8341,6 +8349,180 @@ def test_consume_no_replacement_marks_consumed_on_other_error(self): self.assertIsNone(res._handle) self.assertEqual(res._lifecycle_state, LifecycleState.CLOSED) + def test_invoke_consume_success_does_not_consult_error_slot(self): + """A successful consuming call must not read the error slot at all: + only a failure inspects it.""" + res = self._FakeHandleResource() + res._activate(0xCAFE) + + res._consume_no_replacement(lambda h: 0, "set failed: {}") + + self.assertIsNone(c2pa_module._read_native_error()) + + def test_consume_no_replacement_retains_on_tag_set_by_the_call_itself(self): + """Only a *stale* tag left over from before the call is the + thing being defended against.""" + res = self._FakeHandleResource() + res._activate(0xCAFE) + + def fake_call(handle): + c2pa_module._lib.c2pa_error_set_last( + b"UntrackedPointer: rejected by the call itself") + return -1 + + with self.assertRaises(Error): + res._consume_no_replacement(fake_call, "set failed: {}") + + # Rejected before ownership transferred: handle retained. + self.assertEqual(res._handle, 0xCAFE) + self.assertEqual(res._lifecycle_state, LifecycleState.ACTIVE) + self.assertEqual(self.freed, []) + res.close() + self.assertEqual(self.freed, [0xCAFE]) + + def test_native_section_defers_unrelated_finalizer_free(self): + """A finalizer for a completely unrelated resource firing mid + native-call must not free immediately. + """ + victim = self._FakeHandleResource() + victim._activate(0xCAFE) + bystander = self._FakeHandleResource() + bystander._activate(0xB00B) + + def polluting_free(ptr): + self.freed.append(ptr) + # Freeing and untracked/ pointer writes its own error into the + # same thread-local slot. + c2pa_module._lib.c2pa_error_set_last( + "Other: UntrackedPointer: {:#x}".format(ptr).encode()) + return -1 + ManagedResource._free_native_ptr = staticmethod(polluting_free) + + def ffi_call(handle): + nonlocal bystander + del bystander # last reference dropped: __del__ fires right here + return None # the real call failed but set no error of its own + + # A bare section, not victim._native_call(): the consume needs the + # error section, not a borrow on its own handle. _ensure_not_borrowed + # refuses a consume nested in a _native_call() on the same resource. + with c2pa_module._native_section(): + with self.assertRaises(Error): + victim._consume_no_replacement(ffi_call, "op failed: {}") + + self.assertIsNone( + victim._handle, + "victim was wrongly retained") + self.assertEqual(victim._lifecycle_state, LifecycleState.CLOSED) + # The bystander's free is deferred to the section close, so it + # runs after the consuming call, before the victim's free. + self.assertEqual(self.freed, [0xB00B, 0xCAFE], + "deferred free did not run once, before victim's") + + def test_teardown_deferred_by_own_inflight_and_section_together(self): + """A resource blocked by its own handle being in-flight, + and a wholly separate native-error section is also open on this thread + must not free until both clear, and must free exactly once.""" + res = self._FakeHandleResource() + res._activate(0xCAFE) + + call_cm = res._native_call() + call_cm.__enter__() + try: + section_cm = c2pa_module._native_section() + section_cm.__enter__() + try: + res.close() + self.assertEqual(res._lifecycle_state, LifecycleState.CLOSED) + self.assertEqual(self.freed, [], + "freed while still in flight") + finally: + section_cm.__exit__(None, None, None) + # The independent section closed, but res's own in-flight + # guard is still up: still not freed. + self.assertEqual(self.freed, [], + "flushed while the in-flight guard still held") + finally: + call_cm.__exit__(None, None, None) + # Both gates clear only once native_call's own exit drops inflight + # to 0 -- that is what should finally trigger the free. + self.assertEqual(self.freed, [0xCAFE]) + + def test_nested_native_sections_flush_only_at_outermost_close(self): + """A native-error section opened inside another, already-open one + on the same thread must not flush anything until the outermost + one closes.""" + res = self._FakeHandleResource() + res._activate(0xCAFE) + + outer = c2pa_module._native_section() + outer.__enter__() + try: + inner = c2pa_module._native_section() + inner.__enter__() + try: + res.close() + self.assertEqual(self.freed, []) + finally: + inner.__exit__(None, None, None) + # Inner closed, outer is still open: still deferred. + self.assertEqual(self.freed, [], + "inner section flushed before the outer closed") + finally: + outer.__exit__(None, None, None) + self.assertEqual(self.freed, [0xCAFE]) + + def test_native_section_flush_isolates_exceptions(self): + """One deferred free raising during a section's flush must not + stop the rest of that flush from running.""" + good = self._FakeHandleResource() + good._activate(0xC0FFEE) + bad = self._FakeHandleResource() + bad._activate(0xBAD) + + def flaky_free(ptr): + if ptr == 0xBAD: + raise RuntimeError("simulated free failure") + self.freed.append(ptr) + return 0 + ManagedResource._free_native_ptr = staticmethod(flaky_free) + + with self.assertLogs('c2pa', level='ERROR') as captured: + with c2pa_module._native_section(): + bad.close() + good.close() + + self.assertEqual(self.freed, [0xC0FFEE], + "a failing deferred free stopped the rest") + self.assertTrue( + any('Failed to free native' in line + for line in captured.output), + "the failing deferred free was not logged: " + "{}".format(captured.output)) + + def test_stale_error_not_misattributed_after_preset_error(self): + """A stale tag left by an earlier, unrelated call on this thread + must not be read as this call's own error.""" + # A stale tag from an earlier, unrelated call. + c2pa_module._lib.c2pa_error_set_last( + b"Other: UntrackedPointer: 0xdeadbeef") + + res = self._FakeHandleResource() + res._activate(0xCAFE) + + # Fails without setting any error of its own. + # The sentinel inside _invoke_consume must have cleared + # the stale tag, so this routes to the "no error of our own" branch. + with self.assertRaises(Error): + res._consume_no_replacement(lambda h: -1, "op failed: {}") + + # A misattributed stale tag would have matched + # _PRE_CONSUME_ERROR_TAGS and left the resource ACTIVE. + self.assertIsNone(res._handle) + self.assertEqual(res._lifecycle_state, LifecycleState.CLOSED) + self.assertEqual(self.freed, [0xCAFE], + "unknown ownership must free, not drop the handle") + class TestManagedResourceObjects(TestContextAPIs): """Tests native resource handling management when managed manually. @@ -8602,9 +8784,9 @@ def test_builder_with_archive_null_return_marks_consumed(self): # Mimic a non-tag error: native took ownership then failed and dropped # the value itself, so the handle is marked consumed, not freed. - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") real_call = c2pa_module._lib.c2pa_builder_with_archive - c2pa_module._lib.c2pa_builder_with_archive = lambda b, s: None + c2pa_module._lib.c2pa_builder_with_archive = _fail_with_native_error( + b"Other: mocked test error") # Instrument before the failure... freed = self._instrument_frees() @@ -8638,11 +8820,9 @@ def test_reader_with_fragment_null_return_marks_consumed(self): # Mimic a non-tag error: native took ownership then failed and dropped # the value itself, so the handle is marked consumed, not freed. - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") - real_call = c2pa_module._lib.c2pa_reader_with_fragment - c2pa_module._lib.c2pa_reader_with_fragment = ( - lambda r, f, s, frag: None) + c2pa_module._lib.c2pa_reader_with_fragment = _fail_with_native_error( + b"Other: mocked test error") # Instrument before failure so any free would be counted. freed = self._instrument_frees() @@ -8710,11 +8890,11 @@ def _raise(*_args): @staticmethod def _is_pre_consume_rejection(error_message): - """True if this native error means ownership never transferred.""" + """True if this native error means ownership never transferred. + """ if not error_message: return False - return any(tag in error_message - for tag in ManagedResource._PRE_CONSUME_ERROR_TAGS) + return ManagedResource._is_pre_consume_rejection(error_message) def _stale_reader_handle(self): """A freed, untracked pointer, captured before close() nulls it. @@ -8740,31 +8920,6 @@ def _untracked_reader_handle(): return (ctypes.cast(buf, ctypes.POINTER(c2pa_module.C2paReader)), buf) - def test_with_fragment_pre_consume_rejection_keeps_handle(self): - # Rejected before native lib took ownership, - # so nothing was consumed and the handle is still ours. - 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) - real_handle = reader._handle - - reader._handle = self._stale_reader_handle() - try: - with open(init_path, "rb") as init, \ - open(fragment_path, "rb") as frag: - with self.assertRaises(Error) as caught: - reader.with_fragment("video/mp4", init, frag) - finally: - reader._handle = real_handle - - self.assertIn("UntrackedPointer", str(caught.exception)) - # Ownership never transferred, so the resource stays usable. - self.assertIsNotNone(reader._handle) - self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) - self.assertTrue(reader.json()) - reader.close() - def test_with_fragment_pre_consume_rejection_does_not_leak(self): # A handle dropped on this path leaks one reader per call. init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") @@ -8805,81 +8960,6 @@ def _reader_from_context(self): "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. @@ -8971,10 +9051,9 @@ def test_unknown_failure_drops_handle_without_freeing(self): consumed_handle = reader._handle # Simulate an error being set - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") real_call = c2pa_module._lib.c2pa_reader_with_fragment - c2pa_module._lib.c2pa_reader_with_fragment = ( - lambda r, f, s, frag: None) + c2pa_module._lib.c2pa_reader_with_fragment = _fail_with_native_error( + b"Other: mocked test error") try: with open(init_path, "rb") as init, \ open(fragment_path, "rb") as frag: @@ -9085,17 +9164,14 @@ def test_perf_scenario_bogus_handle_is_rejected(self): reader.close() def test_every_null_return_sets_its_own_error(self): - # Reading the slot without clearing it is only sound because every - # null return sets an error. Check each path reports its own. + # Each null-returning path must report the error it set itself, never + # one left behind by an earlier call. init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") - # Leave a recognisable error behind, so anything stale shows up. - try: - Reader("image/jpeg", io.BytesIO(b"not an image")).json() - except Error: - pass - self.assertIn("NotSupported", c2pa_module._read_native_error() or "") + # Set a recognizable error, so anything stale shows up below. + c2pa_module._lib.c2pa_error_set_last( + b"NotSupported: planted by the test") # Pre-consume rejection: reports UntrackedPointer, not NotSupported. with open(init_path, "rb") as init: @@ -9165,21 +9241,19 @@ def worker(): self.assertEqual(problems, [], "ownership was misjudged under concurrency") - def test_reading_the_native_error_does_not_empty_the_slot(self): - # c2pa_error() peeks, so nothing Python can call empties the slot. - # _consume_and_swap depends on this. - try: - Reader("image/jpeg", io.BytesIO(b"not an image")).json() - except Error: - pass + def test_reading_the_native_error_consumes_it(self): + # c2pa_error() itself peeks, so _read_native_error marks the slot as + # carrying no error once it has read one. + # An error belongs to the caller that observes it; + # leaving it readable lets a later, unrelated failure report it as its own. + c2pa_module._lib.c2pa_error_set_last(b"Io: read me exactly once") first = c2pa_module._read_native_error() self.assertTrue(first, "expected a native error to have been set") - self.assertEqual( - c2pa_module._read_native_error(), first, - "reading emptied the native slot; the comments in " - "_consume_and_swap about a persistent error are now wrong") + self.assertIsNone( + c2pa_module._read_native_error(), + "the native error stayed readable after being reported once") def test_read_native_error_returns_none_for_an_empty_message(self): # c2pa_error() returns an owned pointer to "" when no error is set, @@ -9197,22 +9271,29 @@ def test_read_native_error_returns_none_for_an_empty_message(self): finally: c2pa_module._lib.c2pa_error = original - def test_mocked_null_without_error_is_a_known_limitation(self): - # A null with no error of its own is the case that breaks: the slot - # still holds whatever came before. No native path does this, so it - # is pinned here rather than defended in _consume_and_swap. + def test_null_return_with_no_native_error_is_treated_as_consumed(self): + # A null with no error of its own used to be the case that broke: + # the slot still held whatever an unrelated, earlier call on this same + # (pooled) thread left behind, and a stale UntrackedPointer/ + # WrongPointerType tag would make this call believe it still owned a + # handle the native side already dropped. init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + # A stale, unrelated tag left by a prior call on this thread. c2pa_module._lib.c2pa_error_set_last( b"UntrackedPointer: 0xdeadbeef") with open(init_path, "rb") as init: reader = Reader("video/mp4", init) + consumed_handle = reader._handle real_call = c2pa_module._lib.c2pa_reader_with_fragment + # The fake native call sets no error of its own, + # the planted sentinel _invoke_consume is left in the slot. c2pa_module._lib.c2pa_reader_with_fragment = ( lambda r, f, s, frag: None) + freed = self._instrument_frees() try: with open(init_path, "rb") as init, \ open(fragment_path, "rb") as frag: @@ -9220,16 +9301,14 @@ def test_mocked_null_without_error_is_a_known_limitation(self): reader.with_fragment("video/mp4", init, frag) finally: c2pa_module._lib.c2pa_reader_with_fragment = real_call - # Nothing clears the slot, so a planted tag would follow other - # tests around and change how their failures are classified. - c2pa_module._lib.c2pa_error_set_last( - b"Other: cleared by test teardown") - # The stale tag wins, so the handle is kept. Safe here (the mock - # consumed nothing), and the reader is still usable. - self.assertIsNotNone(reader._handle) - self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) - reader.close() + # The sentinel survived, not the stale tag. + self.assertIsNone(reader._handle) + self.assertEqual(reader._lifecycle_state, LifecycleState.CLOSED) + # Ownership is unknown, so the handle is freed once. c2pa_free + # returns -1 if native had already taken the value. + self.assertEqual(self._free_count(freed, consumed_handle), 1, + "unknown-ownership handle was not freed once") # Backfilling a pointer minted by a direct FFI call. Builder.from_archive # is the only production caller of _wrap_native_handle, so these are the @@ -9378,10 +9457,9 @@ def test_consumed_reader_closes_backing_file(self): self.assertFalse(backing_file.closed) # Simulate an error being set - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") real_call = c2pa_module._lib.c2pa_reader_with_fragment - c2pa_module._lib.c2pa_reader_with_fragment = ( - lambda r, f, s, frag: None) + c2pa_module._lib.c2pa_reader_with_fragment = _fail_with_native_error( + b"Other: mocked test error") try: with open(DEFAULT_TEST_FILE, "rb") as main, \ open(DEFAULT_TEST_FILE, "rb") as frag: @@ -9400,9 +9478,9 @@ def test_consumed_builder_releases_context(self): archive = self._make_archive() # Simulate an error being set - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") real_call = c2pa_module._lib.c2pa_builder_with_archive - c2pa_module._lib.c2pa_builder_with_archive = lambda b, s: None + c2pa_module._lib.c2pa_builder_with_archive = _fail_with_native_error( + b"Other: mocked test error") try: with self.assertRaises(Error): builder.with_archive(archive) @@ -9449,10 +9527,9 @@ def test_consumed_reader_clears_caches(self): self.assertIsNotNone(reader._manifest_json_str_cache) # Simulate an error being set - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") real_call = c2pa_module._lib.c2pa_reader_with_fragment - c2pa_module._lib.c2pa_reader_with_fragment = ( - lambda r, f, s, frag: None) + c2pa_module._lib.c2pa_reader_with_fragment = _fail_with_native_error( + b"Other: mocked test error") try: with open(DEFAULT_TEST_FILE, "rb") as main, \ open(DEFAULT_TEST_FILE, "rb") as frag: @@ -9524,6 +9601,36 @@ def _boom(*args): self.assertIs(ctx.exception.__cause__, sentinel, "signing error dropped the original exception") + def test_sign_reports_the_native_error_it_set(self): + """sign() reads its error in a later section than the call itself. + The signing call runs inside one _native_call() block and the result + check runs in a separate _native_section() afterwards, so anything + that marks the slot as carrying no error on section exit would discard + the real message between the two. + """ + builder = Builder(self.test_manifest) + signer = self._ctx_make_signer() + self.addCleanup(signer.close) + + real_sign = c2pa_module._lib.c2pa_builder_sign + + def _fail(*args): + c2pa_module._lib.c2pa_error_set_last( + b"Signature: native signing refused") + return -1 + + c2pa_module._lib.c2pa_builder_sign = _fail + try: + with self.assertRaises(Error) as ctx: + builder.sign(signer, "image/jpeg", + io.BytesIO(b"x"), io.BytesIO()) + finally: + c2pa_module._lib.c2pa_builder_sign = real_sign + + self.assertIn("native signing refused", str(ctx.exception), + "the native signing error was lost before it was read") + self.assertIsInstance(ctx.exception, Error.Signature) + class TestErrorPlumbing(unittest.TestCase): """Covers the error helpers themselves, which had no direct tests.""" @@ -9655,6 +9762,338 @@ def test_supported_mime_types_reports_the_native_message(self): c2pa_module._get_supported_mime_types(lambda count: None, None) self.assertIn("mime lookup failed", str(ctx.exception)) + def test_reading_an_error_does_not_leave_it_readable(self): + """An error is reportable once, by the reader that observes it. + """ + self._set_native_error("Io: read me once") + + self.assertEqual( + c2pa_module._read_native_error(), "Io: read me once") + self.assertIsNone( + c2pa_module._read_native_error(), + "the same native error was reported a second time") + + def test_handled_error_does_not_survive_later_operations(self): + """A caught failure must not leave its error in-place + (tests the slot is cleaned up). + """ + with self.assertRaises(Error): + Reader("image/jpeg", io.BytesIO(b"not an image")) + + for _ in range(20): + c2pa_module.Stream(io.BytesIO(b"x")) + + self.assertIsNone( + c2pa_module._read_native_error(), + "a handled error was still resident after 20 successful calls") + + def test_later_failure_does_not_inherit_a_handled_errors_type(self): + """A failure with no error of its own must not see an older one. + """ + with self.assertRaises(Error) as first: + Reader("image/jpeg", io.BytesIO(b"not an image")) + self.assertIsInstance(first.exception, Error.NotSupported) + + with self.assertRaises(Error) as second: + c2pa_module._check_ffi_operation_result( + None, "Later unrelated failure: {}") + + self.assertNotIsInstance( + second.exception, Error.NotSupported, + "the later failure inherited the handled error's type") + self.assertIn("Unknown error", str(second.exception)) + self.assertNotIn( + "type is unsupported", str(second.exception), + "the later failure reported the handled error's message") + + def test_the_no_native_error_sentinel_never_reaches_a_caller(self): + """The sentinel is an internal marker, not a message for users.""" + sentinel = c2pa_module._NATIVE_NO_ERROR_TEXT + + c2pa_module._mark_sentinel_no_native_error() + self.assertIsNone( + c2pa_module._read_native_error(), + "the sentinel was reported as if it were a native error") + + c2pa_module._mark_sentinel_no_native_error() + with self.assertRaises(Error) as ctx: + c2pa_module._check_ffi_operation_result(None, "fallback: {}") + self.assertNotIn(sentinel, str(ctx.exception)) + self.assertIn("Unknown error", str(ctx.exception)) + + def test_mark_sentinel_writes_the_learned_text(self): + c2pa_module._mark_sentinel_no_native_error() + raw = c2pa_module._lib.c2pa_error() + try: + text = ctypes.string_at(raw).decode('utf-8') + finally: + c2pa_module._lib.c2pa_string_free(raw) + self.assertEqual(text, c2pa_module._NATIVE_NO_ERROR_TEXT) + + def test_read_native_error_maps_sentinel_to_none(self): + c2pa_module._mark_sentinel_no_native_error() + self.assertIsNone(c2pa_module._read_native_error()) + + def test_read_native_error_marks_the_slot_when_the_pointer_is_null(self): + """A NULL from c2pa_error must still leave the slot marked. + + c2pa_error returns NULL when the stored message cannot be rendered as + a C string. The message stays in the thread-local slot, which is + sticky, so returning without planting the marker leaves that message + readable by the next call that fails without setting an error of its + own, which then reports it as its own failure. + """ + c2pa_module._lib.c2pa_error_set_last(b"Io: unreadable original") + + original = c2pa_module._lib.c2pa_error + try: + c2pa_module._lib.c2pa_error = lambda: None + self.assertIsNone( + c2pa_module._read_native_error(), + "a NULL pointer must read as no error") + finally: + c2pa_module._lib.c2pa_error = original + + self.assertIsNone( + c2pa_module._read_native_error(), + "the message left behind by the NULL branch stayed readable " + "and is now reportable by an unrelated later failure") + + def test_a_failure_after_a_null_read_does_not_inherit_the_old_message(self): + """The message surviving a NULL read must not become someone's error.""" + c2pa_module._lib.c2pa_error_set_last(b"Io: belongs to an earlier call") + + original = c2pa_module._lib.c2pa_error + try: + c2pa_module._lib.c2pa_error = lambda: None + c2pa_module._read_native_error() + finally: + c2pa_module._lib.c2pa_error = original + + with self.assertRaises(Error) as ctx: + c2pa_module._check_ffi_operation_result( + None, "Later unrelated failure: {}") + + self.assertNotIn( + "belongs to an earlier call", str(ctx.exception), + "a later failure reported a message left by an earlier call") + self.assertIn("Unknown error", str(ctx.exception)) + + def test_every_real_rejection_wording_is_classified_as_pre_consume(self): + """The four tags arrive bare or behind the "Other: " wrapper.""" + wrapper = c2pa_module.ManagedResource._NATIVE_ERROR_WRAPPER + classify = c2pa_module.ManagedResource._is_pre_consume_rejection + + for tag in c2pa_module.ManagedResource._PRE_CONSUME_ERROR_TAGS: + bare = f"{tag} some detail" + wrapped = f"{wrapper}{tag} some detail" + self.assertTrue( + classify(bare), + f"a bare {tag} rejection was read as a consumed handle") + self.assertTrue( + classify(wrapped), + f"a wrapped {tag} rejection was read as a consumed handle") + + def test_caller_text_quoting_a_tag_is_not_a_rejection(self): + """A tag inside the message body describes the caller's input. + + Native errors quote caller-supplied strings verbatim: a JSON parse + failure repeats the offending value, an Io failure names the path. + Reading one of those as a pre-consume rejection hands the resource back + as usable after native may already own and have dropped its handle. + """ + classify = c2pa_module.ManagedResource._is_pre_consume_rejection + + forged = ( + 'Json: invalid type: string "NullParameter: x", expected a ' + 'sequence at line 1 column 43', + 'Json: invalid type: string "WrongPointerType: y", expected a ' + 'sequence at line 1 column 46', + "Io: cannot open /tmp/UntrackedPointer: 0xdead.jpg", + "Other: manifest text mentions InvalidBufferSize: in passing", + ) + for message in forged: + self.assertFalse( + classify(message), + f"caller text was read as a pointer rejection: {message!r}") + + def test_caller_text_quoting_a_tag_reaches_the_error_slot(self): + """The forged wording above is what the library really produces.""" + c2pa_module._lib.c2pa_builder_from_json( + b'{"claim_generator_info": "NullParameter: injected"}') + message = c2pa_module._read_native_error() + + self.assertIn( + "NullParameter:", message, + "caller text no longer reaches the error slot verbatim, so this " + "test no longer exercises the case it was written for") + self.assertFalse( + c2pa_module.ManagedResource._is_pre_consume_rejection(message), + f"a caller-supplied string forged a pointer rejection: {message!r}") + + def test_a_failing_flush_does_not_strand_the_rest_of_the_queue(self): + """One resource raising must not skip the resources queued behind it. + """ + flushed = [] + + class Recorder: + def __init__(self, name, raises=None): + self.name = name + self.raises = raises + + def _maybe_flush_pending(self): + if self.raises is not None: + raise self.raises + flushed.append(self.name) + + first = Recorder("first") + middle = Recorder("middle", raises=KeyboardInterrupt()) + last = Recorder("last") + + with self.assertRaises(KeyboardInterrupt): + with c2pa_module._native_section(): + for resource in (first, middle, last): + c2pa_module._register_for_section_flush(resource) + + self.assertEqual( + flushed, ["first", "last"], + "a resource queued behind a failing one was never flushed, " + "so its handle leaks") + + def test_a_failing_flush_still_reports_the_first_exception(self): + """Draining the queue must not swallow the failure. + """ + flushed = [] + + class Recorder: + def __init__(self, name, raises=None): + self.name = name + self.raises = raises + + def _maybe_flush_pending(self): + if self.raises is not None: + raise self.raises + flushed.append(self.name) + + with self.assertRaises(RuntimeError) as ctx: + with c2pa_module._native_section(): + for resource in ( + Recorder("boom", raises=RuntimeError("first failure")), + Recorder("survivor"), + Recorder("later", raises=RuntimeError("second failure"))): + c2pa_module._register_for_section_flush(resource) + + self.assertIn("first failure", str(ctx.exception)) + self.assertEqual( + flushed, ["survivor"], + "a resource between two failing ones was never flushed") + + def test_runtime_does_not_call_error_set_last(self): + """The marker mechanism must not depend on c2pa_error_set_last, + so this module loads against native builds that lack it.""" + for fn in (c2pa_module.ManagedResource._invoke_consume, + c2pa_module._read_native_error, + c2pa_module._mark_sentinel_no_native_error): + self.assertNotIn( + 'c2pa_error_set_last', inspect.getsource(fn)) + + +class TestMarkerOutlivesPointerConsumptionSemantics(unittest.TestCase): + """The marker is needed for reasons independent of pointer ownership. + + The native error slot is sticky and thread-local, so failure paths + that carry no still need to tell an error this call set from an + earlier, unrelated call left behind. + """ + + def setUp(self): + # Leave no message from an earlier test in this thread's slot. + c2pa_module._mark_sentinel_no_native_error() + + def test_non_consuming_failure_does_not_inherit_a_read_error(self): + c2pa_module._lib.c2pa_error_set_last(b"Signature: earlier task") + # The rightful owner reports it, which re-marks the slot. + self.assertEqual( + c2pa_module._read_native_error(), "Signature: earlier task") + + # A later, unrelated failure that sets no error of its own must + # report its own fallback, not the message above. + with self.assertRaises(Error) as ctx: + c2pa_module._check_ffi_operation_result( + 0, "later op failed: {}", check=lambda r: r == 0) + + self.assertNotIn("earlier task", str(ctx.exception)) + self.assertIn("Unknown error", str(ctx.exception)) + self.assertNotIsInstance(ctx.exception, Error.Signature) + + def test_settings_set_failure_reports_its_own_error(self): + settings = Settings() + self.addCleanup(settings.close) + + c2pa_module._lib.c2pa_error_set_last(b"Signature: earlier task") + self.assertEqual( + c2pa_module._read_native_error(), "Signature: earlier task") + + with self.assertRaises(Error) as ctx: + settings.set("builder.thumbnail.enabled", "not-a-json-value") + + self.assertNotIn("earlier task", str(ctx.exception)) + + def test_marker_is_per_thread_across_pooled_reuse(self): + """The slot is thread-local, so a pooled worker must not hand one + task's error to the next task that runs on it.""" + def failing_task(): + c2pa_module._lib.c2pa_error_set_last(b"Io: first task") + return c2pa_module._read_native_error() + + def quiet_task(): + # Sets no error; must not see the previous task's message. + return c2pa_module._read_native_error() + + # One worker guarantees both tasks run on the same OS thread. + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: + self.assertEqual(pool.submit(failing_task).result(), + "Io: first task") + self.assertIsNone( + pool.submit(quiet_task).result(), + "a pooled thread carried an error across unrelated tasks") + + def test_one_thread_marker_does_not_clear_another_threads_error(self): + """Marking on one thread must leave another thread's pending error + readable: the slot is per thread, and so is the marker.""" + set_on_worker = threading.Event() + marked_on_main = threading.Event() + seen = {} + + def worker(): + c2pa_module._lib.c2pa_error_set_last(b"Io: worker error") + set_on_worker.set() + self.assertTrue(marked_on_main.wait(5)) + seen["worker"] = c2pa_module._read_native_error() + + thread = threading.Thread(target=worker, daemon=True) + thread.start() + self.assertTrue(set_on_worker.wait(5)) + + c2pa_module._mark_sentinel_no_native_error() + marked_on_main.set() + thread.join(5) + + self.assertEqual(seen.get("worker"), "Io: worker error") + + def test_marker_path_is_reached_without_any_consuming_call(self): + """The non-consuming path reaches the marker through _read_native_error, + never through _invoke_consume.""" + self.assertIn("_read_native_error", + inspect.getsource( + c2pa_module._check_ffi_operation_result)) + self.assertNotIn("_invoke_consume", + inspect.getsource( + c2pa_module._check_ffi_operation_result)) + # _read_native_error is what re-marks the slot after every read. + self.assertIn("_mark_sentinel_no_native_error", + inspect.getsource(c2pa_module._read_native_error)) + class TestErrorsStillRaiseAfterCleanup(unittest.TestCase): """Each surface that lost a _clear_error_state() call still reports.""" @@ -9736,28 +10175,6 @@ def bad_marshal(handle): self.assertIsNotNone(resource._handle) self.assertEqual(resource._lifecycle_state, LifecycleState.ACTIVE) - def test_pre_consume_rejection_restores_the_resource(self): - """A handle native rejected before taking ownership stays usable. - - The reservation is held until _raise_consume_failure classifies the - error, so no other thread sees the resource as ACTIVE while its - ownership is still undetermined. - """ - resource = Settings() - self.freed.clear() - real_read = c2pa_module._read_native_error - c2pa_module._read_native_error = ( - lambda: "Other: UntrackedPointer: 0x1234") - try: - with self.assertRaises(Error): - resource._consume_no_replacement(lambda h: 1, "consume: {}") - finally: - c2pa_module._read_native_error = real_read - - self.assertEqual(resource._lifecycle_state, LifecycleState.ACTIVE) - self.assertIsNotNone(resource._handle) - self.assertEqual(self.freed, []) - def test_post_consume_failure_keeps_the_resource_closed(self): """An error without a pre-consume tag means native took ownership. diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index 20537e46..16e409e4 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -34,7 +34,7 @@ from c2pa import Builder, C2paError as Error, Reader, C2paSigningAlg as SigningAlg, C2paSignerInfo, Signer, sdk_version # noqa: E501 from c2pa import Context, Settings -from c2pa.c2pa import ManagedResource, Stream, LifecycleState +from c2pa.c2pa import ManagedResource, Stream, LifecycleState, _native_section import c2pa.c2pa as c2pa_module from c2pa.lib import is_foreign_process, record_owner_pid @@ -4157,31 +4157,35 @@ def test_no_nested_op_locks(self): 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 + real_state_lock = ManagedResource._state_lock + + def make_tracking(real): + def tracking(resource): + lock = real(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() + return tracking + + ManagedResource._lock = make_tracking(real_lock) + ManagedResource._state_lock = make_tracking(real_state_lock) try: reader = Reader("image/jpeg", io.BytesIO(data)) reader.json() @@ -4191,6 +4195,7 @@ def __exit__(self, *exc): reader.close() finally: ManagedResource._lock = real_lock + ManagedResource._state_lock = real_state_lock self.assertEqual(violations, [], "a thread held two operation locks at once") @@ -4232,6 +4237,49 @@ def closer_worker(): self._join_all(threads, "concurrent storm") self.assertEqual(errors, []) + def test_native_section_deferred_free_is_thread_local(self): + """Two threads each with their own open native-error section: one + thread's section closing must not flush a free deferred inside + the other thread's still-open section. + """ + freed = self._counted_free() + resource = _ConcreteResource() + resource._activate(0x1001) + + thread_ready = threading.Event() + release_thread = threading.Event() + + def worker(): + with _native_section(): + resource.close() + thread_ready.set() + release_thread.wait(self.JOIN_TIMEOUT) + # Flush happens here, on the worker thread, once its own + # section closes. + + thread = threading.Thread(target=worker) + thread.start() + try: + self.assertTrue( + thread_ready.wait(self.JOIN_TIMEOUT), + "worker thread did not reach its open section in time") + + # A section opened and closed entirely on this (main) thread, + # while the worker's section is still open on its own thread. + with _native_section(): + pass + + self.assertEqual( + freed, [], + "a different thread's section flushed this thread's " + "pending resource") + finally: + release_thread.set() + self._join_all([thread], "native-section worker") + + self.assertEqual(freed, [0x1001], + "worker thread's own section never flushed") + def _counted_free(self): """Patch _free_native_ptr to count frees; returns the list.""" freed = [] @@ -4893,7 +4941,6 @@ def visit(node, active): "borrowed handles used without their own guard:\n " + "\n ".join(unguarded)) - def test_consume_during_concurrent_sign_does_not_crash(self): """Consuming a shared Signer must not free it under a live sign. @@ -5065,6 +5112,99 @@ def test_context_close_during_sign_defers_teardown(self): "the deferred teardown never ran") self.assertIsNone(context._pending_teardown) + def test_deferred_teardown_survives_a_flush_inside_a_section(self): + """A flush blocked by a section must re-register, not drop the free. + + The teardown defers on _inflight, so it is queued for the in-flight + call rather than for a section. When that call finishes inside a + section opened later on this thread, the flush cannot free yet, and + without re-registering nothing would ever free this handle. + """ + context = Context() + freed = [] + real_free = ManagedResource._free_native_ptr + ManagedResource._free_native_ptr = staticmethod( + lambda ptr: (freed.append(ptr), real_free(ptr))[1]) + try: + with context._native_call(): + closer = threading.Thread(target=context.close) + closer.start() + closer.join() + self.assertIsNotNone( + context._pending_teardown, + "close() during a native call should defer") + section = _native_section() + section.__enter__() + + self.assertEqual( + freed, [], + "the flush freed while a native section was still open") + self.assertIsNotNone( + context._pending_teardown, + "the deferral was dropped instead of re-registered") + + section.__exit__(None, None, None) + self.assertEqual( + len(freed), 1, + "the deferred teardown was stranded and never freed") + self.assertIsNone(context._pending_teardown) + finally: + ManagedResource._free_native_ptr = real_free + + def test_abort_consume_leaves_a_queued_teardown_closed(self): + """A resource whose free is already queued must not become usable. + + The deferred free still runs when the section drains, so restoring + ACTIVE would hand the caller a resource that closes underneath it. + """ + context = Context() + with _native_section(): + context.close() + self.assertIsNotNone(context._pending_teardown) + + context._abort_consume(LifecycleState.ACTIVE) + self.assertEqual( + context._lifecycle_state, LifecycleState.CLOSED, + "a resource with a queued teardown was revived") + self.assertFalse( + context.is_valid, + "a resource with a queued teardown reported itself usable") + + def test_section_drain_error_does_not_mask_the_body_error(self): + """The body's exception is what the caller asked for, so it wins.""" + + class FlushRaises: + _pending_teardown = True + + def _maybe_flush_pending(self): + raise RuntimeError("flush failed") + + class BodyError(Exception): + pass + + with self.assertLogs('c2pa', level='ERROR') as logs: + with self.assertRaises(BodyError): + with _native_section(): + c2pa_module._register_for_section_flush(FlushRaises()) + raise BodyError("the error the caller cares about") + + self.assertTrue( + any("flush failed" in line for line in logs.output), + "the flush failure was swallowed instead of logged") + + def test_section_drain_error_still_raises_when_the_body_succeeds(self): + """With no body error, a failed flush is still reported.""" + + class FlushRaises: + _pending_teardown = True + + def _maybe_flush_pending(self): + raise RuntimeError("flush failed") + + with self.assertRaises(RuntimeError): + with _native_section(): + c2pa_module._register_for_section_flush(FlushRaises()) + def test_context_sign_after_close_raises_rather_than_skipping_signer(self): """Signing through a closed Context must raise, not silently succeed.