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,45 @@
 # 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 rather than being returned to a persistent connection pool.
+
+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.
+    * GraphQL requests are protected in that their queries are still cancelled and their connections still discarded, but 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,550 @@
+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',
+)
+
+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)
+
+# 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}>'
+
+
+@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
+    cancelled_aliases: tuple = ()
+    saw_pipelined_data: bool = False
+
+    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
+
+
+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:
+            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
+
+        # 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
+
+    #
+    # Thread lifecycle
+    #
+
+    def start(self):
+        with self._lock:
+            if self._thread is not None and self._thread.is_alive():
+                return
+            self._stopping.clear()
+            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.
+        """
+        with self._lock:
+            registrations = list(self._by_token.values())
+            for registration in registrations:
+                self._discard(registration)
+            self._thread = None
+            self._executor = None
+        for registration in registrations:
+            self._close(registration)
+
+    #
+    # 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)
+        try:
+            fd = registration.peek_sock.fileno()
+        except OSError:
+            fd = -1
+        if fd >= 0:
+            self._by_fd.pop(fd, None)
+            try:
+                self._poll.unregister(fd)
+            except (KeyError, OSError):
+                pass
+
+    @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)
+                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:
+            # 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_pipelined_data:
+            # The peer has pipelined a further request, so this fd is permanently readable-with-data
+            # and can tell us nothing more.
+            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)
+        else:
+            registration.saw_pipelined_data = True
+
+    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
+        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):
+        cancelled = []
+        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
+                    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.
+                    logger.warning(
+                        "Failed to cancel query for request %s on database '%s': %s",
+                        registration.request_id, target.alias, exc,
+                    )
+                else:
+                    cancelled.append(target.alias)
+        finally:
+            with self._lock:
+                registration.cancelled_aliases = tuple(cancelled)
+                if registration.state is RegistrationState.CANCELLING:
+                    registration.state = RegistrationState.CANCELLED
+
+    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
+            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,34 @@ 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,
+)
+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 +41,8 @@ __all__ = (
     'SocialAuthExceptionMiddleware',
 )
 
+disconnect_logger = logging.getLogger('netbox.disconnect')
+
 
 class CommonMiddleware(DjangoCommonMiddleware):
     """
@@ -134,6 +146,181 @@ 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:
+            # 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
+
+        return get_watchdog().register(request, fd, targets)
+
+    @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
+
+        for target in registration.targets:
+            try:
+                pgconn = target.wrapper.connection
+                if pgconn is None or pgconn is not target.pgconn:
+                    # Already closed, or replaced by a reconnect; not ours to discard.
+                    continue
+                # 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)
+                # Close unconditionally rather than deferring to close_if_unusable_or_obsolete(): a
+                # cancellation which found nothing to cancel leaves a perfectly usable connection,
+                # but we cannot distinguish that from one which is about to land. Closing terminates
+                # the backend, so any late cancellation becomes a no-op instead of interrupting an
+                # unrelated query on a connection reused via CONN_MAX_AGE.
+                target.wrapper.close()
+            except Exception:
+                disconnect_logger.exception(
+                    "Error discarding database connection '%s' after cancellation", target.alias
+                )
+
+    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,370 @@
+import fcntl
+import os
+import socket
+import struct
+import sys
+import threading
+import time
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from django.test import SimpleTestCase
+
+from netbox.disconnect import (
+    CANCEL_TIMEOUT,
+    CancelTarget,
+    ClientDisconnectWatchdog,
+    RegistrationState,
+    get_client_fd,
+)
+
+
+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):
+        self.closed = False
+        self.fail = fail
+        self.cancelled = threading.Event()
+        self.cancel_timeouts = []
+        self.info = SimpleNamespace(backend_pid=12345)
+
+    def cancel_safe(self, *, timeout=None):
+        self.cancel_timeouts.append(timeout)
+        self.cancelled.set()
+        if self.fail:
+            raise RuntimeError("simulated cancellation failure")
+
+
+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=()):
+        targets = []
+        for alias in (aliases or ('default',)):
+            wrapper = FakeWrapper(alias=alias, pgconn=FakePgConn(fail=alias in fail_on))
+            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_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)
+
+    #
+    # Cancellation behaviour
+    #
+
+    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_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))

netbox/netbox/tests/test_middleware.py

@@ -1,15 +1,21 @@
 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 +162,362 @@ 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)
+
+    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.rollback_set = None
+
+    def ensure_connection(self):
+        pass
+
+    def set_rollback(self, value):
+        self.rollback_set = value
+
+    def close(self):
+        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
+        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()
+
+    #
+    # 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_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_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()