Closes #23047: Enable aborting DB queries for abandoned HTTP requests

contrib/uwsgi.ini

@@ -20,6 +20,9 @@ need-app = true
 ; do not use multiple interpreters
 single-interpreter = true
 
+; initialize Python threading support, which ABORT_ON_CLIENT_DISCONNECT requires
+enable-threads = true
+
 ; change to the project directory
 chdir = netbox
 

docs/configuration/system.md

@@ -1,5 +1,46 @@
 # System Parameters
 
+## ABORT_ON_CLIENT_DISCONNECT
+
+!!! info "New in NetBox v4.8"
+
+Default: `False`
+
+When enabled, NetBox watches the client connection for the duration of each read-only request. If the client disconnects before the response has been sent, the database queries that request has in flight are cancelled, returning the worker and its database connections to their pools at the moment the client gives up rather than when the request would otherwise have completed.
+
+This is intended for deployments serving automated API consumers which set aggressive client-side timeouts and retry on failure. WSGI provides no cancellation mechanism, so a worker normally runs an abandoned request to completion and discovers the disconnect only when it attempts to write the response. Each retry therefore adds another orphaned request on top of the one already running, and a single impatient client can saturate every worker and every database connection. Aborting on disconnect caps this at roughly one orphaned request per client.
+
+Every database connection opened by the request is cancelled, not only the connection to the default database. Any open transaction is rolled back, and the affected connections are closed, ending their database sessions, rather than being reused. This applies equally to persistent connections (`CONN_MAX_AGE`) and to connection pooling, where the pool replaces each closed connection with a fresh one.
+
+Aborted requests are recorded in the `netbox.disconnect` log at the `INFO` level, with the request ID, method, path, elapsed time, and client IP address. When [`METRICS_ENABLED`](./miscellaneous.md#metrics_enabled) is also set, they are counted by the `netbox_client_disconnects_total` metric, labelled by method and view.
+
+!!! warning "Limitations"
+    * Locating the client connection is WSGI server-specific, and only gunicorn and uWSGI are supported. Under the development server (`runserver`) or any other WSGI server, this parameter has no effect: NetBox logs an informational message once per worker process and serves requests exactly as it would with the parameter disabled.
+    * Under uWSGI, the watchdog runs in a Python thread and so requires uWSGI's `enable-threads` option. The configuration shipped in `contrib/uwsgi.ini` sets it, but a copy of that file made from an earlier NetBox release will not, and the watchdog will not run.
+    * Only requests using the `GET`, `HEAD`, and `OPTIONS` methods are watched. Cancelling a `POST`, `PUT`, `PATCH`, or `DELETE` which the client had abandoned would silently roll back a write that client may believe has landed, so requests using unsafe methods always run to completion.
+    * What is observed is the connection from the reverse proxy to NetBox, not the client's own connection. Whether a client abort reaches NetBox is therefore the proxy's decision: nginx propagates it by default, but `proxy_ignore_client_abort on` (or `uwsgi_ignore_client_abort on`) suppresses it.
+    * Cancellation applies to database queries, not to the request as a whole. A request which has finished querying and is busy rendering a template or serializing a response is not interrupted, and neither are streaming exports (see [`STREAMING_EXPORTS`](./miscellaneous.md#streaming_exports)), whose response begins before their content is generated.
+    * A client which half-closes its connection (shutting down its sending side after transmitting the request, while still waiting to read the response) is indistinguishable from one which has disconnected, and its request is aborted. Most HTTP clients do not do this, but some tools (such as `nc -N`) and health checkers do.
+    * GraphQL queries are normally sent using `POST`, and so are not watched. Only GraphQL queries sent using `GET` are protected, and for those the GraphQL layer reports the resulting error in its own response body rather than as an HTTP 499.
+    * NetBox returns a synthetic `499 Client Closed Request` for an aborted request. This response is never delivered to anyone, as the client has already disconnected; it exists to keep the cancellation out of NetBox's exception handling, so that error-reporting integrations are not flooded with self-inflicted HTTP 500 reports.
+
+A related but distinct control is PostgreSQL's `statement_timeout`, which can be applied to NetBox's database connections via [`DATABASES`](./required-parameters.md#databases):
+
+```python
+DATABASES = {
+    'default': {
+        # ...
+        'OPTIONS': {
+            'options': '-c statement_timeout=30000',  # 30 seconds
+        },
+    }
+}
+```
+
+This imposes a fixed ceiling on every query rather than cancelling only those which nobody is waiting for: it cannot distinguish a legitimately slow query from an orphaned one, and it does not shorten the window during which a worker is occupied by an abandoned request. The two are complementary, and a `statement_timeout` remains a sensible backstop whether or not this parameter is enabled.
+
+---
+
 ## BASE_PATH
 
 Default: `None`

docs/installation/4b-uwsgi.md

@@ -28,6 +28,9 @@ NetBox ships with a default configuration file for uWSGI. To use it, copy `/opt/
 sudo cp /opt/netbox/contrib/uwsgi.ini /opt/netbox/uwsgi.ini
 ```
 
+!!! note "Upgrading an existing installation"
+    The provided configuration enables uWSGI's `enable-threads` option, which is required by the [`ABORT_ON_CLIENT_DISCONNECT`](../configuration/system.md#abort_on_client_disconnect) configuration parameter. Because this file is copied rather than referenced, a copy made from an earlier release of NetBox will not have it, and must be edited to add it before that parameter can take effect.
+
 While the provided configuration should suffice for most initial installations, you may wish to edit this file to change the bound IP address and/or port number, or to make performance-related adjustments. See [the uWSGI documentation](https://uwsgi-docs-additions.readthedocs.io/en/latest/Options.html) for the available configuration parameters and take a minute to review the [Things to know](https://uwsgi-docs.readthedocs.io/en/latest/ThingsToKnow.html) page. Django also provides [additional documentation](https://docs.djangoproject.com/en/stable/howto/deployment/wsgi/uwsgi/) on configuring uWSGI with a Django app.
 
 ## systemd Setup

docs/integrations/prometheus-metrics.md

@@ -13,6 +13,7 @@ NetBox makes use of the [django-prometheus](https://github.com/korfuri/django-pr
 - Per view request latency histograms
 - REST API requests (by endpoint & method)
 - GraphQL API requests
+- Client disconnect counters (by method & view)
 - Request body size histograms
 - Response body size histograms
 - Response code counters
@@ -23,6 +24,12 @@ NetBox makes use of the [django-prometheus](https://github.com/korfuri/django-pr
 
 For the exhaustive list of exposed metrics, visit the `/metrics` endpoint on your NetBox instance.
 
+## Client Disconnects
+
+The `netbox_client_disconnects_total` counter, labelled by `method` and `view`, records requests which NetBox aborted because the HTTP client disconnected before the response was sent. Together with the `netbox.disconnect` log, this identifies which endpoints are accumulating abandoned requests and which clients are generating them.
+
+This counter is populated only when [`ABORT_ON_CLIENT_DISCONNECT`](../configuration/system.md#abort_on_client_disconnect) is enabled and NetBox is running under a supported WSGI server. It remains at zero otherwise: a zero value means NetBox is taking no action on client disconnects, not that no clients are disconnecting.
+
 ## Multi Processing Notes
 
 When deploying NetBox in a multiprocess manner (e.g. running multiple Gunicorn workers) the Prometheus client library requires the use of a shared directory to collect metrics from all worker processes. To configure this, first create or designate a local directory to which the worker processes have read and write access, and then configure your WSGI service (e.g. Gunicorn) to define this path as the `prometheus_multiproc_dir` environment variable.

netbox/netbox/disconnect.py

@@ -0,0 +1,705 @@
+import dataclasses
+import enum
+import errno
+import itertools
+import logging
+import os
+import select
+import socket
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any
+
+from django.db import DEFAULT_DB_ALIAS
+from django.db.backends.base.base import BaseDatabaseWrapper
+
+__all__ = (
+    'ARMED_METHODS',
+    'HTTP_499_CLIENT_CLOSED_REQUEST',
+    'SQLSTATE_QUERY_CANCELED',
+    'CancelTarget',
+    'ClientDisconnectWatchdog',
+    'Registration',
+    'RegistrationState',
+    'get_client_fd',
+    'get_watchdog',
+    'server_exposes_client_socket',
+)
+
+logger = logging.getLogger('netbox.disconnect')
+
+# HTTP methods for which the watchdog is armed. Cancelling an unsafe method which the client has
+# abandoned would silently roll back a write the client may believe has landed, so only safe methods
+# are watched. This is a tuning constant, not a policy knob: it is deliberately not user-configurable.
+ARMED_METHODS = frozenset(('GET', 'HEAD', 'OPTIONS'))
+
+# Seconds between poll() wakeups. Shutdown latency does not depend on this value; a self-pipe is used
+# to interrupt the loop immediately.
+POLL_INTERVAL = 1.0
+
+# Per-connection budget passed to cancel_safe(). psycopg's default of 30 seconds is far too long to
+# tie up a cancellation worker, and a timeout of zero means "no deadline at all".
+CANCEL_TIMEOUT = 2.0
+
+# Total budget for cancelling every connection registered to a single request.
+CANCEL_BUDGET = 6.0
+
+# Size of the pool used to dispatch cancellations off the poll loop. Bounded so that a disconnect
+# storm (e.g. a load balancer dropping every connection at once) cannot spawn unbounded threads.
+CANCEL_WORKERS = 4
+
+# Backstop TTL for a registration. The middleware releases in a finally block, so this should be
+# unreachable; it exists because a silently leaked registration holds a file descriptor open.
+REGISTRATION_MAX_AGE = 3600
+
+# PostgreSQL SQLSTATE raised when a query is cancelled (psycopg.errors.QueryCanceled).
+SQLSTATE_QUERY_CANCELED = '57014'
+
+# Django provides no class for 499, and it is absent from http.HTTPStatus, so the reason phrase must
+# be supplied explicitly or Django reports "Unknown Status Code".
+HTTP_499_CLIENT_CLOSED_REQUEST = 499
+
+# MSG_DONTWAIT keeps the peek non-blocking without setting O_NONBLOCK on the socket. That distinction
+# is critical: O_NONBLOCK lives in the open file description, which dup() shares, so setting it would
+# also flip the WSGI server's own client socket to non-blocking. MSG_DONTWAIT is absent on some
+# platforms; those reach this module only via runserver, where it is already disabled.
+PEEK_FLAGS = socket.MSG_PEEK | getattr(socket, 'MSG_DONTWAIT', 0)
+
+# poll() events which report that the peer has gone, independently of any data still queued for
+# reading. POLLRDHUP (the peer sent FIN) is Linux-specific; elsewhere only POLLHUP and POLLERR, which
+# poll() reports whether or not they are requested, are available.
+POLLRDHUP = getattr(select, 'POLLRDHUP', 0)
+HANGUP_EVENTS = select.POLLHUP | select.POLLERR | POLLRDHUP
+
+# errno values from a peek which indicate the peer is gone.
+DISCONNECT_ERRNOS = frozenset((
+    errno.ECONNRESET,
+    errno.ENOTCONN,
+    errno.EPIPE,
+    errno.ETIMEDOUT,
+))
+
+
+class RegistrationState(enum.IntEnum):
+    """
+    Lifecycle of a single watched request. Every transition is performed under the watchdog's lock,
+    and both ARMED -> CANCELLING (claimed by the watchdog) and ARMED -> RELEASED (reclaimed by the
+    request thread) require the state to still be ARMED, so exactly one of them can win.
+    """
+    ARMED = 0        # The watchdog owns this registration and may still cancel it
+    CANCELLING = 1   # The watchdog has claimed it; cancellation is in flight
+    CANCELLED = 2    # The cancellation attempt has finished (successfully or not)
+    RELEASED = 3     # The request thread reclaimed it; the watchdog must not touch it
+
+
+class CancelTarget:
+    """
+    A single database connection which may need to be cancelled on behalf of a request.
+
+    The psycopg connection object is captured at registration time and compared by identity before
+    cancelling. Without that check, a wrapper which has since reconnected (connections persist for
+    CONN_MAX_AGE) would have its *new* backend cancelled instead.
+    """
+    __slots__ = ('alias', 'backend_pid', 'pgconn', 'wrapper')
+
+    def __init__(self, alias: str, wrapper: BaseDatabaseWrapper, pgconn: Any, backend_pid: int | None = None):
+        self.alias = alias
+        self.wrapper = wrapper
+        self.pgconn = pgconn
+        self.backend_pid = backend_pid
+
+    def __repr__(self):
+        return f'<CancelTarget {self.alias} pid={self.backend_pid}>'
+
+    def terminate(self):
+        """
+        End this connection's database session, so that no cancellation, however late, can land on a
+        query issued over it afterwards. Connection.close() is deliberately not used: a pool
+        configured with close_returns would take the connection back with its session still open.
+        """
+        if not self.pgconn.closed:
+            self.pgconn.pgconn.finish()
+
+
+@dataclasses.dataclass
+class Registration:
+    """
+    One in-flight request being watched. This object is handed back to the caller by register() and
+    passed to release() verbatim, so the watchdog never needs to key anything on the request ID and
+    there is no per-request bookkeeping to leak: the registration dies with the request.
+    """
+    token: int
+    request_id: str
+    method: str
+    path: str
+    started: float
+    peek_sock: socket.socket
+    targets: tuple
+    watchdog: Any = dataclasses.field(repr=False, default=None)
+    state: RegistrationState = RegistrationState.ARMED
+    # Aliases to which a cancellation is being or has been sent, less any whose cancellation failed
+    cancelled_aliases: tuple = ()
+    saw_buffered_data: bool = False
+    handed_off: tuple = ()
+
+    def elapsed(self):
+        return time.monotonic() - self.started
+
+
+def get_client_fd(request):
+    """
+    Return the file descriptor of the client connection for this request, or None if it cannot be
+    determined. Resolving the socket is WSGI server-specific, so each supported server is tried in
+    turn.
+
+    A None return is the expected outcome under the development server, under an unrecognised WSGI
+    server, and in tests. Callers must treat it as "disable silently", never as an error.
+    """
+    # gunicorn places the live client socket object in the WSGI environ. Django's WSGIRequest aliases
+    # request.META to that environ, so it is reachable here.
+    sock = request.META.get('gunicorn.socket')
+    if sock is not None:
+        try:
+            fd = sock.fileno()
+        except OSError:
+            logger.debug("gunicorn.socket present but its fd could not be read", exc_info=True)
+        else:
+            if fd >= 0:
+                return fd
+
+    # uWSGI exposes the connection fd through its extension module, which exists only when actually
+    # running under uWSGI. Import it lazily and tolerate its absence.
+    try:
+        import uwsgi
+        fd = uwsgi.connection_fd()
+    except Exception:
+        pass
+    else:
+        if isinstance(fd, int) and fd >= 0:
+            return fd
+
+    return None
+
+
+def server_exposes_client_socket(request):
+    """
+    Report whether the WSGI server is one which get_client_fd() supports, regardless of whether this
+    particular request's socket was usable. A supported server can still yield no fd for a single
+    request (e.g. the client socket was closed before the request reached us), and that must not be
+    mistaken for an unsupported server.
+    """
+    if 'gunicorn.socket' in request.META:
+        return True
+    try:
+        import uwsgi  # noqa: F401
+    except ImportError:
+        return False
+    return True
+
+
+class ClientDisconnectWatchdog:
+    """
+    A single thread per worker process which watches the client sockets of all in-flight requests and
+    cancels their database queries when a client goes away.
+
+    One thread is used rather than one per request because NetBox's own gunicorn configuration runs
+    multiple request threads per worker; a thread per request would double the concurrency footprint
+    for no benefit.
+    """
+
+    def __init__(self):
+        self._lock = threading.RLock()
+        self._by_token = {}    # token -> Registration
+        self._by_fd = {}       # our duplicated fd -> token
+        self._tokens = itertools.count(1)
+        self._poll = select.poll()
+        self._stopping = threading.Event()
+        self._thread = None
+        self._executor = None
+        self._last_reap = time.monotonic()
+
+        # Self-pipe, so that shutdown and new registrations interrupt poll() immediately rather than
+        # waiting out POLL_INTERVAL.
+        self._wake_r, self._wake_w = os.pipe()
+        os.set_blocking(self._wake_r, False)
+        os.set_blocking(self._wake_w, False)
+        self._poll.register(self._wake_r, select.POLLIN)
+
+    #
+    # Request-thread API
+    #
+
+    def register(self, request, fd, targets):
+        """
+        Begin watching the given client fd on behalf of a request. Returns a Registration to be
+        passed to release(), or None if the registration could not be made.
+        """
+        if not targets or self._stopping.is_set():
+            return None
+
+        # Duplicate the descriptor and take ownership of the copy. The original belongs to the WSGI
+        # server, which may close it at any time; once closed, the kernel is free to reissue that
+        # integer to an unrelated connection, and polling it would then be both a use-after-close and
+        # a route to cancelling the wrong request. Our duplicate cannot be reissued while we hold it.
+        try:
+            dup_fd = os.dup(fd)
+        except OSError:
+            logger.debug("Unable to duplicate client fd %s", fd, exc_info=True)
+            return None
+        try:
+            blocking = os.get_blocking(dup_fd)
+            peek_sock = socket.socket(fileno=dup_fd)
+        except OSError:
+            os.close(dup_fd)
+            logger.debug("Unable to adopt duplicated fd %s", dup_fd, exc_info=True)
+            return None
+
+        # If anything in the process has called socket.setdefaulttimeout(), adopting the fd above set
+        # O_NONBLOCK on it, and so on the WSGI server's own socket (see PEEK_FLAGS). Clear the timeout
+        # so that recv() does not wait for readability, then restore the server's blocking mode
+        # exactly as found: some servers (e.g. gevent workers) run their sockets non-blocking.
+        try:
+            peek_sock.settimeout(None)
+            os.set_blocking(dup_fd, blocking)
+        except OSError:
+            peek_sock.close()
+            logger.debug("Unable to restore blocking mode of fd %s", dup_fd, exc_info=True)
+            return None
+
+        # Cancel the default database first, so that a slow secondary cannot exhaust the budget
+        # before the connection the request is most likely blocked on has been dealt with.
+        targets = tuple(sorted(targets, key=lambda target: target.alias != DEFAULT_DB_ALIAS))
+
+        registration = Registration(
+            token=next(self._tokens),
+            request_id=str(getattr(request, 'id', '')),
+            method=request.method,
+            path=request.path,
+            started=time.monotonic(),
+            peek_sock=peek_sock,
+            targets=targets,
+            watchdog=self,
+        )
+
+        with self._lock:
+            self._by_token[registration.token] = registration
+            self._by_fd[peek_sock.fileno()] = registration.token
+            self._poll.register(peek_sock.fileno(), select.POLLIN)
+
+        self._wake()
+        return registration
+
+    def release(self, registration):
+        """
+        Reclaim a registration and return the state observed at the moment it was reclaimed.
+
+        The return value is the entire cancel/release race protocol. If it is ARMED, the watchdog
+        never touched this request. Anything else means a cancellation was claimed and may still be
+        in flight, and the caller must treat every target connection as unusable.
+        """
+        with self._lock:
+            observed = registration.state
+            if observed is RegistrationState.ARMED:
+                registration.state = RegistrationState.RELEASED
+            # A claimed registration keeps its state: the watchdog still owns the in-flight
+            # cancellation, and the caller needs to see that it lost the race.
+            self._discard(registration)
+
+        # Closing the socket releases our duplicated fd. It happens outside the lock, and only the
+        # side which removed the entry from the registry performs it, so it happens exactly once.
+        self._close(registration)
+        return observed
+
+    def hand_off(self, registration, targets):
+        """
+        Pass the given targets to a cancellation still in flight for this registration, which will
+        terminate them once it has finished. Returns False if no cancellation is in flight, in which
+        case the targets remain the caller's to dispose of.
+
+        Neither thread may dispose of a connection while the other is still using it: closing one
+        mid-cancellation races libpq, and with connection pooling it would hand the backend to another
+        request just before the cancellation lands on it. So whichever finishes last disposes of it.
+        """
+        with self._lock:
+            if registration.state is not RegistrationState.CANCELLING:
+                return False
+            registration.handed_off = tuple(targets)
+            return True
+
+    #
+    # Thread lifecycle
+    #
+
+    def start(self):
+        with self._lock:
+            if self._thread is not None and self._thread.is_alive():
+                return
+            self._stopping.clear()
+            if self._executor is not None:
+                # Restarting after the thread died. Queued cancellations still run; the old workers
+                # exit once they have drained.
+                self._executor.shutdown(wait=False)
+            self._executor = ThreadPoolExecutor(
+                max_workers=CANCEL_WORKERS,
+                thread_name_prefix='netbox-cancel',
+            )
+            self._thread = threading.Thread(
+                target=self._run,
+                name='netbox-disconnect-watchdog',
+                daemon=True,
+            )
+            self._thread.start()
+
+    def is_alive(self):
+        thread = self._thread
+        return thread is not None and thread.is_alive()
+
+    def shutdown(self, timeout=5.0):
+        self._stopping.set()
+        self._wake()
+        thread = self._thread
+        if thread is not None:
+            thread.join(timeout)
+        executor = self._executor
+        if executor is not None:
+            executor.shutdown(wait=False)
+        with self._lock:
+            registrations = list(self._by_token.values())
+            for registration in registrations:
+                self._discard(registration)
+        for registration in registrations:
+            self._close(registration)
+
+    def abandon_after_fork(self):
+        """
+        Drop state inherited from a parent process. The watchdog thread does not survive fork, so the
+        registry describes requests this process never served; the descriptors in it belong to
+        sockets owned by the parent and must be released without being shut down.
+
+        The lock is deliberately not taken. Only the calling thread survives a fork, and if another
+        thread in the parent held the lock at the moment of forking, it would never be released here.
+        """
+        registrations = list(self._by_token.values())
+        self._by_token.clear()
+        self._by_fd.clear()
+        self._thread = None
+        self._executor = None
+        for registration in registrations:
+            self._close(registration)
+        # The self-pipe was inherited too; closing this process's copies leaves the parent's intact.
+        for fd in (self._wake_r, self._wake_w):
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+
+    #
+    # Internals
+    #
+
+    def _discard(self, registration):
+        """Remove a registration from the registry. Must be called with the lock held."""
+        self._by_token.pop(registration.token, None)
+        fd = self._fileno(registration)
+        if fd >= 0:
+            self._by_fd.pop(fd, None)
+            self._unpoll(fd)
+
+    def _unpoll(self, fd):
+        """Stop polling a descriptor. Must be called with the lock held."""
+        try:
+            self._poll.unregister(fd)
+        except (KeyError, OSError):
+            pass
+
+    @staticmethod
+    def _fileno(registration):
+        try:
+            return registration.peek_sock.fileno()
+        except OSError:
+            return -1
+
+    @staticmethod
+    def _close(registration):
+        try:
+            registration.peek_sock.close()
+        except OSError:
+            pass
+
+    def _wake(self):
+        try:
+            os.write(self._wake_w, b'\x00')
+        except (BlockingIOError, OSError):
+            # A full pipe already carries a pending wakeup, which is all we need.
+            pass
+
+    def _drain_wake(self):
+        while True:
+            try:
+                if not os.read(self._wake_r, 4096):
+                    return
+            except (BlockingIOError, OSError):
+                return
+
+    def _run(self):
+        timeout_ms = max(int(POLL_INTERVAL * 1000), 1)
+        while not self._stopping.is_set():
+            try:
+                events = self._poll.poll(timeout_ms)
+            except OSError as exc:
+                logger.warning("poll() failed: %s", exc)
+                # Back off rather than retrying at once: a persistent failure (e.g. ENOMEM) would
+                # otherwise spin this thread and flood the log. Shutdown still interrupts the wait.
+                self._stopping.wait(POLL_INTERVAL)
+                continue
+            for fd, mask in events:
+                if fd == self._wake_r:
+                    self._drain_wake()
+                    continue
+                try:
+                    self._handle(fd, mask)
+                except Exception:
+                    # The watchdog must outlive any single bad registration.
+                    logger.exception("Error handling watched fd %s", fd)
+            self._reap_stale()
+
+    def _handle(self, fd, mask):
+        with self._lock:
+            token = self._by_fd.get(fd)
+            registration = self._by_token.get(token) if token is not None else None
+            if registration is None or registration.state is not RegistrationState.ARMED:
+                return
+            # Guard against acting on an event which was queued before this fd was recycled into a
+            # different registration.
+            try:
+                if registration.peek_sock.fileno() != fd:
+                    return
+            except OSError:
+                return
+
+        if mask & select.POLLNVAL:
+            if not self._poll_now(fd) & select.POLLNVAL:
+                # Stale: the event belonged to a registration released while poll() was running, and
+                # its descriptor number has since been reissued to this one, which is valid.
+                return
+            # We own this descriptor for the lifetime of the registration, so this should be
+            # unreachable. It means the invariant is broken and the fd may already have been reused,
+            # so the one thing we must not do is cancel anything.
+            logger.warning("POLLNVAL on owned fd %s (request %s)", fd, registration.request_id)
+            self._force_release(registration)
+            return
+
+        if registration.saw_buffered_data:
+            # Data is queued ahead of any end-of-stream, so a peek can no longer detect it. Rely on
+            # the hangup events instead, re-polled now rather than read from the mask, which may be
+            # stale.
+            if self._is_hung_up(fd):
+                self._claim_and_cancel(registration)
+            return
+
+        # Always confirm with a peek, including for POLLHUP and POLLERR. Those flags are treated as a
+        # hint rather than a verdict so that a stale event cannot cancel a request which happens to
+        # have inherited the same descriptor number.
+        try:
+            data = registration.peek_sock.recv(1, PEEK_FLAGS)
+        except (BlockingIOError, InterruptedError):
+            # Spurious wakeup; nothing to conclude.
+            return
+        except ConnectionResetError:
+            self._claim_and_cancel(registration)
+            return
+        except OSError as exc:
+            if exc.errno in DISCONNECT_ERRNOS:
+                self._claim_and_cancel(registration)
+            elif exc.errno == errno.EBADF:
+                logger.warning("EBADF on owned fd %s (request %s)", fd, registration.request_id)
+                self._force_release(registration)
+            else:
+                logger.warning("Failed to peek at fd %s: %s", fd, exc)
+                self._force_release(registration)
+            return
+
+        if data == b'':
+            self._claim_and_cancel(registration)
+            return
+
+        # Data is queued which the request has not read: a pipelined request, an unread body, or a TLS
+        # record such as close_notify. poll() is level-triggered, so the fd would now report POLLIN on
+        # every pass and spin this thread. Stop watching for it and rely on the hangup events, which
+        # FIN and RST raise regardless of what is queued ahead of them. The peer may already have gone
+        # (e.g. close_notify followed by FIN), so check immediately as well.
+        registration.saw_buffered_data = True
+        with self._lock:
+            if self._by_fd.get(fd) == registration.token:
+                try:
+                    self._poll.modify(fd, POLLRDHUP)
+                except (KeyError, OSError):
+                    pass
+        if self._is_hung_up(fd):
+            self._claim_and_cancel(registration)
+
+    @staticmethod
+    def _poll_now(fd):
+        """Return the fd's poll() events as of now, rather than as of an event already queued."""
+        probe = select.poll()
+        probe.register(fd, POLLRDHUP)
+        try:
+            events = probe.poll(0)
+        except OSError:
+            return 0
+        return events[0][1] if events else 0
+
+    @classmethod
+    def _is_hung_up(cls, fd):
+        """Report whether the peer has closed or reset the connection, regardless of queued data."""
+        mask = cls._poll_now(fd)
+        return bool(mask & HANGUP_EVENTS) and not mask & select.POLLNVAL
+
+    def _force_release(self, registration):
+        """Evict a registration from the watchdog side, without cancelling anything."""
+        with self._lock:
+            self._discard(registration)
+        self._close(registration)
+
+    def _claim_and_cancel(self, registration):
+        with self._lock:
+            if registration.state is not RegistrationState.ARMED:
+                return
+            registration.state = RegistrationState.CANCELLING
+            # A disconnected fd reports POLLHUP or POLLIN on every pass, and POLLHUP and POLLERR
+            # cannot be masked out, so leaving it in the poll set would spin this thread until the
+            # request thread releases the registration. The fd itself stays open, and registered,
+            # until then.
+            fd = self._fileno(registration)
+            if fd >= 0:
+                self._unpoll(fd)
+        logger.debug(
+            "Client disconnected during request %s (%s %s)",
+            registration.request_id, registration.method, registration.path,
+        )
+        executor = self._executor
+        if executor is None:
+            self._cancel_all(registration)
+        else:
+            # Cancellation opens a fresh authenticated connection to PostgreSQL, which may take
+            # hundreds of milliseconds and can block past its own deadline while libpq resolves the
+            # host. It must not run on the poll loop, where it would stall every other watched request.
+            executor.submit(self._cancel_all, registration)
+
+    def _cancel_all(self, registration):
+        deadline = time.monotonic() + CANCEL_BUDGET
+        try:
+            for target in registration.targets:
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    logger.warning(
+                        "Cancellation budget exhausted for request %s; %s not cancelled",
+                        registration.request_id, target.alias,
+                    )
+                    break
+                try:
+                    # Only ever read an attribute of the wrapper. Its methods call
+                    # validate_thread_sharing() and would raise if called from this thread.
+                    pgconn = target.wrapper.connection
+                    if pgconn is None or pgconn is not target.pgconn or pgconn.closed:
+                        # The wrapper has reconnected or closed since registration, so whatever it
+                        # holds now is not the query we set out to cancel.
+                        continue
+                    # Record the alias before sending: the request thread may receive the resulting
+                    # error, and log the abort, before cancel_safe() has returned here.
+                    registration.cancelled_aliases += (target.alias,)
+                    pgconn.cancel_safe(timeout=min(CANCEL_TIMEOUT, remaining))
+                except Exception as exc:
+                    # One connection failing to cancel must not prevent the others from being tried.
+                    registration.cancelled_aliases = tuple(
+                        alias for alias in registration.cancelled_aliases if alias != target.alias
+                    )
+                    logger.warning(
+                        "Failed to cancel query for request %s on database '%s': %s",
+                        registration.request_id, target.alias, exc,
+                    )
+        finally:
+            with self._lock:
+                if registration.state is RegistrationState.CANCELLING:
+                    registration.state = RegistrationState.CANCELLED
+                handed_off = registration.handed_off
+            for target in handed_off:
+                self._dispose(registration, target)
+
+    @staticmethod
+    def _dispose(registration, target):
+        """Terminate a connection handed off by the request thread, which has already detached it."""
+        try:
+            target.terminate()
+            # Return a pooled connection even though it is now dead: abandoning it would permanently
+            # cost the pool a slot. The pool discards it and opens a replacement.
+            pool = getattr(target.pgconn, '_pool', None)
+            if pool is not None:
+                pool.putconn(target.pgconn)
+        except Exception as exc:
+            logger.warning(
+                "Failed to dispose of connection for request %s on database '%s': %s",
+                registration.request_id, target.alias, exc,
+            )
+
+    def _reap_stale(self):
+        now = time.monotonic()
+        if now - self._last_reap < 1.0:
+            return
+        self._last_reap = now
+        with self._lock:
+            stale = [
+                registration for registration in self._by_token.values()
+                if now - registration.started > REGISTRATION_MAX_AGE
+            ]
+            for registration in stale:
+                self._discard(registration)
+        for registration in stale:
+            logger.error(
+                "Reaped a stale client disconnect registration for request %s after %.0fs; "
+                "this indicates a leaked registration",
+                registration.request_id, registration.elapsed(),
+            )
+            self._close(registration)
+
+
+_watchdog = None
+_watchdog_pid = None
+_watchdog_lock = threading.Lock()
+
+
+def get_watchdog():
+    """
+    Return this process's watchdog, starting it if necessary.
+
+    The thread is created lazily on first use rather than at import or application-ready time. Under
+    uWSGI's default (non-lazy-apps) configuration the WSGI application is loaded in the master process
+    and then forked; a thread started there would live only in the master and would be absent from
+    every process which actually serves requests. Creating it on the first watched request guarantees
+    it is created post-fork, in the worker that needs it.
+    """
+    global _watchdog, _watchdog_pid
+
+    pid = os.getpid()
+    watchdog = _watchdog
+    if watchdog is not None and _watchdog_pid == pid and watchdog.is_alive():
+        return watchdog
+
+    with _watchdog_lock:
+        watchdog = _watchdog
+        if watchdog is None or _watchdog_pid != pid or not watchdog.is_alive():
+            if watchdog is not None and _watchdog_pid != pid:
+                # Inherited across a fork: the registry describes the parent's requests and the
+                # thread does not exist here.
+                watchdog.abandon_after_fork()
+                watchdog = None
+            elif watchdog is not None:
+                # Restart in place, so that requests already registered remain watched.
+                logger.error("Client disconnect watchdog thread died unexpectedly; restarting it")
+            if watchdog is None:
+                watchdog = ClientDisconnectWatchdog()
+            watchdog.start()
+            _watchdog = watchdog
+            _watchdog_pid = pid
+        return _watchdog

netbox/netbox/metrics.py

@@ -1,9 +1,11 @@
+from django.conf import settings
 from django_prometheus import middleware
 from django_prometheus.conf import NAMESPACE
 from prometheus_client import Counter
 
 __all__ = (
     'Metrics',
+    'increment_client_disconnects',
 )
 
 
@@ -38,3 +40,27 @@ class Metrics(middleware.Metrics):
             "Count of total GraphQL API requests",
             namespace=NAMESPACE,
         )
+
+        # Client disconnect metrics
+        self.client_disconnects = self.register_metric(
+            Counter,
+            "netbox_client_disconnects_total",
+            "Count of requests aborted because the HTTP client disconnected, by method & view",
+            ["method", "view"],
+            namespace=NAMESPACE,
+        )
+
+
+def increment_client_disconnects(method, view):
+    """
+    Increment the client disconnect counter.
+
+    ClientDisconnectMiddleware is not a django_prometheus middleware and so has no Metrics instance
+    of its own. Instantiating the singleton unconditionally would register the entire django_prometheus
+    metric set on installations which never expose /metrics, so this is a no-op unless metric
+    exposition is enabled. Always go through get_instance(): calling Metrics() directly bypasses the
+    singleton and re-registers every metric name in the global registry.
+    """
+    if not settings.METRICS_ENABLED:
+        return
+    Metrics.get_instance().client_disconnects.labels(method=method, view=view).inc()

netbox/netbox/middleware.py

@@ -4,24 +4,35 @@ import uuid
 from django.conf import settings
 from django.contrib import auth, messages
 from django.contrib.auth.middleware import RemoteUserMiddleware as RemoteUserMiddleware_
-from django.core.exceptions import ImproperlyConfigured
+from django.core.exceptions import ImproperlyConfigured, MiddlewareNotUsed
 from django.core.signals import got_request_exception
-from django.db import ProgrammingError, connection
-from django.db.utils import InternalError
-from django.http import Http404, HttpResponseRedirect
+from django.db import DEFAULT_DB_ALIAS, ProgrammingError, connection, connections
+from django.db.utils import InternalError, OperationalError
+from django.http import Http404, HttpResponse, HttpResponseRedirect
 from django.middleware.common import CommonMiddleware as DjangoCommonMiddleware
 from django.utils.translation import gettext_lazy as _
 from django_prometheus import middleware
 from social_django.middleware import SocialAuthExceptionMiddleware as SocialAuthExceptionMiddleware_
 
 from netbox.config import clear_config, get_config
-from netbox.metrics import Metrics
+from netbox.disconnect import (
+    ARMED_METHODS,
+    HTTP_499_CLIENT_CLOSED_REQUEST,
+    SQLSTATE_QUERY_CANCELED,
+    CancelTarget,
+    RegistrationState,
+    get_client_fd,
+    get_watchdog,
+    server_exposes_client_socket,
+)
+from netbox.metrics import Metrics, increment_client_disconnects
 from netbox.views import handler_500
 from utilities.api import is_api_request, is_graphql_request
 from utilities.error_handlers import handle_rest_api_exception
-from utilities.request import apply_request_processors
+from utilities.request import apply_request_processors, get_client_ip
 
 __all__ = (
+    'ClientDisconnectMiddleware',
     'CommonMiddleware',
     'CoreMiddleware',
     'MaintenanceModeMiddleware',
@@ -31,6 +42,8 @@ __all__ = (
     'SocialAuthExceptionMiddleware',
 )
 
+disconnect_logger = logging.getLogger('netbox.disconnect')
+
 
 class CommonMiddleware(DjangoCommonMiddleware):
     """
@@ -134,6 +147,215 @@ class CoreMiddleware:
         return None
 
 
+class ClientDisconnectMiddleware:
+    """
+    Cancel a request's in-flight database queries when the HTTP client disconnects before the
+    response has been sent.
+
+    WSGI provides no cancellation mechanism, so a worker ordinarily runs an abandoned request to
+    completion and discovers the disconnect only when it attempts to write the response. A client
+    which times out aggressively and retries therefore adds a further orphaned request on each
+    attempt, and can saturate every worker and database backend on its own.
+
+    This middleware hands the client socket and the request's database connections to a per-process
+    watchdog, which cancels those queries as soon as the client goes away. It disables itself
+    silently when the WSGI server does not expose the client socket, which is the expected outcome
+    under the development server and in tests.
+    """
+
+    def __init__(self, get_response):
+        # Removing the middleware from the chain outright is considerably cheaper than leaving an
+        # inert one in it, and this is disabled by default.
+        if not settings.ABORT_ON_CLIENT_DISCONNECT:
+            raise MiddlewareNotUsed()
+        self.get_response = get_response
+        self._unsupported = False
+
+    def __call__(self, request):
+        registration = self._arm(request)
+        if registration is None:
+            return self.get_response(request)
+
+        request._client_disconnect = registration
+        try:
+            return self.get_response(request)
+        finally:
+            # Releasing must happen whatever the outcome: a registration left behind would let the
+            # watchdog cancel queries belonging to whichever request next uses this worker thread.
+            observed = registration.watchdog.release(registration)
+            self._cleanup(registration, observed)
+            request._client_disconnect = None
+
+    def _arm(self, request):
+        """
+        Register this request with the watchdog, returning the Registration or None if the request
+        is not being watched.
+        """
+        if self._unsupported or request.method not in ARMED_METHODS:
+            return None
+
+        fd = get_client_fd(request)
+        if fd is None:
+            if server_exposes_client_socket(request):
+                # The server is supported but this request's socket was unusable (e.g. already
+                # closed). Leave only this request unwatched.
+                return None
+            # Latch off for the lifetime of this middleware instance, and say so exactly once. This
+            # is a supported configuration, not an error.
+            self._unsupported = True
+            disconnect_logger.info(
+                "Client disconnect detection is unavailable: this WSGI server does not expose the "
+                "client socket. Requests will not be aborted when clients disconnect."
+            )
+            return None
+
+        targets = self._get_cancel_targets()
+        if not targets:
+            return None
+
+        try:
+            return get_watchdog().register(request, fd, targets)
+        except Exception as exc:
+            # Starting the watchdog can fail under resource exhaustion (e.g. EMFILE from os.pipe(), or
+            # no thread available). That must cost this request its protection, not its response. It
+            # is not latched off: the next request retries, so the feature recovers with the process.
+            disconnect_logger.warning(
+                "Unable to watch request %s for client disconnect: %s", getattr(request, 'id', ''), exc
+            )
+            return None
+
+    @staticmethod
+    def _get_cancel_targets():
+        """
+        Capture the database connections to cancel on disconnect.
+
+        This must happen on the request thread: django.db.connections is thread-local, so the
+        watchdog cannot look these up for itself. Connections already open in this thread are used
+        as-is; opening every configured alias here would force a handshake per alias on each cold
+        request, for aliases the request may never touch.
+        """
+        targets = []
+        try:
+            connections[DEFAULT_DB_ALIAS].ensure_connection()
+        except Exception:
+            disconnect_logger.debug("Unable to establish the default database connection", exc_info=True)
+
+        for wrapper in connections.all(initialized_only=True):
+            if wrapper.vendor != 'postgresql':
+                continue
+            try:
+                pgconn = wrapper.connection
+                if pgconn is None or not hasattr(pgconn, 'cancel_safe'):
+                    continue
+                targets.append(CancelTarget(
+                    alias=wrapper.alias,
+                    wrapper=wrapper,
+                    pgconn=pgconn,
+                    backend_pid=getattr(getattr(pgconn, 'info', None), 'backend_pid', None),
+                ))
+            except Exception:
+                disconnect_logger.debug("Skipping database '%s'", wrapper.alias, exc_info=True)
+
+        return targets
+
+    @staticmethod
+    def _cleanup(registration, observed):
+        """
+        Discard any connection which may have been cancelled.
+
+        Driven by the state observed when the registration was reclaimed rather than by whether an
+        exception was seen, because a cancellation which lost the race may still be in flight even
+        though the request completed normally.
+        """
+        if observed is RegistrationState.ARMED:
+            # The watchdog never touched this request, so leave connection reuse alone.
+            return
+
+        # A connection no longer held by the wrapper it was registered with has already been closed, or
+        # replaced by a reconnect, and is not ours to discard.
+        targets = [target for target in registration.targets if target.wrapper.connection is target.pgconn]
+
+        if registration.watchdog.hand_off(registration, targets):
+            # The cancellation is still in flight, so these connections must not be touched here: the
+            # cancellation worker terminates them once it has finished. Detach them so that Django
+            # opens fresh ones when next needed.
+            for target in targets:
+                ClientDisconnectMiddleware._detach(target.wrapper)
+            return
+
+        for target in targets:
+            try:
+                # set_rollback() is only meaningful inside an atomic block, and raises otherwise.
+                # Django has normally unwound every atomic block by this point, so this covers only
+                # those paths which swallowed the exception.
+                if target.wrapper.in_atomic_block:
+                    target.wrapper.set_rollback(True)
+                # Terminate unconditionally rather than deferring to close_if_unusable_or_obsolete(): a
+                # cancellation which found nothing to cancel leaves a perfectly usable connection, but
+                # one which failed or timed out may still land. Terminating before closing matters
+                # under connection pooling, where close() alone would return the session to the pool
+                # for another request to inherit.
+                target.terminate()
+                target.wrapper.close()
+            except Exception:
+                disconnect_logger.exception(
+                    "Error discarding database connection '%s' after cancellation", target.alias
+                )
+
+    @staticmethod
+    def _detach(wrapper):
+        """
+        Drop a wrapper's connection without closing it, mirroring the bookkeeping of
+        BaseDatabaseWrapper.close().
+        """
+        wrapper.run_on_commit = []
+        if wrapper.in_atomic_block:
+            wrapper.closed_in_transaction = True
+            wrapper.needs_rollback = True
+        wrapper.connection = None
+
+    def process_exception(self, request, exception):
+        """
+        Convert a query cancellation caused by a client disconnect into a synthetic 499.
+
+        This runs before CoreMiddleware.process_exception(), so returning a response here keeps the
+        exception out of handler_500() and out of got_request_exception, and error-tracking
+        integrations are not flooded with self-inflicted HTTP 500 reports. The response itself is
+        never delivered: the client has already gone.
+        """
+        if not isinstance(exception, OperationalError):
+            return None
+        if getattr(exception.__cause__, 'sqlstate', None) != SQLSTATE_QUERY_CANCELED:
+            return None
+
+        # The SQLSTATE alone is ambiguous: an operator's statement_timeout produces the same code.
+        # Only a registration which the watchdog actually claimed identifies this as our doing.
+        registration = getattr(request, '_client_disconnect', None)
+        if registration is None or registration.state is RegistrationState.ARMED:
+            return None
+
+        try:
+            client_ip = get_client_ip(request)
+        except ValueError:
+            client_ip = None
+
+        disconnect_logger.info(
+            "Aborted request %s after client disconnected: %s %s (%.3fs, client %s, databases: %s)",
+            registration.request_id,
+            request.method,
+            request.path,
+            registration.elapsed(),
+            client_ip or 'unknown',
+            ', '.join(registration.cancelled_aliases) or 'none',
+        )
+
+        resolver_match = getattr(request, 'resolver_match', None)
+        view_name = (resolver_match.view_name if resolver_match is not None else None) or '<unnamed view>'
+        increment_client_disconnects(method=request.method, view=view_name)
+
+        return HttpResponse(status=HTTP_499_CLIENT_CLOSED_REQUEST, reason='Client Closed Request')
+
+
 class RemoteUserMiddleware(RemoteUserMiddleware_):
     """
     Custom implementation of Django's RemoteUserMiddleware which allows for a user-configurable HTTP header name.

netbox/netbox/settings.py

@@ -81,6 +81,7 @@ elif hasattr(configuration, 'DATABASE') and hasattr(configuration, 'DATABASES'):
     raise ImproperlyConfigured("DATABASE and DATABASES may not be set together. The use of DATABASES is encouraged.")
 
 # Set static config parameters
+ABORT_ON_CLIENT_DISCONNECT = getattr(configuration, 'ABORT_ON_CLIENT_DISCONNECT', False)
 ADMINS = getattr(configuration, 'ADMINS', [])
 ALLOWED_HOSTS = getattr(configuration, 'ALLOWED_HOSTS')  # Required
 API_TOKEN_PEPPERS = getattr(configuration, 'API_TOKEN_PEPPERS', {})
@@ -544,6 +545,7 @@ MIDDLEWARE = [
     'django_htmx.middleware.HtmxMiddleware',
     'netbox.middleware.RemoteUserMiddleware',
     'netbox.middleware.CoreMiddleware',
+    'netbox.middleware.ClientDisconnectMiddleware',
     'netbox.middleware.MaintenanceModeMiddleware',
     'netbox.middleware.SocialAuthExceptionMiddleware',
 ]

netbox/netbox/tests/test_disconnect.py

@@ -0,0 +1,659 @@
+import errno
+import fcntl
+import os
+import select
+import socket
+import struct
+import sys
+import threading
+import time
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from django.test import SimpleTestCase
+
+from netbox.disconnect import (
+    CANCEL_TIMEOUT,
+    CancelTarget,
+    ClientDisconnectWatchdog,
+    RegistrationState,
+    get_client_fd,
+    server_exposes_client_socket,
+)
+
+
+class FakePgConn:
+    """
+    Stand-in for a psycopg connection. Only the attributes the watchdog actually touches are
+    implemented: it reads `closed` and calls `cancel_safe()`, and does nothing else.
+    """
+    def __init__(self, fail=False, gate=None):
+        self.closed = False
+        self.fail = fail
+        self.gate = gate
+        self.in_cancel = threading.Event()
+        self.closed_during_cancel = False
+        self.cancelled = threading.Event()
+        self.cancel_timeouts = []
+        self.info = SimpleNamespace(backend_pid=12345)
+        self.pgconn = SimpleNamespace(finish=self._finish)
+
+    def _finish(self):
+        if self.in_cancel.is_set():
+            self.closed_during_cancel = True
+        self.closed = True
+
+    def cancel_safe(self, *, timeout=None):
+        self.in_cancel.set()
+        try:
+            if self.gate is not None:
+                # Hold the cancellation in flight until the test lets it complete.
+                self.gate.wait(5.0)
+            self.cancel_timeouts.append(timeout)
+            self.cancelled.set()
+            if self.fail:
+                raise RuntimeError("simulated cancellation failure")
+        finally:
+            self.in_cancel.clear()
+
+
+class FakeWrapper:
+    """Stand-in for a Django BaseDatabaseWrapper."""
+    def __init__(self, alias='default', pgconn=None):
+        self.alias = alias
+        self.vendor = 'postgresql'
+        self.connection = pgconn if pgconn is not None else FakePgConn()
+        self.in_atomic_block = False
+        self.closed = False
+        self.rollback_set = None
+
+    def set_rollback(self, value):
+        self.rollback_set = value
+
+    def close(self):
+        self.closed = True
+
+
+def make_request(method='GET', path='/dcim/devices/'):
+    return SimpleNamespace(id='11111111-1111-1111-1111-111111111111', method=method, path=path)
+
+
+class ClientDisconnectWatchdogTestCase(SimpleTestCase):
+    """
+    Exercises the watchdog against real socket pairs. socketpair() gives a genuine pollable,
+    closeable, resettable descriptor pair, so none of this needs a WSGI server or a database.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.watchdog = ClientDisconnectWatchdog()
+        self.watchdog.start()
+        self.addCleanup(self.watchdog.shutdown)
+
+    def make_socketpair(self):
+        server_sock, client_sock = socket.socketpair()
+        self.addCleanup(server_sock.close)
+        self.addCleanup(client_sock.close)
+        return server_sock, client_sock
+
+    def make_targets(self, *aliases, fail_on=(), gate=None):
+        targets = []
+        for alias in (aliases or ('default',)):
+            wrapper = FakeWrapper(alias=alias, pgconn=FakePgConn(fail=alias in fail_on, gate=gate))
+            targets.append(CancelTarget(alias, wrapper, wrapper.connection))
+        return targets
+
+    def wait_for(self, predicate, timeout=5.0):
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if predicate():
+                return True
+            time.sleep(0.005)
+        return False
+
+    def assert_not_cancelled(self, targets, settle=0.2):
+        # Give the watchdog several poll cycles to (incorrectly) act before concluding it did not.
+        time.sleep(settle)
+        for target in targets:
+            self.assertFalse(target.pgconn.cancelled.is_set(), f"{target.alias} was cancelled")
+
+    #
+    # Disconnect detection
+    #
+
+    def test_detects_half_close(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.shutdown(socket.SHUT_WR)
+
+        self.assertTrue(targets[0].pgconn.cancelled.wait(5.0))
+
+    def test_detects_full_close(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.close()
+
+        self.assertTrue(targets[0].pgconn.cancelled.wait(5.0))
+
+    def test_detects_reset(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        # SO_LINGER with a zero timeout forces an RST rather than an orderly shutdown.
+        client_sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
+        client_sock.close()
+
+        self.assertTrue(targets[0].pgconn.cancelled.wait(5.0))
+
+    def test_pipelined_data_is_not_a_disconnect(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.sendall(b'GET /api/ HTTP/1.1\r\n')
+
+        self.assert_not_cancelled(targets)
+
+    def test_buffered_data_does_not_spin(self):
+        """
+        poll() is level-triggered, so an fd with unread data reports POLLIN on every pass. The
+        watchdog must stop asking for it once data is seen, or its thread busy-loops for the rest of
+        the request.
+        """
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        with patch.object(self.watchdog, '_handle', wraps=self.watchdog._handle) as handle:
+            client_sock.sendall(b'GET /api/ HTTP/1.1\r\n')
+            time.sleep(0.3)
+
+        self.assertGreaterEqual(handle.call_count, 1)
+        self.assertLessEqual(handle.call_count, 2)
+        self.assert_not_cancelled(targets, settle=0)
+
+    def test_claimed_registration_does_not_spin(self):
+        """
+        A disconnected fd reports POLLHUP or POLLIN on every pass, whatever events are requested. Once
+        the watchdog has claimed the registration it must stop polling the fd, or its thread busy-loops
+        until the request thread gets around to releasing it.
+        """
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.close()
+        self.assertTrue(self.wait_for(lambda: registration.state is RegistrationState.CANCELLED))
+
+        with patch.object(self.watchdog, '_handle', wraps=self.watchdog._handle) as handle:
+            time.sleep(0.3)
+
+        self.assertEqual(handle.call_count, 0)
+
+    def test_poll_failure_backs_off(self):
+        """A persistent poll() failure must not turn the loop into a tight retry that floods the log."""
+        failing_poll = Mock()
+        failing_poll.poll.side_effect = OSError(errno.ENOMEM, "Cannot allocate memory")
+
+        with self.assertLogs('netbox.disconnect', 'WARNING'):
+            self.watchdog._poll = failing_poll
+            # Interrupt the poll() already in progress on the real object.
+            self.watchdog._wake()
+            time.sleep(0.3)
+
+        self.assertEqual(failing_poll.poll.call_count, 1)
+
+    def test_detects_half_close_after_buffered_data(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.sendall(b'GET /api/ HTTP/1.1\r\n')
+        self.assertTrue(self.wait_for(lambda: registration.saw_buffered_data))
+        client_sock.shutdown(socket.SHUT_WR)
+
+        self.assertTrue(targets[0].pgconn.cancelled.wait(5.0))
+
+    def test_detects_close_queued_behind_data(self):
+        """
+        A clean TLS shutdown sends a close_notify record immediately followed by FIN, so both may be
+        queued by the time the watchdog first looks. The data must not mask the disconnect.
+        """
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        client_sock.sendall(b'\x15\x03\x03\x00\x02\x01\x00')
+        client_sock.shutdown(socket.SHUT_WR)
+
+        self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        self.assertTrue(targets[0].pgconn.cancelled.wait(5.0))
+
+    def test_detects_reset_after_buffered_data(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.sendall(b'GET /api/ HTTP/1.1\r\n')
+        self.assertTrue(self.wait_for(lambda: registration.saw_buffered_data))
+        client_sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
+        client_sock.close()
+
+        self.assertTrue(targets[0].pgconn.cancelled.wait(5.0))
+
+    def test_stale_pollnval_is_ignored(self):
+        """
+        poll() can report POLLNVAL for a descriptor released while it was running, which is then
+        reissued to a new registration before the event is handled. That stale event must not evict
+        the new registration.
+        """
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        self.watchdog._handle(registration.peek_sock.fileno(), select.POLLNVAL)
+
+        self.assertIn(registration.token, self.watchdog._by_token)
+        client_sock.close()
+        self.assertTrue(targets[0].pgconn.cancelled.wait(5.0))
+
+    def test_genuine_pollnval_releases_without_cancelling(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        # Break the ownership invariant behind the watchdog's back.
+        with self.assertLogs('netbox.disconnect', 'WARNING'):
+            os.close(registration.peek_sock.fileno())
+            self.assertTrue(self.wait_for(lambda: registration.token not in self.watchdog._by_token))
+
+        self.assert_not_cancelled(targets, settle=0)
+
+    def test_peek_does_not_alter_socket_flags(self):
+        """
+        O_NONBLOCK lives in the open file description, which dup() shares. Setting it on our
+        duplicate would silently flip the WSGI server's own socket to non-blocking, so the peek must
+        use MSG_DONTWAIT instead.
+        """
+        server_sock, client_sock = self.make_socketpair()
+        before = fcntl.fcntl(server_sock.fileno(), fcntl.F_GETFL)
+        targets = self.make_targets()
+        self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.sendall(b'x')
+        time.sleep(0.2)
+
+        after = fcntl.fcntl(server_sock.fileno(), fcntl.F_GETFL)
+        self.assertEqual(before & os.O_NONBLOCK, after & os.O_NONBLOCK)
+        self.assertEqual(before, after)
+
+    def test_default_timeout_does_not_alter_socket_flags(self):
+        """
+        Adopting an fd into a socket object sets O_NONBLOCK on it when a process-wide default timeout
+        is in effect. The server's blocking mode must be left exactly as found, whichever it was.
+        """
+        previous = socket.getdefaulttimeout()
+        self.addCleanup(socket.setdefaulttimeout, previous)
+
+        for blocking in (True, False):
+            with self.subTest(blocking=blocking):
+                server_sock, client_sock = self.make_socketpair()
+                server_sock.setblocking(blocking)
+                targets = self.make_targets()
+                socket.setdefaulttimeout(10)
+                try:
+                    self.watchdog.register(make_request(), server_sock.fileno(), targets)
+                finally:
+                    socket.setdefaulttimeout(previous)
+
+                self.assertEqual(os.get_blocking(server_sock.fileno()), blocking)
+
+                # The adopted socket must still peek without waiting.
+                client_sock.close()
+                self.assertTrue(targets[0].pgconn.cancelled.wait(5.0))
+
+    #
+    # Cancellation behaviour
+    #
+
+    def test_restart_shuts_down_previous_executor(self):
+        """Restarting after the thread died must not orphan the previous cancellation workers."""
+        old_executor = self.watchdog._executor
+        self.watchdog._stopping.set()
+        self.watchdog._wake()
+        self.watchdog._thread.join(5.0)
+        self.assertFalse(self.watchdog.is_alive())
+
+        self.watchdog.start()
+
+        self.assertTrue(self.watchdog.is_alive())
+        self.assertIsNot(self.watchdog._executor, old_executor)
+        with self.assertRaises(RuntimeError):
+            old_executor.submit(lambda: None)
+
+    def test_abandon_after_fork_closes_inherited_descriptors(self):
+        server_sock, client_sock = self.make_socketpair()
+        watchdog = ClientDisconnectWatchdog()
+        registration = watchdog.register(make_request(), server_sock.fileno(), self.make_targets())
+        fds = (watchdog._wake_r, watchdog._wake_w, registration.peek_sock.fileno())
+
+        watchdog.abandon_after_fork()
+
+        for fd in fds:
+            with self.assertRaises(OSError):
+                os.fstat(fd)
+        self.assertEqual(watchdog._by_token, {})
+        # The server's own socket is untouched.
+        os.fstat(server_sock.fileno())
+
+    def test_abandon_after_fork_ignores_inherited_lock(self):
+        """
+        A lock held by a parent thread at the moment of forking is never released in the child, so
+        abandoning the inherited watchdog must not wait for it.
+        """
+        watchdog = ClientDisconnectWatchdog()
+        held = threading.Event()
+        done = threading.Event()
+        self.addCleanup(done.set)
+
+        def hold_lock():
+            with watchdog._lock:
+                held.set()
+                done.wait(5.0)
+
+        threading.Thread(target=hold_lock, daemon=True).start()
+        self.assertTrue(held.wait(5.0))
+
+        abandoner = threading.Thread(target=watchdog.abandon_after_fork, daemon=True)
+        abandoner.start()
+        abandoner.join(2.0)
+        self.assertFalse(abandoner.is_alive())
+
+    def test_cancel_timeout_is_short(self):
+        """psycopg's 30-second default would tie up a cancellation worker far too long."""
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.close()
+
+        self.assertTrue(targets[0].pgconn.cancelled.wait(5.0))
+        self.assertTrue(self.wait_for(lambda: targets[0].pgconn.cancel_timeouts))
+        self.assertLessEqual(targets[0].pgconn.cancel_timeouts[0], CANCEL_TIMEOUT)
+
+    def test_cancels_all_registered_connections(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets('replica', 'default', 'archive')
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.close()
+
+        for target in targets:
+            self.assertTrue(target.pgconn.cancelled.wait(5.0), f"{target.alias} not cancelled")
+        self.assertTrue(self.wait_for(lambda: len(registration.cancelled_aliases) == 3))
+        # The default alias is cancelled first, so a slow secondary cannot exhaust the budget before
+        # the connection the request is most likely blocked on has been dealt with.
+        self.assertEqual(registration.cancelled_aliases[0], 'default')
+
+    def test_reconnected_wrapper_is_skipped(self):
+        """
+        A wrapper which reconnected since registration holds a different backend, so cancelling it
+        would interrupt an unrelated query. Its siblings must still be cancelled.
+        """
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets('default', 'replica')
+        stale = targets[0]
+        replacement = FakePgConn()
+        stale.wrapper.connection = replacement
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.close()
+
+        self.assertTrue(targets[1].pgconn.cancelled.wait(5.0))
+        self.assertTrue(self.wait_for(lambda: registration.state is RegistrationState.CANCELLED))
+        self.assertFalse(stale.pgconn.cancelled.is_set())
+        self.assertFalse(replacement.cancelled.is_set())
+        self.assertEqual(registration.cancelled_aliases, ('replica',))
+
+    def test_one_failing_cancel_does_not_block_siblings(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets('default', 'replica', fail_on=('default',))
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.close()
+
+        self.assertTrue(targets[1].pgconn.cancelled.wait(5.0))
+        self.assertTrue(self.wait_for(lambda: registration.state is RegistrationState.CANCELLED))
+        self.assertEqual(registration.cancelled_aliases, ('replica',))
+
+    def test_cancelled_alias_recorded_before_cancellation_returns(self):
+        """
+        The request thread can receive QueryCanceled, and log the abort, before cancel_safe() has
+        returned on the worker. The alias must already be recorded by then.
+        """
+        server_sock, client_sock = self.make_socketpair()
+        gate = threading.Event()
+        self.addCleanup(gate.set)
+        targets = self.make_targets('default', gate=gate)
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.close()
+
+        self.assertTrue(targets[0].pgconn.in_cancel.wait(5.0))
+        self.assertEqual(registration.cancelled_aliases, ('default',))
+
+    #
+    # Connection disposal
+    #
+
+    def test_hand_off_defers_disposal_until_cancellation_finishes(self):
+        """
+        Closing a connection while its cancellation is in flight races libpq, and with pooling would
+        hand the backend to another request before the cancellation lands. Connections handed off
+        mid-cancellation must be terminated by the cancellation worker, and only once it has finished.
+        """
+        server_sock, client_sock = self.make_socketpair()
+        gate = threading.Event()
+        self.addCleanup(gate.set)
+        targets = self.make_targets('default', gate=gate)
+        pgconn = targets[0].pgconn
+        pgconn._pool = Mock()
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.close()
+        self.assertTrue(pgconn.in_cancel.wait(5.0))
+        self.assertIs(self.watchdog.release(registration), RegistrationState.CANCELLING)
+        self.assertTrue(self.watchdog.hand_off(registration, targets))
+
+        time.sleep(0.1)
+        self.assertFalse(pgconn.closed)
+        pgconn._pool.putconn.assert_not_called()
+
+        gate.set()
+
+        self.assertTrue(self.wait_for(lambda: pgconn._pool.putconn.called))
+        self.assertTrue(pgconn.closed)
+        self.assertFalse(pgconn.closed_during_cancel)
+        # The pool must get the connection back, already terminated, so it can replace it.
+        pgconn._pool.putconn.assert_called_once_with(pgconn)
+
+    def test_hand_off_refused_once_cancellation_finished(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.close()
+        self.assertTrue(self.wait_for(lambda: registration.state is RegistrationState.CANCELLED))
+
+        self.assertFalse(self.watchdog.hand_off(registration, targets))
+        self.assertFalse(targets[0].pgconn.closed)
+
+    def test_hand_off_refused_when_not_cancelled(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        self.assertIs(self.watchdog.release(registration), RegistrationState.ARMED)
+        self.assertFalse(self.watchdog.hand_off(registration, targets))
+
+    def test_closed_connection_is_skipped(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        targets[0].pgconn.closed = True
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        client_sock.close()
+
+        self.assertTrue(self.wait_for(lambda: registration.state is RegistrationState.CANCELLED))
+        self.assertFalse(targets[0].pgconn.cancelled.is_set())
+
+    #
+    # Registration lifecycle
+    #
+
+    def test_release_before_disconnect_prevents_cancel(self):
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        observed = self.watchdog.release(registration)
+        client_sock.close()
+
+        self.assertIs(observed, RegistrationState.ARMED)
+        self.assertIs(registration.state, RegistrationState.RELEASED)
+        self.assert_not_cancelled(targets)
+
+    def test_release_reports_lost_race(self):
+        """
+        When the watchdog claims a registration at the same moment the request finishes, release()
+        must report that it lost, so the caller knows the connections may still be cancelled.
+        """
+        server_sock, client_sock = self.make_socketpair()
+        targets = self.make_targets()
+        proceed = threading.Event()
+        self.addCleanup(proceed.set)
+
+        original = self.watchdog._cancel_all
+
+        def blocking_cancel(registration):
+            proceed.wait(5.0)
+            original(registration)
+
+        with patch.object(self.watchdog, '_cancel_all', blocking_cancel):
+            registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+            client_sock.close()
+            self.assertTrue(self.wait_for(lambda: registration.state is RegistrationState.CANCELLING))
+
+            observed = self.watchdog.release(registration)
+
+        self.assertIs(observed, RegistrationState.CANCELLING)
+
+    def test_fd_reuse_does_not_cancel_stale_registration(self):
+        """
+        Descriptors are recycled integers. A released registration must never be reachable through a
+        descriptor number which has since been reissued to a different request.
+        """
+        first_server, first_client = self.make_socketpair()
+        second_server, second_client = self.make_socketpair()
+
+        first_targets = self.make_targets('default')
+        first = self.watchdog.register(make_request(), first_server.fileno(), first_targets)
+        first_dup_fd = first.peek_sock.fileno()
+
+        # Release the first registration, then immediately register the second, so that the second
+        # duplicate is allocated the descriptor number the first just gave up.
+        self.watchdog.release(first)
+        second_targets = self.make_targets('default')
+        second = self.watchdog.register(make_request(), second_server.fileno(), second_targets)
+        if second.peek_sock.fileno() != first_dup_fd:
+            self.skipTest("the kernel did not reissue the released descriptor")
+
+        second_client.close()
+
+        self.assertTrue(second_targets[0].pgconn.cancelled.wait(5.0))
+        self.assertFalse(first_targets[0].pgconn.cancelled.is_set())
+
+    def test_release_is_idempotent(self):
+        server_sock, _ = self.make_socketpair()
+        targets = self.make_targets()
+        registration = self.watchdog.register(make_request(), server_sock.fileno(), targets)
+
+        self.assertIs(self.watchdog.release(registration), RegistrationState.ARMED)
+        self.assertIs(self.watchdog.release(registration), RegistrationState.RELEASED)
+
+    def test_register_returns_none_without_targets(self):
+        server_sock, _ = self.make_socketpair()
+        self.assertIsNone(self.watchdog.register(make_request(), server_sock.fileno(), []))
+
+    def test_shutdown_is_prompt(self):
+        """The self-pipe must interrupt poll() rather than letting it wait out POLL_INTERVAL."""
+        started = time.monotonic()
+        self.watchdog.shutdown()
+        elapsed = time.monotonic() - started
+
+        self.assertLess(elapsed, 0.5)
+        self.assertFalse(self.watchdog.is_alive())
+
+    def test_no_fd_leak(self):
+        if not os.path.isdir('/proc/self/fd'):
+            self.skipTest("requires /proc")
+
+        def open_fds():
+            return len(os.listdir('/proc/self/fd'))
+
+        server_sock, _ = self.make_socketpair()
+        # Prime the loop so that one-off allocations are not counted as a leak.
+        self.watchdog.release(self.watchdog.register(make_request(), server_sock.fileno(), self.make_targets()))
+
+        before = open_fds()
+        for _ in range(100):
+            registration = self.watchdog.register(make_request(), server_sock.fileno(), self.make_targets())
+            self.watchdog.release(registration)
+
+        self.assertEqual(open_fds(), before)
+
+
+class GetClientFDTestCase(SimpleTestCase):
+
+    def test_gunicorn_socket(self):
+        server_sock, _ = socket.socketpair()
+        self.addCleanup(server_sock.close)
+        request = SimpleNamespace(META={'gunicorn.socket': server_sock})
+
+        self.assertEqual(get_client_fd(request), server_sock.fileno())
+
+    def test_uwsgi_connection_fd(self):
+        server_sock, _ = socket.socketpair()
+        self.addCleanup(server_sock.close)
+        request = SimpleNamespace(META={})
+        fake_uwsgi = SimpleNamespace(connection_fd=lambda: server_sock.fileno())
+
+        with patch.dict(sys.modules, {'uwsgi': fake_uwsgi}):
+            self.assertEqual(get_client_fd(request), server_sock.fileno())
+
+    def test_no_adapter_returns_none(self):
+        request = SimpleNamespace(META={})
+
+        # Ensure a real uwsgi module (if somehow importable) cannot influence the result.
+        with patch.dict(sys.modules, {'uwsgi': None}):
+            self.assertIsNone(get_client_fd(request))
+
+    def test_closed_gunicorn_socket_returns_none(self):
+        server_sock, _ = socket.socketpair()
+        server_sock.close()
+        request = SimpleNamespace(META={'gunicorn.socket': server_sock})
+
+        with patch.dict(sys.modules, {'uwsgi': None}):
+            self.assertIsNone(get_client_fd(request))
+            # The server is still supported: only this request's socket is unusable.
+            self.assertTrue(server_exposes_client_socket(request))
+
+    def test_server_exposes_client_socket(self):
+        with patch.dict(sys.modules, {'uwsgi': None}):
+            self.assertFalse(server_exposes_client_socket(SimpleNamespace(META={})))
+        with patch.dict(sys.modules, {'uwsgi': SimpleNamespace()}):
+            self.assertTrue(server_exposes_client_socket(SimpleNamespace(META={})))

netbox/netbox/tests/test_middleware.py

@@ -1,15 +1,22 @@
+import errno
 import json
+import time
+import uuid
 from contextlib import contextmanager
 from types import SimpleNamespace
-from unittest.mock import Mock, patch
+from unittest.mock import MagicMock, Mock, patch
 
+from django.core.exceptions import MiddlewareNotUsed
 from django.core.signals import got_request_exception
-from django.db.utils import InternalError
+from django.db.utils import InternalError, OperationalError
 from django.test import RequestFactory, override_settings
 from django.urls import reverse
+from prometheus_client import REGISTRY
+from psycopg import errors as psycopg_errors
 from rest_framework import status
 
-from netbox.middleware import CoreMiddleware, MaintenanceModeMiddleware
+from netbox.disconnect import CancelTarget, Registration, RegistrationState
+from netbox.middleware import ClientDisconnectMiddleware, CoreMiddleware, MaintenanceModeMiddleware
 from utilities.testing import TestCase
 
 
@@ -156,3 +163,461 @@ class CoreMiddlewareTestCase(TestCase):
             response = self.process_internal_error(request, 'Simulated maintenance mode GraphQL error')
 
         self.assert_json_500_response(response)
+
+
+class FakePgConn:
+    def __init__(self):
+        self.closed = False
+        self.info = SimpleNamespace(backend_pid=4242)
+        self.pgconn = SimpleNamespace(finish=self._finish)
+
+    def _finish(self):
+        self.closed = True
+
+    def cancel_safe(self, *, timeout=None):
+        pass
+
+
+class FakeWrapper:
+    def __init__(self, alias='default', in_atomic_block=False):
+        self.alias = alias
+        self.vendor = 'postgresql'
+        self.connection = FakePgConn()
+        self.in_atomic_block = in_atomic_block
+        self.closed = False
+        self.session_ended_before_close = None
+        self.rollback_set = None
+
+    def ensure_connection(self):
+        pass
+
+    def set_rollback(self, value):
+        self.rollback_set = value
+
+    def close(self):
+        self.session_ended_before_close = self.connection.closed
+        self.closed = True
+
+
+@override_settings(ABORT_ON_CLIENT_DISCONNECT=True)
+class ClientDisconnectMiddlewareTestCase(TestCase):
+    """
+    The watchdog itself is covered by netbox.tests.test_disconnect; these tests cover the middleware's
+    arming, response handling, and connection hygiene, with the watchdog and the client socket faked.
+    """
+
+    def setUp(self):
+        super().setUp()
+
+        self.factory = RequestFactory()
+
+    def build_request(self, method='get', path='/dcim/devices/'):
+        request = getattr(self.factory, method)(path)
+        request.id = uuid.uuid4()
+        return request
+
+    def build_registration(self, state=RegistrationState.ARMED, targets=None, release_state=None):
+        if targets is None:
+            targets = self.build_targets(FakeWrapper('default'))
+        registration = Registration(
+            token=1,
+            request_id='test-request',
+            method='GET',
+            path='/dcim/devices/',
+            started=time.monotonic(),
+            peek_sock=Mock(),
+            targets=tuple(targets),
+            watchdog=Mock(),
+        )
+        registration.state = state
+        registration.watchdog.release.return_value = release_state if release_state is not None else state
+        registration.watchdog.hand_off.return_value = False
+        return registration
+
+    def build_targets(self, *wrappers):
+        return [CancelTarget(w.alias, w, w.connection) for w in wrappers]
+
+    @contextmanager
+    def armed(self, registration, fd=7):
+        """Run the middleware with a resolvable client socket and a faked watchdog."""
+        watchdog = Mock()
+        watchdog.register.return_value = registration
+        targets = list(registration.targets)
+        with patch('netbox.middleware.get_client_fd', return_value=fd), \
+                patch('netbox.middleware.get_watchdog', return_value=watchdog), \
+                patch.object(ClientDisconnectMiddleware, '_get_cancel_targets', return_value=targets):
+            yield watchdog
+
+    @staticmethod
+    def cancellation_error(sqlstate='57014'):
+        """Build the OperationalError Django raises when a query is cancelled."""
+        cause = psycopg_errors.lookup(sqlstate)('canceling statement due to user request')
+        error = OperationalError('canceling statement due to user request')
+        error.__cause__ = cause
+        return error
+
+    #
+    # Gating
+    #
+
+    def test_middleware_not_used_when_disabled(self):
+        """
+        Disabled is the default, so it must cost nothing: MiddlewareNotUsed removes the middleware
+        from the chain outright rather than leaving an inert one in it.
+        """
+        with override_settings(ABORT_ON_CLIENT_DISCONNECT=False):
+            with self.assertRaises(MiddlewareNotUsed):
+                ClientDisconnectMiddleware(lambda request: None)
+
+    def test_disabled_without_socket(self):
+        """
+        A WSGI server which does not expose the client socket is a supported configuration, not an
+        error: the middleware must pass the request through and say so exactly once.
+        """
+        response = SimpleNamespace()
+        middleware = ClientDisconnectMiddleware(lambda request: response)
+
+        with patch('netbox.middleware.get_client_fd', return_value=None), \
+                patch('netbox.middleware.get_watchdog') as get_watchdog:
+            with self.assertLogs('netbox.disconnect', 'INFO') as logs:
+                self.assertIs(middleware(self.build_request()), response)
+            # A second request must not repeat the message: the disable is latched.
+            self.assertIs(middleware(self.build_request()), response)
+
+        self.assertEqual(len(logs.records), 1)
+        get_watchdog.assert_not_called()
+
+    def test_unusable_socket_does_not_latch_off(self):
+        """
+        A supported server can yield no fd for a single request, e.g. one whose socket has already
+        been closed. That request goes unwatched, but later requests must still be armed.
+        """
+        middleware = ClientDisconnectMiddleware(lambda request: SimpleNamespace())
+        request = self.build_request()
+        request.META['gunicorn.socket'] = Mock()
+
+        with patch('netbox.middleware.get_client_fd', return_value=None), \
+                patch('netbox.middleware.get_watchdog') as get_watchdog, \
+                self.assertNoLogs('netbox.disconnect', 'INFO'):
+            middleware(request)
+        get_watchdog.assert_not_called()
+
+        registration = self.build_registration()
+        with self.armed(registration) as watchdog:
+            middleware(self.build_request())
+        watchdog.register.assert_called_once()
+
+    #
+    # Arming
+    #
+
+    def test_safe_methods_are_armed(self):
+        for method in ('get', 'head', 'options'):
+            with self.subTest(method=method):
+                registration = self.build_registration()
+                with self.armed(registration) as watchdog:
+                    ClientDisconnectMiddleware(lambda request: SimpleNamespace())(self.build_request(method))
+                watchdog.register.assert_called_once()
+                registration.watchdog.release.assert_called_once_with(registration)
+
+    def test_unsafe_methods_not_armed(self):
+        """
+        Cancelling an abandoned write would silently roll back a change the client may believe has
+        landed, so unsafe methods always run to completion.
+        """
+        for method in ('post', 'put', 'patch', 'delete'):
+            with self.subTest(method=method):
+                registration = self.build_registration()
+                with self.armed(registration) as watchdog:
+                    request = self.build_request(method)
+                    middleware = ClientDisconnectMiddleware(lambda request: SimpleNamespace())
+                    middleware(request)
+                    watchdog.register.assert_not_called()
+
+                    # Even if something else cancelled the query, an unsafe method must not be
+                    # converted into a 499.
+                    self.assertIsNone(middleware.process_exception(request, self.cancellation_error()))
+
+    def test_watchdog_failure_does_not_fail_request(self):
+        """
+        Starting the watchdog can fail under resource exhaustion. That must cost the request its
+        protection, not its response, and must not latch the feature off for later requests.
+        """
+        failures = {
+            'start': RuntimeError("can't start new thread"),
+            'register': OSError(errno.EMFILE, 'Too many open files'),
+        }
+        for stage, exc in failures.items():
+            with self.subTest(stage=stage):
+                watchdog = Mock()
+                watchdog.register.side_effect = exc
+                get_watchdog = Mock(side_effect=exc) if stage == 'start' else Mock(return_value=watchdog)
+                targets = self.build_targets(FakeWrapper('default'))
+                response = SimpleNamespace()
+                middleware = ClientDisconnectMiddleware(lambda request: response)
+
+                with patch('netbox.middleware.get_client_fd', return_value=7), \
+                        patch('netbox.middleware.get_watchdog', get_watchdog), \
+                        patch.object(ClientDisconnectMiddleware, '_get_cancel_targets', return_value=targets), \
+                        self.assertLogs('netbox.disconnect', 'WARNING'):
+                    self.assertIs(middleware(self.build_request()), response)
+                    self.assertIs(middleware(self.build_request()), response)
+
+                self.assertEqual(get_watchdog.call_count, 2)
+
+    def test_all_open_connections_are_registered(self):
+        """Every PostgreSQL connection open for the request is cancelled, not only the default."""
+        default = FakeWrapper('default')
+        replica = FakeWrapper('replica')
+        other = FakeWrapper('mysql_thing')
+        other.vendor = 'mysql'
+
+        connections = MagicMock()
+        connections.__getitem__.return_value = default
+        connections.all.return_value = [replica, default, other]
+
+        with patch('netbox.middleware.connections', connections):
+            targets = ClientDisconnectMiddleware._get_cancel_targets()
+
+        connections.all.assert_called_once_with(initialized_only=True)
+        self.assertEqual({target.alias for target in targets}, {'default', 'replica'})
+
+    #
+    # Response handling
+    #
+
+    def test_disconnect_returns_499(self):
+        registration = self.build_registration(state=RegistrationState.CANCELLED)
+        registration.cancelled_aliases = ('default',)
+        request = self.build_request()
+        request._client_disconnect = registration
+        middleware = ClientDisconnectMiddleware(lambda request: None)
+
+        with self.assertLogs('netbox.disconnect', 'INFO'):
+            response = middleware.process_exception(request, self.cancellation_error())
+
+        self.assertEqual(response.status_code, 499)
+        self.assertEqual(response.reason_phrase, 'Client Closed Request')
+
+    def test_cancellation_without_our_flag_propagates(self):
+        """
+        An operator's statement_timeout produces the same SQLSTATE, so the SQLSTATE alone must not be
+        enough to claim the cancellation as ours.
+        """
+        registration = self.build_registration(state=RegistrationState.ARMED)
+        request = self.build_request()
+        request._client_disconnect = registration
+        middleware = ClientDisconnectMiddleware(lambda request: None)
+
+        self.assertIsNone(middleware.process_exception(request, self.cancellation_error()))
+
+    def test_cancellation_without_registration_propagates(self):
+        request = self.build_request()
+        middleware = ClientDisconnectMiddleware(lambda request: None)
+
+        self.assertIsNone(middleware.process_exception(request, self.cancellation_error()))
+
+    def test_other_operational_error_propagates(self):
+        registration = self.build_registration(state=RegistrationState.CANCELLED)
+        request = self.build_request()
+        request._client_disconnect = registration
+        middleware = ClientDisconnectMiddleware(lambda request: None)
+
+        self.assertIsNone(middleware.process_exception(request, self.cancellation_error('40001')))
+        self.assertIsNone(middleware.process_exception(request, InternalError('unrelated')))
+
+    def test_cancellation_does_not_fire_got_request_exception(self):
+        """
+        Returning a response here is what keeps the exception out of CoreMiddleware.process_exception
+        (which does fire the signal) and out of handler_500. This asserts the mechanism: nothing
+        escapes, and we send nothing ourselves.
+        """
+        registration = self.build_registration(state=RegistrationState.CANCELLED)
+        request = self.build_request()
+        request._client_disconnect = registration
+        middleware = ClientDisconnectMiddleware(lambda request: None)
+
+        captured = []
+        got_request_exception.connect(lambda sender, request, **kwargs: captured.append(request), weak=False)
+        receiver = got_request_exception.receivers[-1][1]
+        self.addCleanup(got_request_exception.disconnect, receiver)
+
+        with self.assertLogs('netbox.disconnect', 'INFO'):
+            response = middleware.process_exception(request, self.cancellation_error())
+
+        self.assertEqual(response.status_code, 499)
+        self.assertEqual(captured, [])
+
+    def test_view_label_falls_back_when_url_unresolved(self):
+        registration = self.build_registration(state=RegistrationState.CANCELLED)
+        request = self.build_request()
+        request._client_disconnect = registration
+        request.resolver_match = None
+        middleware = ClientDisconnectMiddleware(lambda request: None)
+
+        with patch('netbox.middleware.increment_client_disconnects') as increment:
+            with self.assertLogs('netbox.disconnect', 'INFO'):
+                middleware.process_exception(request, self.cancellation_error())
+
+        increment.assert_called_once_with(method='GET', view='<unnamed view>')
+
+    #
+    # Connection hygiene
+    #
+
+    def test_cleanup_closes_connections_when_cancelled(self):
+        wrappers = [FakeWrapper('default'), FakeWrapper('replica')]
+        registration = self.build_registration(
+            state=RegistrationState.CANCELLED,
+            targets=self.build_targets(*wrappers),
+            release_state=RegistrationState.CANCELLED,
+        )
+
+        with self.armed(registration):
+            ClientDisconnectMiddleware(lambda request: SimpleNamespace())(self.build_request())
+
+        for wrapper in wrappers:
+            self.assertTrue(wrapper.closed, f"{wrapper.alias} was not closed")
+
+    def test_cleanup_ends_session_before_close(self):
+        """
+        Under connection pooling, close() returns the connection to the pool. Its session must already
+        have ended, or a late cancellation could land on whichever request checks it out next.
+        """
+        wrapper = FakeWrapper('default')
+        registration = self.build_registration(
+            state=RegistrationState.CANCELLED,
+            targets=self.build_targets(wrapper),
+            release_state=RegistrationState.CANCELLED,
+        )
+
+        with self.armed(registration):
+            ClientDisconnectMiddleware(lambda request: SimpleNamespace())(self.build_request())
+
+        self.assertTrue(wrapper.closed)
+        self.assertTrue(wrapper.session_ended_before_close)
+
+    def test_cleanup_hands_off_while_cancellation_in_flight(self):
+        """
+        Connections must not be closed while a cancellation may still be using them. They are handed
+        to the cancellation worker instead, and detached from Django without being closed.
+        """
+        for in_atomic_block in (False, True):
+            with self.subTest(in_atomic_block=in_atomic_block):
+                wrapper = FakeWrapper('default', in_atomic_block=in_atomic_block)
+                pgconn = wrapper.connection
+                targets = self.build_targets(wrapper)
+                registration = self.build_registration(
+                    state=RegistrationState.CANCELLING,
+                    targets=targets,
+                    release_state=RegistrationState.CANCELLING,
+                )
+                registration.watchdog.hand_off.return_value = True
+
+                with self.armed(registration):
+                    ClientDisconnectMiddleware(lambda request: SimpleNamespace())(self.build_request())
+
+                registration.watchdog.hand_off.assert_called_once_with(registration, targets)
+                self.assertFalse(wrapper.closed)
+                self.assertFalse(pgconn.closed)
+                self.assertIsNone(wrapper.connection)
+                self.assertEqual(getattr(wrapper, 'closed_in_transaction', False), in_atomic_block)
+                self.assertEqual(getattr(wrapper, 'needs_rollback', False), in_atomic_block)
+
+    def test_no_cleanup_when_not_cancelled(self):
+        """The happy path must not close connections, or CONN_MAX_AGE reuse is defeated."""
+        wrapper = FakeWrapper('default')
+        registration = self.build_registration(
+            state=RegistrationState.ARMED,
+            targets=self.build_targets(wrapper),
+            release_state=RegistrationState.ARMED,
+        )
+
+        with self.armed(registration):
+            ClientDisconnectMiddleware(lambda request: SimpleNamespace())(self.build_request())
+
+        self.assertFalse(wrapper.closed)
+        self.assertIsNone(wrapper.rollback_set)
+
+    def test_set_rollback_only_inside_atomic_block(self):
+        """set_rollback() raises outside an atomic block, which would turn the 499 into a 500."""
+        for in_atomic_block in (False, True):
+            with self.subTest(in_atomic_block=in_atomic_block):
+                wrapper = FakeWrapper('default', in_atomic_block=in_atomic_block)
+                registration = self.build_registration(
+                    state=RegistrationState.CANCELLED,
+                    targets=self.build_targets(wrapper),
+                    release_state=RegistrationState.CANCELLED,
+                )
+
+                with self.armed(registration):
+                    ClientDisconnectMiddleware(lambda request: SimpleNamespace())(self.build_request())
+
+                self.assertEqual(wrapper.rollback_set, True if in_atomic_block else None)
+                self.assertTrue(wrapper.closed)
+
+    def test_reconnected_wrapper_is_not_closed(self):
+        wrapper = FakeWrapper('default')
+        targets = self.build_targets(wrapper)
+        wrapper.connection = FakePgConn()  # reconnected since registration
+        registration = self.build_registration(
+            state=RegistrationState.CANCELLED,
+            targets=targets,
+            release_state=RegistrationState.CANCELLED,
+        )
+
+        with self.armed(registration):
+            ClientDisconnectMiddleware(lambda request: SimpleNamespace())(self.build_request())
+
+        self.assertFalse(wrapper.closed)
+
+    def test_registration_released_when_view_raises(self):
+        """Release lives in a finally: a leaked registration would poison the next request."""
+        registration = self.build_registration()
+
+        def get_response(request):
+            raise RuntimeError('boom')
+
+        with self.armed(registration):
+            with self.assertRaises(RuntimeError):
+                ClientDisconnectMiddleware(get_response)(self.build_request())
+
+        registration.watchdog.release.assert_called_once_with(registration)
+
+    #
+    # Metrics
+    #
+
+    @override_settings(METRICS_ENABLED=True)
+    def test_metric_incremented_on_disconnect(self):
+        registration = self.build_registration(state=RegistrationState.CANCELLED)
+        request = self.build_request()
+        request._client_disconnect = registration
+        request.resolver_match = SimpleNamespace(view_name='dcim:device_list')
+        middleware = ClientDisconnectMiddleware(lambda request: None)
+
+        labels = {'method': 'GET', 'view': 'dcim:device_list'}
+        # Counters live in a process-global registry, so only the delta is meaningful.
+        before = REGISTRY.get_sample_value('netbox_client_disconnects_total', labels) or 0
+
+        with self.assertLogs('netbox.disconnect', 'INFO'):
+            middleware.process_exception(request, self.cancellation_error())
+
+        after = REGISTRY.get_sample_value('netbox_client_disconnects_total', labels) or 0
+        self.assertEqual(after - before, 1)
+
+    def test_metric_not_incremented_when_metrics_disabled(self):
+        """
+        Touching the singleton would register the whole django_prometheus metric set on installations
+        which never expose /metrics.
+        """
+        registration = self.build_registration(state=RegistrationState.CANCELLED)
+        request = self.build_request()
+        request._client_disconnect = registration
+        middleware = ClientDisconnectMiddleware(lambda request: None)
+
+        with patch('netbox.metrics.Metrics.get_instance') as get_instance:
+            with self.assertLogs('netbox.disconnect', 'INFO'):
+                middleware.process_exception(request, self.cancellation_error())
+
+        get_instance.assert_not_called()