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 9fe6b33406..bce51d8cf5 100644
--- a/.evergreen/generated_configs/tasks.yml
+++ b/.evergreen/generated_configs/tasks.yml
@@ -2895,6 +2895,34 @@ 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
+    exec_timeout_secs: 7200
+    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 72af2599f7..7152e715ee 100644
--- a/.evergreen/generated_configs/variants.yml
+++ b/.evergreen/generated_configs/variants.yml
@@ -489,6 +489,16 @@ buildvariants:
     expansions:
       SUB_TEST_NAME: pyopenssl
 
+  # 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 aaded3130c..cdb05b3d93 100644
--- a/.evergreen/scripts/generate_config.py
+++ b/.evergreen/scripts/generate_config.py
@@ -341,6 +341,13 @@ def create_mod_wsgi_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")
@@ -659,6 +666,35 @@ 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}"
+        # TSan builds a fully instrumented free-threaded CPython from source
+        # before it can run anything, which does not fit in the project-wide
+        # 60 minute exec timeout.
+        exec_timeout_secs = 7200 if sanitizer == "tsan" else None
+        tasks.append(
+            EvgTask(
+                name=name,
+                tags=tags,
+                exec_timeout_secs=exec_timeout_secs,
+                commands=[server_func, test_func],
+            )
+        )
+    return tasks
+
+
 def create_test_non_standard_tasks():
     """For variants that set a TEST_NAME."""
     tasks = []
@@ -1315,6 +1351,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..4b15b2a24b
--- /dev/null
+++ b/.evergreen/scripts/run-sanitizer-tests.sh
@@ -0,0 +1,180 @@
+#!/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_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
+    export CFLAGS="-fsanitize=address,undefined -fno-omit-frame-pointer -g -O0"
+    export LDFLAGS="-fsanitize=address,undefined"
+    RUNTIME_LIB=$("$CC" -print-file-name=libasan.so)
+    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
+
+    if [ ! -f "$RUNTIME_LIB" ]; then
+      echo "Could not locate the $SANITIZER runtime library (got: $RUNTIME_LIB). Is the matching sanitizer runtime package installed?" >&2
+      exit 1
+    fi
+
+    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)
+    ;;
+  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
+
+    # Building CPython from source needs its usual dev dependencies. Best
+    # effort: if apt-get isn't available or permitted, the OpenSSL header
+    # check below still catches the case that breaks the pip installs.
+    sudo apt-get update -qq || true
+    # Package list matches CPython's own Doc/using/unix.rst and
+    # Tools/scripts/posix-deps-apt.sh, so the optional extension modules
+    # (_zstd, _gdbm, _tkinter, ...) build instead of silently skipping.
+    sudo apt-get install -y --no-install-recommends \
+      build-essential libssl-dev zlib1g-dev libbz2-dev libffi-dev \
+      libreadline-dev libsqlite3-dev liblzma-dev pkg-config libb2-dev \
+      libgdbm-dev libgdbm-compat-dev libncurses5-dev libzstd-dev tk-dev \
+      uuid-dev curl || 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 after apt-get install. pip needs a working ssl module to reach PyPI. 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)
+    ;;
+  *)
+    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..e0d0bdca57
--- /dev/null
+++ b/.evergreen/tsan-suppressions.txt
@@ -0,0 +1,15 @@
+# PyMongo-specific ThreadSanitizer suppressions. Currently none are needed.
+#
+# .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 in PyMongo's own C extensions that has
+# been investigated and judged benign, with a comment saying why.
+#
+# Reference: https://github.com/google/sanitizers/wiki/ThreadSanitizerSuppressions
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/doc/changelog.rst b/doc/changelog.rst
index f90025963f..bf133c4c5e 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -50,6 +50,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/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py
index 537d9ea7f5..45c950404f 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 d56206b127..8a39ae3603 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 842c470b8f..25c332cbf4 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
 
@@ -104,6 +105,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_monitor.py b/test/test_monitor.py
index 3f24a5f2a2..bb00b7403c 100644
--- a/test/test_monitor.py
+++ b/test/test_monitor.py
@@ -20,6 +20,7 @@
 import gc
 import subprocess
 import sys
+import time
 import warnings
 from functools import partial
 
@@ -100,6 +101,53 @@ def test_cleanup_executors_on_client_close(self):
         for executor in executors:
             wait_until(lambda: executor._stopped, f"closed executor: {executor._name}", timeout=5)
 
+    @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))
+
+    @client_context.require_async
+    def test_close_stops_monitor_task(self):
+        """PYTHON-6048: the async counterpart of
+        test_close_stops_monitor_thread.
+        """
+        client = self.create_client()
+        server = next(iter(client._topology._servers.values()))
+        monitor = server._monitor
+
+        self.assertIsNotNone(monitor._executor._task)
+        self.assertFalse(monitor._executor._task.done())
+
+        client.close()
+
+        self.assertEqual(client._topology._monitor_tasks, [])
+        self.assertTrue(monitor._executor._task.done())
+        self.assertTrue(monitor._rtt_monitor._executor._task.done())
+
     @client_context.require_sync
     def test_no_thread_start_runtime_err_on_shutdown(self):
         """Test we silence noisy runtime errors fired when the MongoClient spawns a new thread
diff --git a/test/test_raw_bson_shared.py b/test/test_raw_bson_shared.py
index b7d54879aa..e5b23f75a3 100644
--- a/test/test_raw_bson_shared.py
+++ b/test/test_raw_bson_shared.py
@@ -17,6 +17,7 @@
 import gc
 import pickle
 import sys
+import threading
 import unittest
 import uuid
 
@@ -40,6 +41,20 @@
     b"\x00\x00\x00\x02street\x00\r\x00\x00\x00Baker Street\x00\x00\x00\x00"
 )
 
+N_THREADS = 16
+N_ITERS = 200
+BIG_PAYLOAD = "x" * 8000
+
+
+def _make_shared_doc_bytes() -> 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()