diff --git a/.evergreen/generated_configs/functions.yml b/.evergreen/generated_configs/functions.yml index 9fbc1f8500..92ac126633 100644 --- a/.evergreen/generated_configs/functions.yml +++ b/.evergreen/generated_configs/functions.yml @@ -85,6 +85,20 @@ functions: params: directory: src + # Run sanitizer tests + run sanitizer tests: + - command: subprocess.exec + params: + binary: bash + args: + - .evergreen/just.sh + - run-sanitizer-tests + working_dir: src + include_expansions_in_env: + - SANITIZER + - UV_PYTHON + type: test + # Run server run server: - command: subprocess.exec diff --git a/.evergreen/generated_configs/tasks.yml b/.evergreen/generated_configs/tasks.yml index e2677e7399..4b6ad1c815 100644 --- a/.evergreen/generated_configs/tasks.yml +++ b/.evergreen/generated_configs/tasks.yml @@ -2831,6 +2831,33 @@ tasks: - func: send dashboard data tags: [perf] + # Sanitizer tests + - name: test-sanitizer-asan + commands: + - func: run server + vars: + VERSION: latest + TOPOLOGY: standalone + AUTH: noauth + SSL: nossl + - func: run sanitizer tests + vars: + SANITIZER: asan + tags: [sanitizer, pr] + - name: test-sanitizer-tsan + commands: + - func: run server + vars: + VERSION: latest + TOPOLOGY: standalone + AUTH: noauth + SSL: nossl + - func: run sanitizer tests + vars: + SANITIZER: tsan + UV_PYTHON: 3.14t + tags: [sanitizer, pr, free-threaded] + # Search index tests - name: test-search-index-helpers commands: diff --git a/.evergreen/generated_configs/variants.yml b/.evergreen/generated_configs/variants.yml index 26027888fe..124a09eb4f 100644 --- a/.evergreen/generated_configs/variants.yml +++ b/.evergreen/generated_configs/variants.yml @@ -520,6 +520,16 @@ buildvariants: SUB_TEST_NAME: pyopenssl tags: [pr] + # Sanitizer tests + - name: sanitizers-ubuntu-22 + tasks: + - name: test-sanitizer-asan + - name: test-sanitizer-tsan + display_name: Sanitizers Ubuntu-22 + run_on: + - ubuntu2204-small + tags: [pr] + # Search index tests - name: search-index-helpers-rhel8 tasks: diff --git a/.evergreen/scripts/generate_config.py b/.evergreen/scripts/generate_config.py index 97fd9b0176..b2242e02da 100644 --- a/.evergreen/scripts/generate_config.py +++ b/.evergreen/scripts/generate_config.py @@ -400,6 +400,13 @@ def create_no_c_ext_variants(): return [create_variant(tasks, display_name, host=host, expansions=expansions)] +def create_sanitizer_variants(): + host = HOSTS["ubuntu22"] + tasks = ["test-sanitizer-asan", "test-sanitizer-tsan"] + display_name = get_variant_name("Sanitizers", host) + return [create_variant(tasks, display_name, host=host, tags=["pr"])] + + def create_disable_test_commands_variants(): host = DEFAULT_HOST expansions = dict(AUTH="auth", SSL="ssl", DISABLE_TEST_COMMANDS="1") @@ -718,6 +725,30 @@ def create_no_toolchain_tasks(): return tasks +def create_sanitizer_tasks(): + tasks = [] + # (sanitizer, free-threaded UV_PYTHON or None) + configs = [("asan", None), ("tsan", "3.14t")] + for sanitizer, python in configs: + tags = ["sanitizer", "pr"] + server_vars = dict(VERSION="latest", TOPOLOGY="standalone", AUTH="noauth", SSL="nossl") + server_func = FunctionCall(func="run server", vars=server_vars) + test_vars = dict(SANITIZER=sanitizer) + if python: + test_vars["UV_PYTHON"] = python + tags.append("free-threaded") + test_func = FunctionCall(func="run sanitizer tests", vars=test_vars) + name = f"test-sanitizer-{sanitizer}" + tasks.append( + EvgTask( + name=name, + tags=tags, + commands=[server_func, test_func], + ) + ) + return tasks + + def create_test_non_standard_tasks(): """For variants that set a TEST_NAME.""" tasks = [] @@ -1362,6 +1393,13 @@ def create_run_tests_func(): return "run tests", [setup_cmd, test_cmd] +def create_run_sanitizer_tests_func(): + includes = ["SANITIZER", "UV_PYTHON"] + args = [".evergreen/just.sh", "run-sanitizer-tests"] + sub_cmd = get_subprocess_exec(include_expansions_in_env=includes, args=args) + return "run sanitizer tests", [sub_cmd] + + def create_test_numpy_func(): includes = ["TOOLCHAIN_VERSION", "COVERAGE"] test_cmd = get_subprocess_exec( diff --git a/.evergreen/scripts/run-sanitizer-tests.sh b/.evergreen/scripts/run-sanitizer-tests.sh new file mode 100755 index 0000000000..7ebdcc2bcf --- /dev/null +++ b/.evergreen/scripts/run-sanitizer-tests.sh @@ -0,0 +1,197 @@ +#!/usr/bin/env bash +# Clean sanitizer rebuild of the C extensions, run the fixed BSON/client +# test files under the matching sanitizer runtime, and fail on any +# sanitizer diagnostic even if pytest itself exits 0. +# +# ASan runs against a prebuilt interpreter with libasan.so LD_PRELOADed. +# TSan builds a fully instrumented free-threaded CPython from source +# instead: TSan cannot see synchronization in code compiled without +# -fsanitize=thread, so LD_PRELOADing it onto a prebuilt interpreter +# reports false races inside CPython's own free-threading internals. +set -eu + +cd "$(dirname "${BASH_SOURCE[0]}")/../.." + +SANITIZER=${SANITIZER:?"SANITIZER must be set to 'asan' or 'tsan'"} +UV_PYTHON=${UV_PYTHON:-3.13} +TEST_FILES=(test/test_bson.py test/test_message.py test/test_raw_bson.py test/test_raw_bson_shared.py test/test_client.py) + +# The free-threaded CPython the TSan task builds. Kept in step with the +# "3.14t" entry in .evergreen/scripts/generate_config_utils.py's CPYTHONS. +CPYTHON_TAG=v3.14.0 +CPYTHON_SRC=.tsan-cpython-src +CPYTHON_INSTALL=.tsan-cpython-install + +# A stale build/ or venv can leave a .so or install linked against the wrong +# sanitizer's runtime without a build error, so always start clean. +rm -rf build +rm -f bson/*.so pymongo/*.so + +export CC=${CC:-clang} +export CXX=${CXX:-clang++} +export PYMONGO_C_EXT_MUST_BUILD=1 + +case "$SANITIZER" in + asan) + rm -rf .sanitizer-venv + # PYMONGO_TEST_ALLOC_FAILURE compiles the test-only fail-once realloc + # hook (bson/buffer.c) so the realloc-failure regression tests run. + export CFLAGS="-fsanitize=address,undefined -fno-omit-frame-pointer -g -O0 -DPYMONGO_TEST_ALLOC_FAILURE" + export LDFLAGS="-fsanitize=address,undefined" + + # clang's ASan runtime also provides the UBSan handlers, so preloading + # libasan alone resolves the symbols from both -fsanitize=address and + # -fsanitize=undefined. Preloading a separate libubsan_standalone aborts + # during init: the two runtimes both bring sanitizer_common and fight + # over sigaction. clang names the runtime libclang_rt.asan-x86_64.so, and + # -print-file-name can also return a bare name or a linker script, which + # ld.so refuses to preload, so resolve the candidates and require a real + # ELF shared object. + resolve_runtime() { + local name path + for name in "$@"; do + path=$("$CC" -print-file-name="$name") + if [ -f "$path" ] && [ "$(head -c 4 "$path" | od -An -tx1 | tr -d " \n")" = "7f454c46" ]; then + printf "%s\n" "$path" + return 0 + fi + done + return 1 + } + RUNTIME_LIB=$(resolve_runtime libclang_rt.asan-x86_64.so libasan.so) || { + echo "Could not locate a loadable ASan runtime through $CC -print-file-name. Is the matching sanitizer runtime package installed?" >&2 + exit 1 + } + export ASAN_OPTIONS="detect_leaks=0" + export UBSAN_OPTIONS="print_stacktrace=1:halt_on_error=1" + # Route CPython's own allocations through the system allocator so ASan's + # redzones can see them; pymalloc otherwise hides real bugs and adds + # noise. The free-threaded TSan interpreter doesn't accept this value, + # so it's scoped to ASan only. + export PYTHONMALLOC=malloc + + uv venv --python "$UV_PYTHON" .sanitizer-venv + VENV_PYTHON=.sanitizer-venv/bin/python3 + uv pip install --python "$VENV_PYTHON" -e . --reinstall + uv pip install --python "$VENV_PYTHON" -r requirements/test.txt + + PYTEST_CMD=(env "LD_PRELOAD=$RUNTIME_LIB" "$VENV_PYTHON" -m pytest) + + # The C-specific tests skip silently when the extensions are missing, + # which would turn a broken rebuild into a false-green task. Import the + # extension modules directly so a failure shows the real ImportError. + env "LD_PRELOAD=$RUNTIME_LIB" "$VENV_PYTHON" -c "import bson._cbson, pymongo._cmessage" + ;; + tsan) + rm -rf "$CPYTHON_SRC" "$CPYTHON_INSTALL" + + # TSan's shadow mapping can fail to reserve its address ranges under the + # default ASLR entropy on recent kernels, which crashes the process at + # startup. CPython's own CI lowers the entropy the same way. Evergreen + # hosts may not permit it, so this is best effort. + sudo sysctl -w vm.mmap_rnd_bits=28 || true + + # Fail here rather than after the 20-30 minute build: without these + # headers _ssl won't build and the pip installs below cannot reach PyPI. + if [ ! -f /usr/include/openssl/ssl.h ]; then + echo "OpenSSL development headers not found at /usr/include/openssl/ssl.h. This host has no way to install them: apt-get isn't available here, so building CPython with a working ssl module isn't possible. Aborting before the CPython build." >&2 + exit 1 + fi + + CPYTHON_INSTALL_ABS="$(pwd)/$CPYTHON_INSTALL" + git clone --depth 1 --branch "$CPYTHON_TAG" https://github.com/python/cpython.git "$CPYTHON_SRC" + + # TSan accepts one suppressions file, so concatenate the pinned tag's own + # suppressions with this repo's additions. The 3.14 branch's file is not + # empty: even a fully instrumented build needs its ~24 entries. + CPYTHON_SUPPRESSIONS="$CPYTHON_SRC/Tools/tsan/suppressions_free_threading.txt" + if [ ! -f "$CPYTHON_SUPPRESSIONS" ]; then + echo "Expected CPython's own TSan suppressions at $CPYTHON_SUPPRESSIONS, but the file is missing from the $CPYTHON_TAG source tree." >&2 + exit 1 + fi + COMBINED_SUPPRESSIONS="$(pwd)/.tsan-suppressions-combined.txt" + cat "$CPYTHON_SUPPRESSIONS" .evergreen/tsan-suppressions.txt > "$COMBINED_SUPPRESSIONS" + + # Flags mirror CPython's own TSan CI job (.github/workflows/reusable-san.yml). + # CFLAGS/LDFLAGS are deliberately left unset here: --with-thread-sanitizer + # and --with-pydebug already supply the right flags, and overriding them + # would fight configure. + # + # Unlike CPython's CI this does not rebuild OpenSSL with TSan, which is + # only needed to keep the ssl tests quiet. The tests below don't use ssl, + # but pip does need it to reach PyPI, so _ssl still has to build against + # the system OpenSSL. + ( + cd "$CPYTHON_SRC" + ./configure \ + --with-thread-sanitizer \ + --with-pydebug \ + --disable-gil \ + --prefix="$CPYTHON_INSTALL_ABS" + make -j"$(nproc 2>/dev/null || echo 4)" + make install + ) + + # --disable-gil adds a "t" suffix and --with-pydebug adds a "d", so this + # build installs as pythonX.Ytd. bin/python3 is the last-resort fallback. + TSAN_PYTHON="" + for candidate in "$CPYTHON_INSTALL"/bin/python3.*td "$CPYTHON_INSTALL"/bin/python3.*t "$CPYTHON_INSTALL"/bin/python3; do + if [ -x "$candidate" ]; then + TSAN_PYTHON="$candidate" + break + fi + done + if [ -z "$TSAN_PYTHON" ]; then + echo "Could not find the interpreter built from source under $CPYTHON_INSTALL/bin:" >&2 + ls -l "$CPYTHON_INSTALL/bin" >&2 || true + exit 1 + fi + echo "Using TSan-instrumented interpreter: $TSAN_PYTHON" + "$TSAN_PYTHON" -VV + "$TSAN_PYTHON" -c 'import sysconfig, sys; sys.exit(0 if sysconfig.get_config_var("Py_GIL_DISABLED") else "interpreter is not free-threaded")' + if ! "$TSAN_PYTHON" -c 'import ssl' >/dev/null 2>&1; then + echo "Warning: the interpreter built from source has no working ssl module, so pip cannot reach PyPI. Install the system OpenSSL development headers on this host." >&2 + fi + + # Build PyMongo's C extensions with the same instrumentation as the + # interpreter. These are plain shell env vars so pip's isolated build + # subprocess inherits them; build isolation is left on so pip resolves + # hatchling's build dependencies itself. + export CFLAGS="-fsanitize=thread -fno-omit-frame-pointer -g -O0" + export LDFLAGS="-fsanitize=thread" + "$TSAN_PYTHON" -m pip install -e . + "$TSAN_PYTHON" -m pip install -r requirements/test.txt + + # Set after the build: halt_on_error=1 would abort the CPython build and + # the installs on any diagnostic raised by those tools themselves. + # handle_segv=0 matches CPython's own TSan CI job. + TSAN_OPTIONS="halt_on_error=1:handle_segv=0:suppressions=$COMBINED_SUPPRESSIONS" + export TSAN_OPTIONS + + # No LD_PRELOAD: both the interpreter and the extensions link the TSan + # runtime at build time. + PYTEST_CMD=("$TSAN_PYTHON" -m pytest) + + # The C-specific tests skip silently when the extensions are missing, + # which would turn a broken rebuild into a false-green task. Import the + # extension modules directly so a failure shows the real ImportError. + "$TSAN_PYTHON" -c "import bson._cbson, pymongo._cmessage" + ;; + *) + echo "Unknown SANITIZER: $SANITIZER (expected 'asan' or 'tsan')" >&2 + exit 1 + ;; +esac + +LOG_FILE=$(mktemp) +set +e +"${PYTEST_CMD[@]}" -v --capture=no "${TEST_FILES[@]}" 2>&1 | tee "$LOG_FILE" +PYTEST_STATUS=${PIPESTATUS[0]} +set -e + +if grep -qE "ERROR: (AddressSanitizer|LeakSanitizer)|runtime error:|WARNING: ThreadSanitizer|SUMMARY: (Address|Undefined|ThreadSanitizer)" "$LOG_FILE"; then + echo "Sanitizer diagnostic detected in test output, failing task" >&2 + exit 1 +fi + +exit "$PYTEST_STATUS" diff --git a/.evergreen/tsan-suppressions.txt b/.evergreen/tsan-suppressions.txt new file mode 100644 index 0000000000..3de4172915 --- /dev/null +++ b/.evergreen/tsan-suppressions.txt @@ -0,0 +1,24 @@ +# PyMongo-specific ThreadSanitizer suppressions. +# +# .evergreen/scripts/run-sanitizer-tests.sh concatenates this file with the +# pinned CPython tag's own Tools/tsan/suppressions_free_threading.txt and +# passes the result as TSan's single suppressions= path. That upstream file +# is not empty on the 3.14 branch: it carries roughly two dozen entries +# (assign_version_tag, update_one_slot, _PyFrame_GetCode, rangeiter_next, +# list_ass_slice_lock_held, PyObject_Realloc, mi_block_set_nextx, +# pthread_create, and others) that CPython's own fully instrumented TSan CI +# job needs to run clean. Do not duplicate those here. +# +# Add an entry below only for a race that has been investigated and judged +# benign, with a comment saying why: either a race in PyMongo's own C +# extensions, or a CPython-internal race (both stacks within CPython) that +# upstream's own suppressions do not yet cover. +# +# Reference: https://github.com/google/sanitizers/wiki/ThreadSanitizerSuppressions + +# CPython-internal: under free-threading the GC's delete_garbage path +# (memory_clear) and a worker thread's deferred refcount merge +# (memory_dealloc) can both enter _memory_release for the same +# memoryview. Both stacks are entirely within CPython; pymongo only +# creates and releases the memoryviews concurrently. +race_top:_memory_release diff --git a/.gitignore b/.gitignore index 6c4a512018..d637851c9f 100644 --- a/.gitignore +++ b/.gitignore @@ -46,3 +46,9 @@ xunit-results/ coverage.xml server.log .coverage + +# sanitizer test task scratch (see .evergreen/scripts/run-sanitizer-tests.sh) +.sanitizer-venv/ +.tsan-cpython-src/ +.tsan-cpython-install/ +.tsan-suppressions-combined.txt diff --git a/bson/_cbsonmodule.c b/bson/_cbsonmodule.c index d383cee308..8177a0dc68 100644 --- a/bson/_cbsonmodule.c +++ b/bson/_cbsonmodule.c @@ -161,6 +161,15 @@ extern int cbson_long_long_to_str(long long num, char* str, size_t size) { return 0; } +static PyObject* _test_fail_next_realloc(PyObject* self, PyObject* Py_UNUSED(ignored)) { +#ifdef PYMONGO_TEST_ALLOC_FAILURE + pymongo_buffer_test_fail_next_realloc(); + Py_RETURN_TRUE; +#else + Py_RETURN_FALSE; +#endif +} + static PyObject* _test_long_long_to_str(PyObject* self, PyObject* args) { // Test extreme values Py_ssize_t maxNum = PY_SSIZE_T_MAX; @@ -3516,6 +3525,7 @@ static PyMethodDef _CBSONMethods[] = { "Decode all elements of a raw BSON document into a result mapping."}, {"_array_of_documents_to_buffer", _cbson_array_of_documents_to_buffer, METH_VARARGS, "Convert raw array of documents to a stream of BSON documents"}, {"_test_long_long_to_str", _test_long_long_to_str, METH_VARARGS, "Test conversion of extreme and common Py_ssize_t values to str."}, + {"_test_fail_next_realloc", _test_fail_next_realloc, METH_NOARGS, "Test-only hook: fail the next buffer growth exactly once. Returns False if the hook is not compiled in."}, {NULL, NULL, 0, NULL} }; diff --git a/bson/buffer.c b/bson/buffer.c index cc75202746..48e4a2cc21 100644 --- a/bson/buffer.c +++ b/bson/buffer.c @@ -25,6 +25,17 @@ #define INITIAL_BUFFER_SIZE 256 +#ifdef PYMONGO_TEST_ALLOC_FAILURE +/* Test-only hook: when enabled, the next buffer_grow call fails exactly + * once, deterministically reproducing a realloc failure (including the + * free of the old block, the NULL buffer pointer, and the stale size). */ +static int _test_fail_next_realloc = 0; + +void pymongo_buffer_test_fail_next_realloc(void) { + _test_fail_next_realloc = 1; +} +#endif + struct buffer { char* buffer; int size; @@ -91,6 +102,16 @@ static int buffer_grow(buffer_t buffer, int min_length) { size = min_length; } } +#ifdef PYMONGO_TEST_ALLOC_FAILURE + if (_test_fail_next_realloc) { + _test_fail_next_realloc = 0; + /* Replicate the realloc failure path exactly. */ + free(old_buffer); + buffer->buffer = NULL; + set_memory_error(); + return 1; + } +#endif buffer->buffer = (char*)realloc(buffer->buffer, sizeof(char) * size); if (buffer->buffer == NULL) { free(old_buffer); diff --git a/bson/buffer.h b/bson/buffer.h index a78e34e4de..2535aad076 100644 --- a/bson/buffer.h +++ b/bson/buffer.h @@ -48,4 +48,9 @@ buffer_position pymongo_buffer_get_position(buffer_t buffer); char* pymongo_buffer_get_buffer(buffer_t buffer); void pymongo_buffer_update_position(buffer_t buffer, buffer_position new_position); +/* Test-only: enable a hook that makes the next buffer growth fail exactly + * once. Only functional when the extension is built with + * PYMONGO_TEST_ALLOC_FAILURE. */ +void pymongo_buffer_test_fail_next_realloc(void); + #endif diff --git a/doc/changelog.rst b/doc/changelog.rst index 5fb2fb4731..394a87fbf0 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -90,6 +90,11 @@ PyMongo 4.18 brings a number of changes including: - Fixed a bug on Windows, and on macOS when using PyOpenSSL, where ``SSL_CERT_FILE``/``SSL_CERT_DIR`` were merged with, rather than replacing, the OS/certifi certificate store. +- Fixed a race between a server monitor's background thread and connection + pool teardown during + :meth:`~pymongo.synchronous.mongo_client.MongoClient.close`. Closing a + client now waits (with a bounded timeout) for its monitor threads to stop + before returning, adding up to a couple hundred milliseconds to ``close()``. - Added general availability support for Queryable Encryption prefix, suffix, and substring queries against MongoDB 9.0+. These queries require libmongocrypt 1.20.0 or later: diff --git a/justfile b/justfile index df678cdbac..d9eabc0b83 100644 --- a/justfile +++ b/justfile @@ -129,6 +129,9 @@ coverage-xml: run-server *args="": bash .evergreen/scripts/run-server.sh {{args}} +run-sanitizer-tests *args="": + bash .evergreen/scripts/run-sanitizer-tests.sh {{args}} + [group('server')] stop-server: bash .evergreen/scripts/stop-server.sh diff --git a/pymongo/_cmessagemodule.c b/pymongo/_cmessagemodule.c index 1597652d69..0d04c65d0d 100644 --- a/pymongo/_cmessagemodule.c +++ b/pymongo/_cmessagemodule.c @@ -922,6 +922,15 @@ _cbson_encode_batched_write_command(PyObject* self, PyObject* args) { return result; } +static PyObject* _test_fail_next_realloc(PyObject* self, PyObject* Py_UNUSED(ignored)) { +#ifdef PYMONGO_TEST_ALLOC_FAILURE + pymongo_buffer_test_fail_next_realloc(); + Py_RETURN_TRUE; +#else + Py_RETURN_FALSE; +#endif +} + static PyMethodDef _CMessageMethods[] = { {"_query_message", _cbson_query_message, METH_VARARGS, "create a query message to be sent to MongoDB"}, @@ -935,6 +944,8 @@ static PyMethodDef _CMessageMethods[] = { "Create the next batched insert, update, or delete using OP_MSG"}, {"_encode_batched_op_msg", _cbson_encode_batched_op_msg, METH_VARARGS, "Encode the next batched insert, update, or delete using OP_MSG"}, + {"_test_fail_next_realloc", _test_fail_next_realloc, METH_NOARGS, + "Test-only hook: fail the next buffer growth exactly once. Returns False if the hook is not compiled in."}, {NULL, NULL, 0, NULL} }; diff --git a/pymongo/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py index ba0e3804a6..c4cd4107ef 100644 --- a/pymongo/asynchronous/monitor.py +++ b/pymongo/asynchronous/monitor.py @@ -16,7 +16,6 @@ from __future__ import annotations -import asyncio import atexit import time import weakref @@ -106,9 +105,9 @@ async def close(self) -> None: """ self.gc_safe_close() - async def join(self) -> None: + async def join(self, timeout: Optional[int] = None) -> None: """Wait for the monitor to stop.""" - await self._executor.join() + await self._executor.join(timeout) def request_check(self) -> None: """If the monitor is sleeping, wake it soon.""" @@ -184,10 +183,9 @@ def gc_safe_close(self) -> None: self._rtt_monitor.gc_safe_close() self.cancel_check() - async def join(self) -> None: - await asyncio.gather( - self._executor.join(), self._rtt_monitor.join(), return_exceptions=True - ) # type: ignore[func-returns-value] + async def join(self, timeout: Optional[int] = None) -> None: + await self._executor.join(timeout) + await self._rtt_monitor.join(timeout) async def close(self) -> None: self.gc_safe_close() diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index c4f1a3fa96..b390e0fb94 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -16,7 +16,6 @@ from __future__ import annotations -import asyncio import os import queue import random @@ -241,8 +240,9 @@ async def select_servers( else: server_timeout = server_selection_timeout - # Cleanup any completed monitor tasks safely - if not _IS_SYNC and self._monitor_tasks: + # Join any monitors closed under the topology lock. Doing it here, + # outside the lock, keeps the join off the paths that hold it. + if self._monitor_tasks: await self.cleanup_monitors() async with self._lock: @@ -454,8 +454,7 @@ async def _process_change( and self._description.topology_type not in SRV_POLLING_TOPOLOGIES ): await self._srv_monitor.close() - if not _IS_SYNC: - self._monitor_tasks.append(self._srv_monitor) + self._monitor_tasks.append(self._srv_monitor) # Wake anything waiting in select_servers(). self._condition.notify_all() @@ -614,8 +613,7 @@ async def close(self) -> None: old_td = self._description for server in self._servers.values(): await server.close() - if not _IS_SYNC: - self._monitor_tasks.append(server._monitor) + self._monitor_tasks.append(server._monitor) # Mark all servers Unknown. self._description = self._description.reset() @@ -626,12 +624,14 @@ async def close(self) -> None: # Stop SRV polling thread. if self._srv_monitor: await self._srv_monitor.close() - if not _IS_SYNC: - self._monitor_tasks.append(self._srv_monitor) + self._monitor_tasks.append(self._srv_monitor) self._opened = False self._closed = True + # Join the monitors we just closed, now that the lock is released. + await self.cleanup_monitors() + # Publish only after releasing the lock. if self._sdam._publish_tp: self._description = TopologyDescription( @@ -847,8 +847,7 @@ async def _update_servers(self) -> None: for address, server in list(self._servers.items()): if not self._description.has_server(address): await server.close() - if not _IS_SYNC: - self._monitor_tasks.append(server._monitor) + self._monitor_tasks.append(server._monitor) self._servers.pop(address) def _create_pool_for_server(self, address: _Address) -> Pool: @@ -934,13 +933,25 @@ def _error_message(self, selector: Callable[[Selection], Selection]) -> str: return ",".join(str(server.error) for server in servers if server.error) async def cleanup_monitors(self) -> None: + """Join monitors closed earlier while the topology lock was held. + + Always call this with the lock released: a monitor being joined may + itself be blocked acquiring the lock. Per-monitor join timeout: up to + 1s for RttMonitor/SrvMonitor, up to 2s for Monitor (sequential joins + of executor and rtt_monitor). Called from select_servers() and close(). + """ tasks = [] try: while self._monitor_tasks: tasks.append(self._monitor_tasks.pop()) except IndexError: pass - await asyncio.gather(*[t.join() for t in tasks], return_exceptions=True) # type: ignore[func-returns-value] + for t in tasks: + try: + await t.join(1) + except Exception: # noqa: S110 + # One monitor failing to stop must not block the rest. + pass def __repr__(self) -> str: msg = "" diff --git a/pymongo/periodic_executor.py b/pymongo/periodic_executor.py index 4b979ca9f9..f9f6463257 100644 --- a/pymongo/periodic_executor.py +++ b/pymongo/periodic_executor.py @@ -211,7 +211,11 @@ def join(self, timeout: Optional[int] = None) -> None: try: self._thread.join(timeout) except (ReferenceError, RuntimeError): - # Thread already terminated, or not yet started. + # Thread already terminated, or not yet started. This also + # covers a thread joining itself (RuntimeError: cannot join + # current thread), which happens when close() is called from + # within the monitor's own thread; that's safe to ignore + # since the thread is about to exit anyway. pass def wake(self) -> None: diff --git a/pymongo/synchronous/monitor.py b/pymongo/synchronous/monitor.py index 9a25757f03..2ccb9acb53 100644 --- a/pymongo/synchronous/monitor.py +++ b/pymongo/synchronous/monitor.py @@ -16,7 +16,6 @@ from __future__ import annotations -import asyncio import atexit import time import weakref @@ -106,9 +105,9 @@ def close(self) -> None: """ self.gc_safe_close() - def join(self) -> None: + def join(self, timeout: Optional[int] = None) -> None: """Wait for the monitor to stop.""" - self._executor.join() + self._executor.join(timeout) def request_check(self) -> None: """If the monitor is sleeping, wake it soon.""" @@ -184,8 +183,9 @@ def gc_safe_close(self) -> None: self._rtt_monitor.gc_safe_close() self.cancel_check() - def join(self) -> None: - asyncio.gather(self._executor.join(), self._rtt_monitor.join(), return_exceptions=True) # type: ignore[func-returns-value] + def join(self, timeout: Optional[int] = None) -> None: + self._executor.join(timeout) + self._rtt_monitor.join(timeout) def close(self) -> None: self.gc_safe_close() diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index c6468c8912..4ab845aee4 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -16,7 +16,6 @@ from __future__ import annotations -import asyncio import os import queue import random @@ -241,8 +240,9 @@ def select_servers( else: server_timeout = server_selection_timeout - # Cleanup any completed monitor tasks safely - if not _IS_SYNC and self._monitor_tasks: + # Join any monitors closed under the topology lock. Doing it here, + # outside the lock, keeps the join off the paths that hold it. + if self._monitor_tasks: self.cleanup_monitors() with self._lock: @@ -454,8 +454,7 @@ def _process_change( and self._description.topology_type not in SRV_POLLING_TOPOLOGIES ): self._srv_monitor.close() - if not _IS_SYNC: - self._monitor_tasks.append(self._srv_monitor) + self._monitor_tasks.append(self._srv_monitor) # Wake anything waiting in select_servers(). self._condition.notify_all() @@ -612,8 +611,7 @@ def close(self) -> None: old_td = self._description for server in self._servers.values(): server.close() - if not _IS_SYNC: - self._monitor_tasks.append(server._monitor) + self._monitor_tasks.append(server._monitor) # Mark all servers Unknown. self._description = self._description.reset() @@ -624,12 +622,14 @@ def close(self) -> None: # Stop SRV polling thread. if self._srv_monitor: self._srv_monitor.close() - if not _IS_SYNC: - self._monitor_tasks.append(self._srv_monitor) + self._monitor_tasks.append(self._srv_monitor) self._opened = False self._closed = True + # Join the monitors we just closed, now that the lock is released. + self.cleanup_monitors() + # Publish only after releasing the lock. if self._sdam._publish_tp: self._description = TopologyDescription( @@ -845,8 +845,7 @@ def _update_servers(self) -> None: for address, server in list(self._servers.items()): if not self._description.has_server(address): server.close() - if not _IS_SYNC: - self._monitor_tasks.append(server._monitor) + self._monitor_tasks.append(server._monitor) self._servers.pop(address) def _create_pool_for_server(self, address: _Address) -> Pool: @@ -932,13 +931,25 @@ def _error_message(self, selector: Callable[[Selection], Selection]) -> str: return ",".join(str(server.error) for server in servers if server.error) def cleanup_monitors(self) -> None: + """Join monitors closed earlier while the topology lock was held. + + Always call this with the lock released: a monitor being joined may + itself be blocked acquiring the lock. Per-monitor join timeout: up to + 1s for RttMonitor/SrvMonitor, up to 2s for Monitor (sequential joins + of executor and rtt_monitor). Called from select_servers() and close(). + """ tasks = [] try: while self._monitor_tasks: tasks.append(self._monitor_tasks.pop()) except IndexError: pass - asyncio.gather(*[t.join() for t in tasks], return_exceptions=True) # type: ignore[func-returns-value] + for t in tasks: + try: + t.join(1) + except Exception: # noqa: S110 + # One monitor failing to stop must not block the rest. + pass def __repr__(self) -> str: msg = "" diff --git a/test/asynchronous/test_monitor.py b/test/asynchronous/test_monitor.py index ed4a025580..dcb2709198 100644 --- a/test/asynchronous/test_monitor.py +++ b/test/asynchronous/test_monitor.py @@ -20,6 +20,7 @@ import gc import subprocess import sys +import time import warnings from functools import partial @@ -108,6 +109,53 @@ async def test_cleanup_executors_on_client_close(self): lambda: executor._stopped, f"closed executor: {executor._name}", timeout=5 ) + @async_client_context.require_sync + def test_close_stops_monitor_thread(self): + """PYTHON-6048: close() must join the monitor thread before + returning, so callers don't race with it while it may still be + reading from a socket that close() is about to tear down. + """ + client = self.create_client() + server = next(iter(client._topology._servers.values())) + monitor = server._monitor + + def thread_alive(executor): + try: + return executor._thread is not None and executor._thread.is_alive() + except ReferenceError: + return False + + deadline = time.monotonic() + 10 + while not thread_alive(monitor._executor) and time.monotonic() < deadline: + time.sleep(0.1) + self.assertTrue(thread_alive(monitor._executor), "monitor thread never started") + + client.close() + + # close() drains the deferred-join queue after releasing the + # topology lock, so every monitor it closed has been joined. + self.assertEqual(client._topology._monitor_tasks, []) + self.assertFalse(thread_alive(monitor._executor)) + self.assertFalse(thread_alive(monitor._rtt_monitor._executor)) + + @async_client_context.require_async + async def test_close_stops_monitor_task(self): + """PYTHON-6048: the async counterpart of + test_close_stops_monitor_thread. + """ + client = await self.create_client() + server = next(iter(client._topology._servers.values())) + monitor = server._monitor + + self.assertIsNotNone(monitor._executor._task) + self.assertFalse(monitor._executor._task.done()) + + await client.close() + + self.assertEqual(client._topology._monitor_tasks, []) + self.assertTrue(monitor._executor._task.done()) + self.assertTrue(monitor._rtt_monitor._executor._task.done()) + @async_client_context.require_sync def test_no_thread_start_runtime_err_on_shutdown(self): """Test we silence noisy runtime errors fired when the AsyncMongoClient spawns a new thread diff --git a/test/test_bson.py b/test/test_bson.py index 7ce21dd17f..b4c1a37936 100644 --- a/test/test_bson.py +++ b/test/test_bson.py @@ -61,7 +61,8 @@ from bson.codec_options import CodecOptions, DatetimeConversion from bson.datetime_ms import _DATETIME_ERROR_SUGGESTION from bson.dbref import DBRef -from bson.errors import InvalidBSON, InvalidDocument +from bson.decimal128 import Decimal128 +from bson.errors import BSONError, InvalidBSON, InvalidDocument from bson.int64 import Int64 from bson.max_key import MaxKey from bson.min_key import MinKey @@ -1842,5 +1843,175 @@ def test_long_long_to_string(self): print("_cbson was not imported. Check compilation logs.") +def _nested_document(depth): + doc = {"leaf": 1} + for _ in range(depth): + doc = {"d": doc} + return doc + + +def _typed_documents(): + """One document per BSON type, at value extremes where applicable.""" + return [ + {}, + {"int32": 1}, + {"int64 max": Int64(2**63 - 1), "int64 min": Int64(-(2**63))}, + {"double": 1.5, "double min": sys.float_info.min, "double max": sys.float_info.max}, + {"string": "hello é", "string empty": ""}, + {"string big": "x" * 70000}, + {"binary": Binary(b"\x00\xffbin", 0), "binary old": Binary(b"old", 2), "binary empty": Binary(b"", 0)}, + { + "datetime zero": DatetimeMS(0), + "datetime min": DatetimeMS(-(2**63)), + "datetime max": DatetimeMS(2**63 - 1), + }, + {"timestamp": Timestamp(0, 1), "timestamp max": Timestamp(0xFFFFFFFF, 0xFFFFFFFF)}, + { + "decimal128": Decimal128("1.25"), + "decimal128 nan": Decimal128("NaN"), + "decimal128 inf": Decimal128("Infinity"), + "decimal128 neg0": Decimal128("-0"), + }, + {"regex": Regex("a*b", "im"), "regex empty": Regex("", "")}, + {"code": Code("function() { return 1; }"), "code scope": Code("f", {"x": [1, 2]})}, + { + "dbref": DBRef("coll", "id"), + "objectid": ObjectId(b"0123456789ab"), + "minkey": MinKey(), + "maxkey": MaxKey(), + "null": None, + "bool": True, + }, + {"array": [1, "two", None, {"nested": True}], "subdoc": {"b": {"c": 1}}}, + {"bytes": b"raw bytes"}, + {"key" * 63 + "e": 1}, + {"deep": _nested_document(50)}, + ] + + +_DATETIME_MS_CODEC = CodecOptions(datetime_conversion=DatetimeConversion.DATETIME_MS) + + +class TestBSONEncodeDecodeBoundaries(unittest.TestCase): + """Encode every BSON type at value extremes and re-encode byte-stably. + + Drives buffer_assure_space through its doubling growth transitions and + every _downcast_and_check path under sanitizer builds.""" + + def _round_trip(self, doc): + encoded = bson.encode(doc, codec_options=_DATETIME_MS_CODEC) + decoded = bson.decode(encoded, codec_options=_DATETIME_MS_CODEC) + # Compare byte-level stability rather than decoded == doc so that + # NaN payloads (which are not equal to themselves) round-trip too. + self.assertEqual( + bson.encode(decoded, codec_options=_DATETIME_MS_CODEC), encoded + ) + + def test_round_trip_value_extremes(self): + for doc in _typed_documents(): + self._round_trip(doc) + + def test_round_trip_size_boundaries(self): + # String sizes that cross buffer doubling boundaries. + for size in (127, 128, 255, 256, 2**16, 2**16 + 1): + self._round_trip({"a": "x" * size}) + + def test_encode_realloc_failure_is_safe(self): + try: + from bson import _cbson + except ImportError: + self.skipTest("_cbson not built") + if not _cbson._test_fail_next_realloc(): + self.skipTest("realloc failure hook not compiled in") + with self.assertRaises(MemoryError): + bson.encode({"a": "x" * 1000}) + + +class TestBSONHostileLengths(unittest.TestCase): + """Decode must reject malformed length fields with a BSONError. + + Exercises the int-boundary validation in the C decoder without + allocating multi-GiB buffers.""" + + def _assert_bson_error(self, data, msg=None): + with self.assertRaises(BSONError, msg=msg): + bson.decode(data) + + def test_declared_document_lengths(self): + base = bson.encode({"a": "b" * 20, "c": [1, 2]}) + for length in (0, -1, 0x7F7F7F7F, 0x7FFFFFFF, len(base) + 1): + data = bytearray(base) + data[0:4] = struct.pack(" bytes: + return encode( + { + "small": {"n": 1}, + "big": {"payload": BIG_PAYLOAD, "n": 42}, + "arr": [{"payload": BIG_PAYLOAD, "idx": i} for i in range(3)], + } + ) + class _TaggedRawBSONDocument(RawBSONDocument): """RawBSONDocument subclass with a different __init__ signature and @@ -321,5 +336,142 @@ def test_contains_dbref(self): self.assertEqual(doc["value"].raw, raw_encoded) +class TestRawBSONDocumentConcurrency(unittest.TestCase): + """Concurrency and buffer-lifetime regression tests for zero-copy + RawBSONDocument, intended to also run under ASan/UBSan/TSan in CI + (see .evergreen/scripts/run-sanitizer-tests.sh).""" + + def test_concurrent_reads_on_shared_document(self): + doc_bytes = _make_shared_doc_bytes() + shared_doc = RawBSONDocument(doc_bytes) + + # Confirm we're actually exercising the zero-copy memoryview path, + # not accidentally testing a byte-copy fallback. + big_check = shared_doc["big"] + self.assertIsInstance(big_check.raw, memoryview) + self.assertTrue(big_check.raw.readonly) + + errors: list[BaseException] = [] + errors_lock = threading.Lock() + + def worker() -> None: + try: + for _ in range(N_ITERS): + big = shared_doc["big"] + assert big["payload"] == BIG_PAYLOAD + assert big["n"] == 42 + assert isinstance(big.raw, memoryview) + + arr = shared_doc["arr"] + for j, item in enumerate(arr): + assert item["idx"] == j + assert item["payload"] == BIG_PAYLOAD + + small = shared_doc["small"] + assert small["n"] == 1 + + # Re-encodes a view-backed subdocument while other + # threads may be reading the same underlying buffer. + reencoded = encode({"again": big}) + assert isinstance(reencoded, bytes) + + multi = decode_all(doc_bytes * 2, DEFAULT_RAW_BSON_OPTIONS) + assert len(multi) == 2 + for d in multi: + assert isinstance(d["big"].raw, memoryview) + except BaseException as exc: + with errors_lock: + errors.append(exc) + raise + + threads = [threading.Thread(target=worker) for _ in range(N_THREADS)] + for t in threads: + t.start() + for t in threads: + t.join() + + self.assertEqual(errors, []) + + def test_buffer_survives_after_source_bytes_are_dropped(self): + # Buffer-lifetime edge case: slice a bytearray, wrap it in a + # RawBSONDocument, then mutate/drop the original backing buffer to + # make sure the document doesn't hold a dangling view. + encoded = _make_shared_doc_bytes() + for _ in range(50): + buf = bytearray(encoded) + b"\x00" * 10 + view = memoryview(buf)[: len(encoded)] + raw = RawBSONDocument(view) + _ = raw["big"] + _ = dict(raw["small"]) + _ = list(raw["arr"]) + copied = dict(raw) + del raw + del view + buf[:] = b"\xff" * len(buf) + del buf + gc.collect() + self.assertEqual(copied["small"]["n"], 1) + + +class _FakeBulkWriteContext: + """Stand-in for pymongo.message._BulkWriteContext. + + The C batched-message builders only read four numeric attributes off + the context object, so a plain object exposing those is enough to + drive them without a live server connection. + """ + + max_bson_size = 16 * 1024 * 1024 + max_write_batch_size = 100000 + max_message_size = 48 * 1024 * 1024 + max_split_size = 16 * 1024 * 1024 + + +class TestBatchedMessageBuilderConcurrency(unittest.TestCase): + def test_concurrent_batched_message_building(self): + try: + from pymongo import _cmessage + except ImportError: + self.skipTest("pymongo._cmessage C extension is not built") + + has_op_msg = hasattr(_cmessage, "_encode_batched_op_msg") + has_write_cmd = hasattr(_cmessage, "_encode_batched_write_command") + if not (has_op_msg or has_write_cmd): + self.skipTest("no batched message builder found on pymongo._cmessage") + + ns = "db.coll" + docs = [{"_id": i, "payload": "y" * 100} for i in range(50)] + command = {"insert": "coll", "ordered": True} + ctx = _FakeBulkWriteContext() + insert_op = 0 # pymongo.message._INSERT + + errors: list[BaseException] = [] + errors_lock = threading.Lock() + + def worker() -> None: + try: + for _ in range(N_ITERS): + if has_op_msg: + _cmessage._encode_batched_op_msg( + insert_op, command, docs, True, DEFAULT_RAW_BSON_OPTIONS, ctx + ) + if has_write_cmd: + _cmessage._encode_batched_write_command( + ns, insert_op, command, docs, DEFAULT_RAW_BSON_OPTIONS, ctx + ) + except BaseException as exc: + with errors_lock: + errors.append(exc) + raise + + threads = [threading.Thread(target=worker) for _ in range(N_THREADS)] + for t in threads: + t.start() + for t in threads: + t.join() + + self.assertEqual(errors, []) + + if __name__ == "__main__": unittest.main()