diff --git a/ci/unit/test_nlt_node.sh b/ci/unit/test_nlt_node.sh
index 77099b5bd2c..2b7725b67a8 100755
--- a/ci/unit/test_nlt_node.sh
+++ b/ci/unit/test_nlt_node.sh
@@ -51,7 +51,9 @@ mkdir -p nlt_logs
 sudo mount -t tmpfs tmpfs nlt_logs
 sudo chown jenkins:jenkins nlt_logs
 
+# Unbuffered so the console shows exactly where a hang occurs.
 TMPDIR="$(pwd)/nlt_logs" \
+    PYTHONUNBUFFERED=1 \
     HTTPS_PROXY="${DAOS_HTTPS_PROXY:-}" \
     NO_PROXY="${DAOS_NO_PROXY:-}" \
     exec ./utils/node_local_test.py "$@"
diff --git a/src/client/dfs/duns.c b/src/client/dfs/duns.c
index 8b8ae332da2..a710a6ab5f4 100644
--- a/src/client/dfs/duns.c
+++ b/src/client/dfs/duns.c
@@ -1,5 +1,6 @@
 /**
  * (C) Copyright 2019-2024 Intel Corporation.
+ * (C) Copyright 2026 Hewlett Packard Enterprise Development LP
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -899,6 +900,69 @@ duns_link_lustre_path(const char *pool, const char *cont, daos_cont_layout_t typ
 }
 #endif
 
+/* Bounds queued-notification delivery latency, normally microseconds; does not wait out
+ * dentry expiry on mounts with long timeouts.
+ */
+#define DUNS_RESOLVE_TIMEOUT_MS     10000
+#define DUNS_RESOLVE_BACKOFF_MS_MAX 50
+
+/* Poll until dfuse binds the entry point to the new container, or we
+ * hit the timeout.
+ */
+static int
+duns_wait_for_resolution(const char *path, uuid_t cont_uuid)
+{
+	struct timespec start, now;
+	int             backoff_ms = 1;
+	int             rc;
+
+	clock_gettime(CLOCK_MONOTONIC, &start);
+
+	while (1) {
+		struct dfuse_il_reply il_reply = {};
+		int                   fd;
+		int64_t               elapsed_ms;
+
+		fd = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW);
+		if (fd == -1) {
+			rc = errno;
+			/* ENOLINK: dfuse attempted entry-point resolution and the container
+			 * connect failed - expected during rebinding, and terminal here since a
+			 * failed connect will not heal within the poll window.
+			 */
+			if (rc != ENOLINK)
+				D_ERROR("Failed to open %s to verify resolution: %d (%s)\n", path,
+					rc, strerror(rc));
+			return rc;
+		}
+
+		rc = ioctl(fd, DFUSE_IOCTL_IL, &il_reply);
+		close(fd);
+		if (rc == -1) {
+			rc = errno;
+			D_ERROR("Dfuse IL ioctl failed on %s: %d (%s)\n", path, rc, strerror(rc));
+			return rc;
+		}
+
+		if (uuid_compare(il_reply.fir_cont, cont_uuid) == 0)
+			return 0;
+
+		clock_gettime(CLOCK_MONOTONIC, &now);
+		elapsed_ms =
+		    (now.tv_sec - start.tv_sec) * 1000 + (now.tv_nsec - start.tv_nsec) / 1000000;
+		if (elapsed_ms >= DUNS_RESOLVE_TIMEOUT_MS) {
+			D_ERROR("Entry point %s still bound to container " DF_UUIDF " after %dms, "
+				"expected " DF_UUIDF "\n",
+				path, DP_UUID(il_reply.fir_cont), DUNS_RESOLVE_TIMEOUT_MS,
+				DP_UUID(cont_uuid));
+			return ENOLINK;
+		}
+
+		usleep(backoff_ms * 1000);
+		backoff_ms = min(backoff_ms * 2, DUNS_RESOLVE_BACKOFF_MS_MAX);
+	}
+}
+
 int
 duns_create_path(daos_handle_t poh, const char *path, struct duns_attr_t *attrp)
 {
@@ -1099,20 +1163,12 @@ duns_create_path(daos_handle_t poh, const char *path, struct duns_attr_t *attrp)
 		goto err_cont;
 	}
 	if (backend_dfuse) {
-		struct stat finfo;
-		/*
-		 * This next stat will cause dfuse to lookup the entry point and perform a
-		 * container connect, therefore this data will be read from root of the new
-		 * container, not the directory.
-		 *
-		 * TODO: This could call getxattr to verify success.
+		/* Confirm dfuse has looked up the entry point and connected to the new
+		 * container.
 		 */
-		rc = stat(path, &finfo);
-		if (rc) {
-			rc = errno;
-			D_ERROR("Failed to access new container: %d (%s)\n", rc, strerror(rc));
-			goto err_link;
-		}
+		rc = duns_wait_for_resolution(path, attrp->da_cuuid);
+		if (rc)
+			goto err_verify;
 	}
 
 	return rc;
@@ -1126,6 +1182,13 @@ duns_create_path(daos_handle_t poh, const char *path, struct duns_attr_t *attrp)
 	else if (attrp->da_type != DAOS_PROP_CO_LAYOUT_UNKNOWN)
 		unlink(path);
 	return rc;
+err_verify:
+	/* clean up the path before destroying the linked container */
+	rmdir(path);
+	rc2 = daos_cont_destroy(poh, attrp->da_cont, 1, NULL);
+	if (rc2)
+		D_ERROR("Failed to cleanup created container %s (%d)\n", attrp->da_cont, rc2);
+	return rc;
 }
 
 int
diff --git a/src/client/dfuse/SConscript b/src/client/dfuse/SConscript
index c699dd2acbb..94eea938c69 100644
--- a/src/client/dfuse/SConscript
+++ b/src/client/dfuse/SConscript
@@ -240,6 +240,9 @@ def scons():
 
     cenv.Install(os.path.join("$PREFIX", 'bin'), dfuse_bin)
 
+    if prereqs.test_requested():
+        SConscript('tests/SConscript', exports={'denv': cenv})
+
 
 if __name__ == "SCons.Script":
     scons()
diff --git a/src/client/dfuse/dfuse.h b/src/client/dfuse/dfuse.h
index ab14572fa1f..5fdb1b77480 100644
--- a/src/client/dfuse/dfuse.h
+++ b/src/client/dfuse/dfuse.h
@@ -13,6 +13,11 @@
 #include <fused/fuse.h>
 #include <fused/fuse_lowlevel.h>
 
+/* Only inval.c may call these directly. */
+#ifndef DFUSE_NOTIFY_RAW_OK
+#pragma GCC poison fuse_lowlevel_notify_inval_entry fuse_lowlevel_notify_expire_entry fuse_lowlevel_notify_delete fuse_lowlevel_notify_inval_inode
+#endif
+
 #include <gurt/list.h>
 #include <gurt/hash.h>
 #include <gurt/atomic.h>
@@ -76,6 +81,11 @@ struct dfuse_info {
 	ATOMIC uint64_t      di_fh_count;
 	ATOMIC uint64_t      di_pool_count;
 	ATOMIC uint64_t      di_container_count;
+
+	ATOMIC uint64_t      di_notify_enqueued;
+	ATOMIC uint64_t      di_notify_coalesced;
+	ATOMIC uint64_t      di_notify_delivered;
+	ATOMIC uint64_t      di_notify_dropped;
 };
 
 struct dfuse_eq {
@@ -1201,6 +1211,17 @@ ival_thread_stop();
 void
 ival_fini();
 
+/* Fire-and-forget reverse notifications, delivered by the notify thread in inval.c */
+void
+dfuse_notify_inval_entry(struct dfuse_info *dfuse_info, fuse_ino_t parent, const char *name);
+
+void
+dfuse_notify_delete(struct dfuse_info *dfuse_info, fuse_ino_t parent, fuse_ino_t ino,
+		    const char *name);
+
+void
+dfuse_notify_inval_inode(struct dfuse_info *dfuse_info, fuse_ino_t ino);
+
 /* Data caching functions */
 
 /* Mark the data cache as up-to-date from now */
diff --git a/src/client/dfuse/dfuse_main.c b/src/client/dfuse/dfuse_main.c
index 872ae186dbe..bfa56b08567 100644
--- a/src/client/dfuse/dfuse_main.c
+++ b/src/client/dfuse/dfuse_main.c
@@ -1,6 +1,6 @@
 /**
  * (C) Copyright 2016-2024 Intel Corporation.
- * (C) Copyright 2025 Hewlett Packard Enterprise Development LP
+ * (C) Copyright 2025-2026 Hewlett Packard Enterprise Development LP
  * (C) Copyright 2025 Google LLC
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
@@ -977,6 +977,15 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (dfuse_info)
+		DFUSE_TRA_INFO(dfuse_info,
+			       "Notify: enqueued=" DF_U64 " coalesced=" DF_U64 " delivered=" DF_U64
+			       " dropped=" DF_U64,
+			       atomic_load_relaxed(&dfuse_info->di_notify_enqueued),
+			       atomic_load_relaxed(&dfuse_info->di_notify_coalesced),
+			       atomic_load_relaxed(&dfuse_info->di_notify_delivered),
+			       atomic_load_relaxed(&dfuse_info->di_notify_dropped));
+
 	DFUSE_TRA_DOWN(dfuse_info);
 	daos_fini();
 out_debug:
diff --git a/src/client/dfuse/inval.c b/src/client/dfuse/inval.c
index 3ddcc052d86..8f7c5c19e28 100644
--- a/src/client/dfuse/inval.c
+++ b/src/client/dfuse/inval.c
@@ -1,6 +1,7 @@
 /**
  * (C) Copyright 2016-2024 Intel Corporation.
  * (C) Copyright 2025 Google LLC
+ * (C) Copyright 2026 Hewlett Packard Enterprise Development LP
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -10,6 +11,8 @@
 #include <math.h>
 
 #include "dfuse_common.h"
+/* Allow this file to call the raw fuse_lowlevel_notify_* symbols. */
+#define DFUSE_NOTIFY_RAW_OK
 #include "dfuse.h"
 
 /* Evict inodes based on timeout.
@@ -92,7 +95,7 @@ struct dfuse_time_entry {
 struct dfuse_ival {
 	d_list_t             time_entry_list;
 	struct fuse_session *session;
-	bool                 session_dead;
+	ATOMIC bool          session_dead;
 };
 
 /* The core data from struct dfuse_inode_entry.  No additional inode references are held on inodes
@@ -180,7 +183,7 @@ ival_loop(int *sleep_time)
 	DFUSE_TRA_DEBUG(&ival_data, "Unlocking, allowing to sleep for %d seconds", *sleep_time);
 	D_MUTEX_UNLOCK(&ival_lock);
 
-	if (idx == 0 || ival_data.session_dead)
+	if (idx == 0 || atomic_load_relaxed(&ival_data.session_dead))
 		return false;
 
 	for (int i = 0; i < idx; i++) {
@@ -194,7 +197,7 @@ ival_loop(int *sleep_time)
 		if (rc && rc != -ENOENT && rc != -EBADF)
 			DHS_ERROR(&ival_data, -rc, "notify_inval_entry() failed");
 		if (rc == -EBADF)
-			ival_data.session_dead = true;
+			atomic_store_relaxed(&ival_data.session_dead, true);
 	}
 
 	return (idx == EVICT_COUNT);
@@ -236,6 +239,315 @@ ival_thread_fn(void *arg)
 	return NULL;
 }
 
+/* Advisory reverse notifications.
+ *
+ * The fuse_lowlevel_notify_*() calls block in the kernel on inode locks that in-flight requests
+ * hold while waiting on dfuse, so request handlers must never issue one directly; they enqueue
+ * here and a dedicated thread delivers.  Nothing waits on delivery: a stalled or dropped entry
+ * leaves a dentry cached until its normal timeout, the same as a failed kernel call.  Duplicate
+ * queued entries coalesce; an entry leaves the coalesce index when popped, before delivery, so
+ * an enqueue racing delivery is never absorbed into a stale notification.
+ */
+enum notify_kind {
+	NOTIFY_INVAL_ENTRY,
+	NOTIFY_DELETE,
+	NOTIFY_INVAL_INODE,
+};
+
+/* Queued by copy; the source inode may be released once enqueued */
+struct notify_entry {
+	d_list_t         ne_list;
+	d_list_t         ne_hlist; /* coalesce bucket, valid only while queued */
+	enum notify_kind ne_kind;
+	fuse_ino_t       ne_parent;
+	fuse_ino_t       ne_ino;
+	char             ne_name[NAME_MAX + 1];
+	size_t           ne_namelen;
+};
+
+/* Drop rather than grow without bound; invalidation is advisory */
+#define NOTIFY_QUEUE_MAX        16384
+
+/* Depth at which delivery is judged stalled rather than merely busy */
+#define NOTIFY_CONGEST_HIGH     (NOTIFY_QUEUE_MAX / 2)
+
+#define NOTIFY_COALESCE_BUCKETS 4096
+
+static pthread_mutex_t     notify_lock = PTHREAD_MUTEX_INITIALIZER;
+static d_list_t            notify_queue;
+static d_list_t            notify_coalesce[NOTIFY_COALESCE_BUCKETS];
+static sem_t               notify_sem;
+static bool                notify_stop;
+static unsigned int        notify_queued;
+static pthread_t           notify_thread;
+static struct d_slab_type *notify_slab;
+
+/* Count of enqueues seen at/above NOTIFY_CONGEST_HIGH; gates the congestion warning rate.
+ * Under notify_lock.
+ */
+static uint64_t            notify_congest_total;
+
+/* Trace identity; carries dfuse_info for the teardown paths, which take no arguments */
+static struct dfuse_notify {
+	struct dfuse_info *dn_info;
+} notify_data;
+
+/* FNV-1a over the coalesce key; 64-bit inos are folded as two 32-bit halves */
+static uint32_t
+notify_hash(enum notify_kind kind, fuse_ino_t parent, fuse_ino_t ino, const char *name,
+	    size_t namelen)
+{
+	uint32_t h = 2166136261u;
+	size_t   i;
+
+	h = (h ^ (uint32_t)kind) * 16777619u;
+	h = (h ^ (uint32_t)parent) * 16777619u;
+	h = (h ^ (uint32_t)(parent >> 32)) * 16777619u;
+	h = (h ^ (uint32_t)ino) * 16777619u;
+	h = (h ^ (uint32_t)(ino >> 32)) * 16777619u;
+	for (i = 0; i < namelen; i++)
+		h = (h ^ (unsigned char)name[i]) * 16777619u;
+
+	return h % NOTIFY_COALESCE_BUCKETS;
+}
+
+/* Must be called with notify_lock held. */
+static struct notify_entry *
+notify_coalesce_find(enum notify_kind kind, fuse_ino_t parent, fuse_ino_t ino, const char *name,
+		     size_t namelen, uint32_t bucket)
+{
+	struct notify_entry *ne;
+
+	d_list_for_each_entry(ne, &notify_coalesce[bucket], ne_hlist) {
+		if (ne->ne_kind != kind || ne->ne_parent != parent || ne->ne_ino != ino)
+			continue;
+		if (ne->ne_namelen != namelen)
+			continue;
+		if (namelen != 0 && memcmp(ne->ne_name, name, namelen) != 0)
+			continue;
+		return ne;
+	}
+
+	return NULL;
+}
+
+/* Must be called with notify_lock held. */
+static struct notify_entry *
+notify_dequeue_locked(void)
+{
+	struct notify_entry *ne;
+
+	if (d_list_empty(&notify_queue))
+		return NULL;
+
+	ne = d_list_entry(notify_queue.next, struct notify_entry, ne_list);
+	d_list_del(&ne->ne_list);
+	d_list_del(&ne->ne_hlist);
+	notify_queued--;
+
+	return ne;
+}
+
+/* Discard everything queued, at stop or after EBADF.  Must be called with notify_lock held. */
+static void
+notify_drain_locked(void)
+{
+	struct notify_entry *ne, *nep;
+	unsigned int         count = 0;
+
+	d_list_for_each_entry_safe(ne, nep, &notify_queue, ne_list) {
+		d_list_del(&ne->ne_list);
+		d_list_del(&ne->ne_hlist);
+		d_slab_release(notify_slab, ne);
+		count++;
+	}
+	notify_queued = 0;
+
+	if (count == 0)
+		return;
+
+	atomic_fetch_add_relaxed(&notify_data.dn_info->di_notify_dropped, count);
+	DFUSE_TRA_DEBUG(&notify_data, "Drained %u queued notifications", count);
+}
+
+/* Rate-limited: a mass drop must not flood the log */
+static void
+notify_drop(struct dfuse_info *dfuse_info, enum notify_kind kind, fuse_ino_t parent)
+{
+	uint64_t total;
+
+	total = atomic_fetch_add_relaxed(&dfuse_info->di_notify_dropped, 1) + 1;
+
+	if (total == 1 || (total % 1000) == 0)
+		DFUSE_TRA_WARNING(&notify_data,
+				  "Dropped advisory notify kind %d parent %#lx, %lu total", kind,
+				  parent, total);
+}
+
+/* Rate-limited: sustained congestion must not flood the log.  Called with notify_lock held. */
+static void
+notify_congest_warn(unsigned int depth)
+{
+	notify_congest_total++;
+
+	if (notify_congest_total == 1 || (notify_congest_total % 4096) == 0)
+		DFUSE_TRA_WARNING(&notify_data, "Notify queue congested: depth %u", depth);
+}
+
+/* Blocking here is expected; only the notify thread runs this */
+static void
+notify_run(struct notify_entry *ne)
+{
+	int rc = 0;
+
+	switch (ne->ne_kind) {
+	case NOTIFY_INVAL_ENTRY:
+		rc = fuse_lowlevel_notify_inval_entry(ival_data.session, ne->ne_parent, ne->ne_name,
+						      ne->ne_namelen);
+		break;
+	case NOTIFY_DELETE:
+		rc = fuse_lowlevel_notify_delete(ival_data.session, ne->ne_parent, ne->ne_ino,
+						 ne->ne_name, ne->ne_namelen);
+		break;
+	case NOTIFY_INVAL_INODE:
+		rc = fuse_lowlevel_notify_inval_inode(ival_data.session, ne->ne_ino, 0, 0);
+		break;
+	}
+
+	/* Session is gone; latch (shared with ival_loop) and let the loop drain */
+	if (rc == -EBADF) {
+		atomic_store_relaxed(&ival_data.session_dead, true);
+		atomic_fetch_add_relaxed(&notify_data.dn_info->di_notify_dropped, 1);
+		return;
+	}
+
+	/* ENOENT/ENOTDIR just mean nothing was cached; any other error is unexpected and does
+	 * not count as delivered.
+	 */
+	if (rc != 0 && rc != -ENOENT && rc != -ENOTDIR) {
+		DHS_ERROR(&notify_data, -rc, "notify() failed, kind %d parent %#lx", ne->ne_kind,
+			  ne->ne_parent);
+		return;
+	}
+
+	atomic_fetch_add_relaxed(&notify_data.dn_info->di_notify_delivered, 1);
+}
+
+static void *
+notify_thread_fn(void *arg)
+{
+	while (1) {
+		struct notify_entry *ne = NULL;
+		bool                 stop;
+		bool                 dead;
+
+		if (sem_wait(&notify_sem) != 0) {
+			D_ASSERTF(errno == EINTR, "sem_wait: %d (%s)\n", errno, strerror(errno));
+			continue;
+		}
+
+		D_MUTEX_LOCK(&notify_lock);
+		stop = notify_stop;
+		dead = atomic_load_relaxed(&ival_data.session_dead);
+		if (stop || dead)
+			notify_drain_locked();
+		else
+			ne = notify_dequeue_locked();
+		D_MUTEX_UNLOCK(&notify_lock);
+
+		if (stop)
+			return NULL;
+
+		if (ne == NULL)
+			continue;
+
+		notify_run(ne);
+		d_slab_release(notify_slab, ne);
+	}
+	return NULL;
+}
+
+/* Best effort; failures result in dropped notifications. */
+static void
+notify_enqueue(struct dfuse_info *dfuse_info, enum notify_kind kind, fuse_ino_t parent,
+	       fuse_ino_t ino, const char *name)
+{
+	struct notify_entry *ne;
+	size_t               namelen = name ? strnlen(name, NAME_MAX) : 0;
+	uint32_t             bucket  = notify_hash(kind, parent, ino, name, namelen);
+
+	D_MUTEX_LOCK(&notify_lock);
+
+	if (notify_stop) {
+		D_MUTEX_UNLOCK(&notify_lock);
+		notify_drop(dfuse_info, kind, parent);
+		return;
+	}
+
+	if (notify_coalesce_find(kind, parent, ino, name, namelen, bucket) != NULL) {
+		D_MUTEX_UNLOCK(&notify_lock);
+		atomic_fetch_add_relaxed(&dfuse_info->di_notify_coalesced, 1);
+		DFUSE_TRA_DEBUG(&notify_data, "Coalesced kind %d parent %#lx " DF_DE, kind, parent,
+				DP_DE(name ? name : ""));
+		return;
+	}
+
+	if (notify_queued >= NOTIFY_QUEUE_MAX) {
+		D_MUTEX_UNLOCK(&notify_lock);
+		notify_drop(dfuse_info, kind, parent);
+		return;
+	}
+
+	ne = d_slab_acquire(notify_slab);
+	if (ne == NULL) {
+		D_MUTEX_UNLOCK(&notify_lock);
+		notify_drop(dfuse_info, kind, parent);
+		return;
+	}
+
+	ne->ne_kind    = kind;
+	ne->ne_parent  = parent;
+	ne->ne_ino     = ino;
+	ne->ne_namelen = namelen;
+	if (namelen > 0)
+		memcpy(ne->ne_name, name, namelen);
+	ne->ne_name[namelen] = '\0';
+
+	d_list_add_tail(&ne->ne_hlist, &notify_coalesce[bucket]);
+	d_list_add_tail(&ne->ne_list, &notify_queue);
+	notify_queued++;
+
+	if (notify_queued >= NOTIFY_CONGEST_HIGH)
+		notify_congest_warn(notify_queued);
+
+	D_MUTEX_UNLOCK(&notify_lock);
+
+	atomic_fetch_add_relaxed(&dfuse_info->di_notify_enqueued, 1);
+	DFUSE_TRA_DEBUG(&notify_data, "Enqueued kind %d parent %#lx " DF_DE, kind, parent,
+			DP_DE(name ? name : ""));
+
+	sem_post(&notify_sem);
+}
+
+void
+dfuse_notify_inval_entry(struct dfuse_info *dfuse_info, fuse_ino_t parent, const char *name)
+{
+	notify_enqueue(dfuse_info, NOTIFY_INVAL_ENTRY, parent, 0, name);
+}
+
+void
+dfuse_notify_delete(struct dfuse_info *dfuse_info, fuse_ino_t parent, fuse_ino_t ino,
+		    const char *name)
+{
+	notify_enqueue(dfuse_info, NOTIFY_DELETE, parent, ino, name);
+}
+
+void
+dfuse_notify_inval_inode(struct dfuse_info *dfuse_info, fuse_ino_t ino)
+{
+	notify_enqueue(dfuse_info, NOTIFY_INVAL_INODE, 0, ino, NULL);
+}
+
 /* Allocate and insert a new time value entry */
 static int
 ival_bucket_add(d_list_t *list, double timeout)
@@ -263,28 +575,55 @@ int
 ival_init(struct dfuse_info *dfuse_info)
 {
 	int rc;
+	int i;
 
 	DFUSE_TRA_UP(&ival_data, dfuse_info, "invalidator");
+	DFUSE_TRA_UP(&notify_data, dfuse_info, "notify");
 
 	D_INIT_LIST_HEAD(&ival_data.time_entry_list);
 
+	notify_data.dn_info = dfuse_info;
+	D_INIT_LIST_HEAD(&notify_queue);
+	for (i = 0; i < NOTIFY_COALESCE_BUCKETS; i++)
+		D_INIT_LIST_HEAD(&notify_coalesce[i]);
+
 	rc = sem_init(&ival_sem, 0, 0);
 	if (rc != 0)
 		D_GOTO(out, rc = errno);
 
+	rc = sem_init(&notify_sem, 0, 0);
+	if (rc != 0) {
+		rc = errno;
+		goto ival_sem;
+	}
+
 	rc = ival_bucket_add(&ival_data.time_entry_list, 0);
 	if (rc)
-		goto sem;
+		goto notify_sem;
 
 out:
 	return rc;
-sem:
+notify_sem:
+	sem_destroy(&notify_sem);
+ival_sem:
 	sem_destroy(&ival_sem);
+	DFUSE_TRA_DOWN(&notify_data);
 	DFUSE_TRA_DOWN(&ival_data);
 	return rc;
 }
 
-/* Start the thread.  Not called until after fuse is mounted */
+/* Register the notify entry slab type.  Split out of ival_thread_start() so tests can prepare
+ * the queue for notify_enqueue()/notify_dequeue_locked() without starting the delivery thread.
+ */
+static int
+notify_init(struct dfuse_info *dfuse_info)
+{
+	struct d_slab_reg notify_slab_reg = {POOL_TYPE_INIT(notify_entry, ne_list)};
+
+	return d_slab_register(&dfuse_info->di_slab, &notify_slab_reg, dfuse_info, &notify_slab);
+}
+
+/* Start the threads.  Not called until after fuse is mounted */
 int
 ival_thread_start(struct dfuse_info *dfuse_info)
 {
@@ -292,16 +631,31 @@ ival_thread_start(struct dfuse_info *dfuse_info)
 
 	ival_data.session = dfuse_info->di_session;
 
-	rc = pthread_create(&ival_thread, NULL, ival_thread_fn, NULL);
+	rc = notify_init(dfuse_info);
+	if (rc != -DER_SUCCESS)
+		return daos_der2errno(rc);
+
+	rc = pthread_create(&notify_thread, NULL, notify_thread_fn, NULL);
 	if (rc != 0)
-		goto out;
+		return rc;
+	pthread_setname_np(notify_thread, "dfuse notify");
+
+	rc = pthread_create(&ival_thread, NULL, ival_thread_fn, NULL);
+	if (rc != 0) {
+		D_MUTEX_LOCK(&notify_lock);
+		notify_stop = true;
+		D_MUTEX_UNLOCK(&notify_lock);
+		sem_post(&notify_sem);
+		pthread_join(notify_thread, NULL);
+		notify_thread = 0;
+		return rc;
+	}
 	pthread_setname_np(ival_thread, "dfuse inval");
 
-out:
-	return rc;
+	return 0;
 }
 
-/* Stop thread, remove all inodes from the invalidation queues and teardown all data structures
+/* Stop threads, remove all inodes from the invalidation queues and teardown all data structures.
  * May be called without thread_start() having been called.
  */
 void
@@ -314,6 +668,18 @@ ival_thread_stop()
 	if (ival_thread)
 		pthread_join(ival_thread, NULL);
 	ival_thread = 0;
+
+	/* Latch stop before waking the thread: anything still landing behind an in-flight
+	 * request during unmount is dropped rather than queued behind a thread about to exit.
+	 */
+	D_MUTEX_LOCK(&notify_lock);
+	notify_stop = true;
+	D_MUTEX_UNLOCK(&notify_lock);
+	sem_post(&notify_sem);
+
+	if (notify_thread)
+		pthread_join(notify_thread, NULL);
+	notify_thread = 0;
 }
 
 void
@@ -331,6 +697,10 @@ ival_fini()
 		d_list_del(&dte->dte_list);
 		D_FREE(dte);
 	}
+
+	sem_destroy(&notify_sem);
+	sem_destroy(&ival_sem);
+	DFUSE_TRA_DOWN(&notify_data);
 	DFUSE_TRA_DOWN(&ival_data);
 }
 
diff --git a/src/client/dfuse/ops/ioctl.c b/src/client/dfuse/ops/ioctl.c
index 8d073b859ff..3526ec47262 100644
--- a/src/client/dfuse/ops/ioctl.c
+++ b/src/client/dfuse/ops/ioctl.c
@@ -1,5 +1,6 @@
 /**
  * (C) Copyright 2016-2024 Intel Corporation.
+ * (C) Copyright 2026 Hewlett Packard Enterprise Development LP
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -44,16 +45,7 @@ handle_il_ioctl(struct dfuse_obj_hdl *oh, fuse_req_t req)
 		il_reply.fir_flags |= DFUSE_IOCTL_FLAGS_MCACHE;
 
 	if (oh->doh_writeable) {
-		rc = fuse_lowlevel_notify_inval_inode(dfuse_info->di_session,
-						      oh->doh_ie->ie_stat.st_ino, 0, 0);
-
-		if (rc == 0) {
-			DFUSE_TRA_DEBUG(oh, "inval inode %#lx rc is %d", oh->doh_ie->ie_stat.st_ino,
-					rc);
-		} else {
-			DFUSE_TRA_ERROR(oh, "inval inode %#lx rc is %d", oh->doh_ie->ie_stat.st_ino,
-					rc);
-		}
+		dfuse_notify_inval_inode(dfuse_info, oh->doh_ie->ie_stat.st_ino);
 
 		/* Mark this file handle as using the IL or similar, and if this is new then mark
 		 * the inode as well
diff --git a/src/client/dfuse/ops/lookup.c b/src/client/dfuse/ops/lookup.c
index 1f1f25201b9..d7bba8c2979 100644
--- a/src/client/dfuse/ops/lookup.c
+++ b/src/client/dfuse/ops/lookup.c
@@ -1,6 +1,6 @@
 /**
  * (C) Copyright 2016-2024 Intel Corporation.
- * (C) Copyright 2025 Hewlett Packard Enterprise Development LP
+ * (C) Copyright 2025-2026 Hewlett Packard Enterprise Development LP
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -163,10 +163,7 @@ dfuse_reply_entry(struct dfuse_info *dfuse_info, struct dfuse_inode_entry *ie,
 	if (wipe_parent == 0)
 		return;
 
-	rc = fuse_lowlevel_notify_inval_entry(dfuse_info->di_session, wipe_parent, wipe_name,
-					      strnlen(wipe_name, NAME_MAX));
-	if (rc && rc != -ENOENT)
-		DS_ERROR(-rc, "inval_entry() failed");
+	dfuse_notify_inval_entry(dfuse_info, wipe_parent, wipe_name);
 
 	return;
 out_err:
diff --git a/src/client/dfuse/ops/open.c b/src/client/dfuse/ops/open.c
index 5a5f7d3494e..183bdba6783 100644
--- a/src/client/dfuse/ops/open.c
+++ b/src/client/dfuse/ops/open.c
@@ -242,11 +242,7 @@ dfuse_cb_release(fuse_req_t req, fuse_ino_t ino, struct fuse_file_info *fi)
 		dfuse_inode_decref(dfuse_info, oh->doh_parent_dir);
 	}
 	if (ie) {
-		rc = fuse_lowlevel_notify_inval_entry(dfuse_info->di_session, ie->ie_parent,
-						      ie->ie_name, strnlen(ie->ie_name, NAME_MAX));
-
-		if (rc != 0 && rc != -ENOENT)
-			DHS_ERROR(ie, -rc, "inval_entry() error");
+		dfuse_notify_inval_entry(dfuse_info, ie->ie_parent, ie->ie_name);
 		dfuse_inode_decref(dfuse_info, ie);
 	}
 	dfuse_oh_free(dfuse_info, oh);
diff --git a/src/client/dfuse/ops/opendir.c b/src/client/dfuse/ops/opendir.c
index 60d1e6ab4d5..d6fce8667d0 100644
--- a/src/client/dfuse/ops/opendir.c
+++ b/src/client/dfuse/ops/opendir.c
@@ -1,5 +1,6 @@
 /**
  * (C) Copyright 2016-2024 Intel Corporation.
+ * (C) Copyright 2026 Hewlett Packard Enterprise Development LP
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -81,13 +82,7 @@ dfuse_cb_releasedir(fuse_req_t req, struct dfuse_inode_entry *ino, struct fuse_f
 
 	DFUSE_REPLY_ZERO_OH(oh, req);
 	if (ie) {
-		int rc;
-
-		rc = fuse_lowlevel_notify_inval_entry(dfuse_info->di_session, ie->ie_parent,
-						      ie->ie_name, strnlen(ie->ie_name, NAME_MAX));
-
-		if (rc != 0 && rc != -ENOENT)
-			DHS_ERROR(ie, -rc, "inval_entry() error");
+		dfuse_notify_inval_entry(dfuse_info, ie->ie_parent, ie->ie_name);
 		dfuse_inode_decref(dfuse_info, ie);
 	}
 	dfuse_oh_free(dfuse_info, oh);
diff --git a/src/client/dfuse/ops/rename.c b/src/client/dfuse/ops/rename.c
index fcf95b82466..e74f143c087 100644
--- a/src/client/dfuse/ops/rename.c
+++ b/src/client/dfuse/ops/rename.c
@@ -1,5 +1,6 @@
 /**
  * (C) Copyright 2016-2023 Intel Corporation.
+ * (C) Copyright 2026 Hewlett Packard Enterprise Development LP
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -18,7 +19,6 @@ dfuse_oid_moved(struct dfuse_info *dfuse_info, daos_obj_id_t *oid, struct dfuse_
 		const char *name, struct dfuse_inode_entry *newparent, const char *newname)
 {
 	struct dfuse_inode_entry *ie;
-	int                       rc;
 	ino_t                     ino;
 
 	dfuse_compute_inode(parent->ie_dfs, oid, &ino);
@@ -34,11 +34,7 @@ dfuse_oid_moved(struct dfuse_info *dfuse_info, daos_obj_id_t *oid, struct dfuse_
 		(strncmp(ie->ie_name, name, NAME_MAX) != 0)) {
 		DFUSE_TRA_DEBUG(ie, "Invalidating old name");
 
-		rc = fuse_lowlevel_notify_inval_entry(dfuse_info->di_session, ie->ie_parent,
-						      ie->ie_name, strnlen(ie->ie_name, NAME_MAX));
-
-		if (rc && rc != -ENOENT)
-			DFUSE_TRA_ERROR(ie, "inval_entry() returned: %d (%s)", rc, strerror(-rc));
+		dfuse_notify_inval_entry(dfuse_info, ie->ie_parent, ie->ie_name);
 	}
 
 	/* Update the inode entry data */
diff --git a/src/client/dfuse/ops/setxattr.c b/src/client/dfuse/ops/setxattr.c
index e50c010b377..861a7303e38 100644
--- a/src/client/dfuse/ops/setxattr.c
+++ b/src/client/dfuse/ops/setxattr.c
@@ -1,5 +1,6 @@
 /**
  * (C) Copyright 2019-2022 Intel Corporation.
+ * (C) Copyright 2026 Hewlett Packard Enterprise Development LP
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -46,22 +47,15 @@ dfuse_cb_setxattr(fuse_req_t req, struct dfuse_inode_entry *inode,
 
 	rc = dfs_setxattr(inode->ie_dfs->dfs_ns, inode->ie_obj, name, value, size, flags);
 	if (rc == 0) {
-		/* Optionally remove the dentry to force a new lookup on access.
-		 * If the xattr is to set a UNS entry point, and dentry_dir
-		 * caching is enabled then invalidate the dentry here, to force
-		 * a lookup which will check the xattr and return the linked
-		 * container.  The fuse header says this potentially deadlocks
-		 * however it does appear to work, and calling this after the
-		 * reply will introduce a race condition that future lookups
-		 * will be skipped.
+		/* Setting a UNS entry point renumbers the inode, so invalidate the dentry to
+		 * force a fresh lookup.  Safe out of order: a racing lookup holds the parent
+		 * shared while it instantiates, the invalidation takes it exclusive, so a stale
+		 * dentry is always in place before the queued invalidation kills it.
 		 */
 		if (duns_attr && inode->ie_dfs->dfc_dentry_dir_timeout > 0) {
 			struct dfuse_info *dfuse_info = fuse_req_userdata(req);
 
-			rc = fuse_lowlevel_notify_inval_entry(dfuse_info->di_session,
-							      inode->ie_parent, inode->ie_name,
-							      strnlen(inode->ie_name, NAME_MAX));
-			DFUSE_TRA_INFO(inode, "inval_entry() rc is %d", rc);
+			dfuse_notify_inval_entry(dfuse_info, inode->ie_parent, inode->ie_name);
 		}
 		DFUSE_REPLY_ZERO(inode, req);
 		return;
diff --git a/src/client/dfuse/ops/unlink.c b/src/client/dfuse/ops/unlink.c
index 778d44e0bed..9af5a68ba28 100644
--- a/src/client/dfuse/ops/unlink.c
+++ b/src/client/dfuse/ops/unlink.c
@@ -1,5 +1,6 @@
 /**
  * (C) Copyright 2016-2023 Intel Corporation.
+ * (C) Copyright 2026 Hewlett Packard Enterprise Development LP
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -18,7 +19,6 @@ dfuse_oid_unlinked(struct dfuse_info *dfuse_info, fuse_req_t req, daos_obj_id_t
 		   struct dfuse_inode_entry *parent, const char *name)
 {
 	struct dfuse_inode_entry *ie;
-	int                       rc;
 	fuse_ino_t                ino;
 	ino_t                     parent_ino;
 
@@ -46,9 +46,7 @@ dfuse_oid_unlinked(struct dfuse_info *dfuse_info, fuse_req_t req, daos_obj_id_t
 	 * unlinked so will destroy it anyway, but there is a race here so try and destroy it
 	 * even though most of the time we expect this to fail.
 	 */
-	rc = fuse_lowlevel_notify_inval_inode(dfuse_info->di_session, ino, 0, 0);
-	if (rc && rc != -ENOENT)
-		DHS_ERROR(ie, -rc, "inval_inode() error");
+	dfuse_notify_inval_inode(dfuse_info, ino);
 
 	/* If the kernel was aware of this inode at an old location then remove that which should
 	 * trigger a forget call.  Checking the test logs shows that we do see the forget anyway
@@ -58,10 +56,7 @@ dfuse_oid_unlinked(struct dfuse_info *dfuse_info, fuse_req_t req, daos_obj_id_t
 		DFUSE_TRA_DEBUG(ie, "Telling kernel to forget %#lx " DF_DE, ie->ie_parent,
 				DP_DE(ie->ie_name));
 
-		rc = fuse_lowlevel_notify_delete(dfuse_info->di_session, ie->ie_parent, ino,
-						 ie->ie_name, strnlen(ie->ie_name, NAME_MAX));
-		if (rc && rc != -ENOENT)
-			DHS_ERROR(ie, -rc, "notify_delete() error");
+		dfuse_notify_delete(dfuse_info, ie->ie_parent, ino, ie->ie_name);
 	}
 
 	/* Drop the ref again */
diff --git a/src/client/dfuse/tests/SConscript b/src/client/dfuse/tests/SConscript
new file mode 100644
index 00000000000..7883d06abec
--- /dev/null
+++ b/src/client/dfuse/tests/SConscript
@@ -0,0 +1,17 @@
+"""Build DFuse unit tests"""
+
+
+def scons():
+    """Execute build"""
+    Import('denv')
+
+    tenv = denv.Clone()
+    tenv.AppendUnique(LIBS=['cmocka'])
+
+    notify_tests = tenv.d_test_program('notify_tests', 'notify_tests.c')
+
+    tenv.Install('$PREFIX/bin/', [notify_tests])
+
+
+if __name__ == "SCons.Script":
+    scons()
diff --git a/src/client/dfuse/tests/notify_tests.c b/src/client/dfuse/tests/notify_tests.c
new file mode 100644
index 00000000000..86510c64631
--- /dev/null
+++ b/src/client/dfuse/tests/notify_tests.c
@@ -0,0 +1,341 @@
+/**
+ * (C) Copyright 2026 Hewlett Packard Enterprise Development LP
+ *
+ * SPDX-License-Identifier: BSD-2-Clause-Patent
+ */
+#include <stdarg.h>
+#include <stddef.h>
+#include <setjmp.h>
+#include <string.h>
+#include <stdio.h>
+#include <cmocka.h>
+
+#include <daos/common.h>
+#include <daos/debug.h>
+
+/* DFUSE_NOTIFY_RAW_OK is defined inside inval.c before it pulls in dfuse.h, so the poison
+ * pragma there never fires for this TU; these stand in for the real fuse session calls.
+ */
+static int stub_notify_rc;
+static int stub_inval_entry_calls;
+static int stub_delete_calls;
+static int stub_inval_inode_calls;
+
+#include "../inval.c"
+
+int
+fuse_lowlevel_notify_inval_entry(struct fuse_session *se, fuse_ino_t parent, const char *name,
+				 size_t namelen)
+{
+	stub_inval_entry_calls++;
+	return stub_notify_rc;
+}
+
+int
+fuse_lowlevel_notify_expire_entry(struct fuse_session *se, fuse_ino_t parent, const char *name,
+				  size_t namelen)
+{
+	return stub_notify_rc;
+}
+
+int
+fuse_lowlevel_notify_delete(struct fuse_session *se, fuse_ino_t parent, fuse_ino_t child,
+			    const char *name, size_t namelen)
+{
+	stub_delete_calls++;
+	return stub_notify_rc;
+}
+
+int
+fuse_lowlevel_notify_inval_inode(struct fuse_session *se, fuse_ino_t ino, off_t off, off_t len)
+{
+	stub_inval_inode_calls++;
+	return stub_notify_rc;
+}
+
+/* Eviction sweep is out of scope here; ival_loop() is compiled but never invoked. */
+bool
+dfuse_dentry_get_valid(struct dfuse_inode_entry *ie, double max_age, double *timeout)
+{
+	return false;
+}
+
+static struct dfuse_info tnotify_info;
+
+static void
+tnotify_reset(void)
+{
+	memset(&tnotify_info, 0, sizeof(tnotify_info));
+	memset(&ival_data, 0, sizeof(ival_data));
+	memset(&notify_data, 0, sizeof(notify_data));
+	notify_queued          = 0;
+	notify_stop            = false;
+	notify_congest_total   = 0;
+	stub_notify_rc         = 0;
+	stub_inval_entry_calls = 0;
+	stub_delete_calls      = 0;
+	stub_inval_inode_calls = 0;
+
+	assert_int_equal(d_slab_init(&tnotify_info.di_slab, &tnotify_info), -DER_SUCCESS);
+	assert_int_equal(ival_init(&tnotify_info), 0);
+	assert_int_equal(notify_init(&tnotify_info), -DER_SUCCESS);
+}
+
+static int
+tnotify_setup(void **state)
+{
+	tnotify_reset();
+	return 0;
+}
+
+static int
+tnotify_teardown(void **state)
+{
+	D_MUTEX_LOCK(&notify_lock);
+	notify_drain_locked();
+	D_MUTEX_UNLOCK(&notify_lock);
+
+	ival_fini();
+	d_slab_destroy(&tnotify_info.di_slab);
+	return 0;
+}
+
+static void
+test_coalesce_duplicate(void **state)
+{
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo");
+	assert_int_equal(notify_queued, 1);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_enqueued), 1);
+
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo");
+	assert_int_equal(notify_queued, 1);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_coalesced), 1);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_enqueued), 1);
+}
+
+static void
+test_coalesce_key_fields_differ(void **state)
+{
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo");
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 10, 0, "foo");
+	assert_int_equal(notify_queued, 2);
+
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 11, 0, "foo");
+	assert_int_equal(notify_queued, 3);
+
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "bar");
+	assert_int_equal(notify_queued, 4);
+
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_coalesced), 0);
+}
+
+/* Same parent+name but different child ino: the kernel verifies child identity on delete, so
+ * collapsing these would silently drop one of the two deletions (design-review finding).
+ */
+static void
+test_coalesce_delete_distinct_child_ino(void **state)
+{
+	notify_enqueue(&tnotify_info, NOTIFY_DELETE, 10, 100, "foo");
+	notify_enqueue(&tnotify_info, NOTIFY_DELETE, 10, 200, "foo");
+
+	assert_int_equal(notify_queued, 2);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_coalesced), 0);
+}
+
+static void
+test_pop_before_delivery_not_coalesced(void **state)
+{
+	struct notify_entry *ne;
+
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo");
+
+	D_MUTEX_LOCK(&notify_lock);
+	ne = notify_dequeue_locked();
+	D_MUTEX_UNLOCK(&notify_lock);
+	assert_non_null(ne);
+	d_slab_release(notify_slab, ne);
+
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo");
+	assert_int_equal(notify_queued, 1);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_coalesced), 0);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_enqueued), 2);
+}
+
+static void
+test_backstop_drop_at_max(void **state)
+{
+	int i;
+
+	for (i = 0; i < NOTIFY_QUEUE_MAX; i++)
+		notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 0, i + 1, NULL);
+
+	assert_int_equal(notify_queued, NOTIFY_QUEUE_MAX);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 0);
+
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 0, NOTIFY_QUEUE_MAX + 1, NULL);
+
+	assert_int_equal(notify_queued, NOTIFY_QUEUE_MAX);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 1);
+}
+
+static void
+test_stop_latch_drops(void **state)
+{
+	notify_stop = true;
+
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo");
+
+	assert_int_equal(notify_queued, 0);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 1);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_enqueued), 0);
+}
+
+static void
+test_congest_warn_silent_below_threshold(void **state)
+{
+	fuse_ino_t ino = 1;
+
+	for (; notify_queued < NOTIFY_CONGEST_HIGH - 1; ino++)
+		notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 0, ino, NULL);
+
+	assert_int_equal(notify_congest_total, 0);
+}
+
+static void
+test_congest_warn_at_and_above_threshold(void **state)
+{
+	fuse_ino_t ino = 1;
+
+	for (; notify_queued < NOTIFY_CONGEST_HIGH; ino++)
+		notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 0, ino, NULL);
+	assert_int_equal(notify_congest_total, 1);
+
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_INODE, 0, ino++, NULL);
+	assert_int_equal(notify_congest_total, 2);
+}
+
+static void
+test_delivery_rc_enoent_ignored(void **state)
+{
+	struct notify_entry *ne;
+
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo");
+
+	D_MUTEX_LOCK(&notify_lock);
+	ne = notify_dequeue_locked();
+	D_MUTEX_UNLOCK(&notify_lock);
+	assert_non_null(ne);
+
+	stub_notify_rc = -ENOENT;
+	notify_run(ne);
+	d_slab_release(notify_slab, ne);
+
+	assert_false(atomic_load_relaxed(&ival_data.session_dead));
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_delivered), 1);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 0);
+}
+
+static void
+test_delivery_rc_ebadf_latches_and_drains(void **state)
+{
+	struct notify_entry *ne;
+
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 10, 0, "foo");
+	notify_enqueue(&tnotify_info, NOTIFY_INVAL_ENTRY, 11, 0, "bar");
+
+	D_MUTEX_LOCK(&notify_lock);
+	ne = notify_dequeue_locked();
+	D_MUTEX_UNLOCK(&notify_lock);
+	assert_non_null(ne);
+
+	stub_notify_rc = -EBADF;
+	notify_run(ne);
+	d_slab_release(notify_slab, ne);
+
+	assert_true(atomic_load_relaxed(&ival_data.session_dead));
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 1);
+
+	/* Session dead: the delivery thread drains whatever remains queued */
+	D_MUTEX_LOCK(&notify_lock);
+	notify_drain_locked();
+	D_MUTEX_UNLOCK(&notify_lock);
+
+	assert_int_equal(notify_queued, 0);
+	assert_int_equal(atomic_load_relaxed(&tnotify_info.di_notify_dropped), 2);
+}
+
+static void
+test_hash_same_key_same_bucket(void **state)
+{
+	uint32_t b1 = notify_hash(NOTIFY_DELETE, 42, 7, "somefile", 8);
+	uint32_t b2 = notify_hash(NOTIFY_DELETE, 42, 7, "somefile", 8);
+
+	assert_int_equal(b1, b2);
+}
+
+#define HASH_SAMPLE_COUNT 4096
+
+static void
+test_hash_distribution_smoke(void **state)
+{
+	static unsigned int counts[NOTIFY_COALESCE_BUCKETS];
+	unsigned int        max_count = 0;
+	int                 i;
+
+	memset(counts, 0, sizeof(counts));
+
+	for (i = 0; i < HASH_SAMPLE_COUNT; i++) {
+		char     name[32];
+		uint32_t bucket;
+
+		snprintf(name, sizeof(name), "file-%d", i);
+		bucket = notify_hash(NOTIFY_INVAL_ENTRY, i, 0, name, strlen(name));
+		counts[bucket]++;
+		if (counts[bucket] > max_count)
+			max_count = counts[bucket];
+	}
+
+	/* Loose FNV-1a smoke check: average is 1/bucket here, no bucket should be swamped */
+	assert_true(max_count < 20);
+}
+
+int
+main(void)
+{
+	const struct CMUnitTest tests[] = {
+	    cmocka_unit_test_setup_teardown(test_coalesce_duplicate, tnotify_setup,
+					    tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_coalesce_key_fields_differ, tnotify_setup,
+					    tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_coalesce_delete_distinct_child_ino, tnotify_setup,
+					    tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_pop_before_delivery_not_coalesced, tnotify_setup,
+					    tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_backstop_drop_at_max, tnotify_setup,
+					    tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_stop_latch_drops, tnotify_setup, tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_congest_warn_silent_below_threshold, tnotify_setup,
+					    tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_congest_warn_at_and_above_threshold, tnotify_setup,
+					    tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_delivery_rc_enoent_ignored, tnotify_setup,
+					    tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_delivery_rc_ebadf_latches_and_drains,
+					    tnotify_setup, tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_hash_same_key_same_bucket, tnotify_setup,
+					    tnotify_teardown),
+	    cmocka_unit_test_setup_teardown(test_hash_distribution_smoke, tnotify_setup,
+					    tnotify_teardown),
+	};
+	int rc;
+
+	rc = daos_debug_init(DAOS_LOG_DEFAULT);
+	if (rc != 0)
+		return rc;
+
+	rc = cmocka_run_group_tests_name("dfuse notify queue", tests, NULL, NULL);
+
+	daos_debug_fini();
+
+	return rc;
+}
diff --git a/utils/cq/words.dict b/utils/cq/words.dict
index 801a166ed92..0211097656f 100644
--- a/utils/cq/words.dict
+++ b/utils/cq/words.dict
@@ -184,6 +184,7 @@ epilog
 errored
 ethernet
 fallocate
+faulthandler
 fchmod
 fcntl
 filename
diff --git a/utils/node_local_test.py b/utils/node_local_test.py
index 7d2e3191ab2..49a94197e27 100755
--- a/utils/node_local_test.py
+++ b/utils/node_local_test.py
@@ -20,6 +20,7 @@
 import argparse
 import copy
 import errno
+import faulthandler
 import functools
 import importlib
 import json
@@ -47,6 +48,19 @@
 import xattr
 import yaml
 
+# How long a process may show no sign of progress before it is declared stalled.
+STALL_SECS = int(os.environ.get('NLT_STALL_SECS', '300'))
+# Factor to be used for calculating timeouts when running with valgrind.
+VALGRIND_SLOWDOWN = 6
+
+# Bounds on the stall diagnostics themselves.
+DIAG_TIMEOUT = 60
+STACK_READ_TIMEOUT = 15
+DUMP_DEADLINE_SECS = 120
+
+# How long a killed process gets to shut down; longer than this is assumed to be hung.
+KILL_GRACE = 30
+
 
 class NLTestFail(Exception):
     """Used to indicate test failure"""
@@ -125,10 +139,30 @@ def __init__(self, json_file, args):
             os.makedirs(self.tmp_dir)
 
         self._compress_procs = []
+        self._cleaned = False
 
     def __del__(self):
+        self.cleanup()
+
+    def cleanup(self):
+        """Report any leaked dfuse mount, flush compression and remove the working directory"""
+        if self._cleaned:
+            return
+        self._cleaned = True
+
+        leaked = {conn: mnt for conn, mnt in _dfuse_connection_ids().items()
+                  if mnt.startswith(self.dfuse_parent_dir)}
+        if leaked:
+            print(f'Leaked dfuse mounts at exit: {sorted(leaked.values())}', flush=True)
+            for line in _abort_fuse_connections()[1]:
+                print(line, flush=True)
+
         self.flush_bz2()
-        os.rmdir(self.dfuse_parent_dir)
+
+        try:
+            os.rmdir(self.dfuse_parent_dir)
+        except OSError as err:
+            print(f'Could not remove {self.dfuse_parent_dir}: {err}', flush=True)
 
     def set_wf(self, wf):
         """Set the WarningsFactory object"""
@@ -1341,7 +1375,7 @@ class DFuse():
     # pylint: disable-next=too-many-arguments
     def __init__(self, daos, conf, pool=None, container=None, mount_path=None, uns_path=None,
                  caching=True, wbcache=True, multi_user=False, ro=False, dump_h=False,
-                 read_h=False, file_h=None):
+                 read_h=False, file_h=None, thread_count=None):
         if mount_path:
             self.dir = mount_path
         else:
@@ -1357,6 +1391,7 @@ def __init__(self, daos, conf, pool=None, container=None, mount_path=None, uns_p
         self.conf = conf
         self.multi_user = multi_user
         self.cores = 0
+        self.thread_count = thread_count
         self._daos = daos
         self.caching = caching
         self.wbcache = wbcache
@@ -1431,7 +1466,9 @@ def start(self, v_hint=None, use_oopt=False):
         if self.multi_user:
             cmd.append('--multi-user')
 
-        if not self.cores:
+        if self.thread_count:
+            cmd.extend(['--thread-count', str(self.thread_count)])
+        elif not self.cores:
             # Use a lower default thread-count for NLT due to running tests in parallel.
             cmd.extend(['--thread-count', '4'])
 
@@ -1519,7 +1556,8 @@ def stop(self, ignore_einval=False):
         print('Stopping fuse')
 
         if self.container:
-            self.run_query(use_json=True)
+            # This queries the mount that may itself be wedged.
+            self.run_query(use_json=True, timeout=120)
         ret = umount(self.dir)
         if ret:
             umount(self.dir, background=True)
@@ -1606,10 +1644,10 @@ def il_cmd(self, cmd, check_read=True, check_write=True, check_fstat=True):
         assert ret.returncode == 0, ret
         return ret
 
-    def run_query(self, use_json=False, quiet=False):
+    def run_query(self, use_json=False, quiet=False, timeout=None):
         """Run filesystem query"""
         rc = run_daos_cmd(self.conf, ['filesystem', 'query', self.dir],
-                          use_json=use_json, log_check=quiet, valgrind=quiet)
+                          use_json=use_json, log_check=quiet, valgrind=quiet, timeout=timeout)
         print(rc)
         return rc
 
@@ -1730,7 +1768,8 @@ def run_daos_cmd(conf,
                  log_check=True,
                  ignore_busy=False,
                  use_json=False,
-                 cwd=None):
+                 cwd=None,
+                 timeout=None):
     """Run a DAOS command
 
     Run a command, returning what subprocess.run() would.
@@ -1777,8 +1816,24 @@ def run_daos_cmd(conf,
 
     cmd_env['DAOS_AGENT_DRPC_DIR'] = conf.agent_dir
 
-    rc = subprocess.run(exec_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-                        env=cmd_env, check=False, cwd=cwd)
+    # Avoid getting stuck on child processes that are blocked in the kernel.
+    # pylint: disable-next=consider-using-with
+    proc = subprocess.Popen(exec_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                            env=cmd_env, cwd=cwd)
+    timed_out = False
+    try:
+        stdout, stderr = proc.communicate(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        timed_out = True
+        print(f'Timeout after {timeout}s running {" ".join(cmd)}', flush=True)
+        proc.kill()
+        try:
+            stdout, stderr = proc.communicate(timeout=KILL_GRACE)
+        except subprocess.TimeoutExpired:
+            print(f'Command did not exit after kill: {" ".join(cmd)}', flush=True)
+            stdout, stderr = b'', b''
+    returncode = -signal.SIGKILL if proc.returncode is None else proc.returncode
+    rc = subprocess.CompletedProcess(exec_cmd, returncode, stdout, stderr)
 
     if rc.stderr != b'':
         print('Stderr from command')
@@ -1807,7 +1862,14 @@ def run_daos_cmd(conf,
         conf.valgrind_errors = True
         rc.returncode = 0
     if use_json:
-        rc.json = json.loads(rc.stdout.decode('utf-8'))
+        try:
+            rc.json = json.loads(rc.stdout.decode('utf-8'))
+        except json.JSONDecodeError:
+            if timed_out:
+                print(f'No JSON output from timed-out command: {" ".join(cmd)}')
+                rc.json = None
+            else:
+                raise
     dcr.rc = rc
     return dcr
 
@@ -4662,6 +4724,142 @@ def import_torch(self, server):
         return importlib.import_module('pydaos.torch')
 
 
+class StressTests(PosixTests):
+    """Stress and race reproducers, run sequentially against a dedicated server"""
+
+    @staticmethod
+    def generate_test_list():
+        """Generate list of stress tests"""
+        return [x for x in dir(StressTests)
+                if x.startswith('test') and x not in dir(PosixTests)]
+
+    def _run_uns_wedge_test(self, v_hint, churn_sh, create_path_fn, create_must_succeed):
+        """Race UNS container-create against a churn script; detect a wedged mount.
+
+        Fails by timing out - a wedged mount answers nothing further and blocked
+        processes cannot be killed.  churn_sh is a shell template with {dfuse_dir}/
+        {stop_file} placeholders; create_path_fn(dfuse_dir, idx) builds the create path.
+        """
+        # --thread-count is reduced by the event-queue count, so this leaves exactly one
+        # worker; more would need enough concurrency to consume every worker to wedge.
+        dfuse = DFuse(self.server, self.conf, caching=True, container=self.container,
+                      thread_count=2)
+        dfuse.use_valgrind = False
+        # Churn generates high op volume; debug logs would blow the NLT log budget.
+        dfuse.log_mask = 'WARN'
+        dfuse.start(v_hint=v_hint)
+
+        churn = []
+        stop_file = join(dfuse.dir, 'stop_churn')
+        cmd = churn_sh.format(dfuse_dir=dfuse.dir, stop_file=stop_file)
+
+        cmd_env = get_base_env()
+        cmd_env['D_LOG_MASK'] = 'WARN'
+        cmd_env['DAOS_AGENT_DRPC_DIR'] = self.conf.agent_dir
+
+        wedged = False
+        aborted = False
+        stuck = False
+        successes = 0
+        proc = None
+        try:
+            for _ in range(4):
+                # pylint: disable-next=consider-using-with
+                churn.append(subprocess.Popen(['sh', '-c', cmd]))
+
+            for idx in range(8):
+                create_cmd = [join(self.conf['PREFIX'], 'bin', 'daos'), 'container', 'create',
+                              '--type', 'POSIX', '--path', create_path_fn(dfuse.dir, idx)]
+                # Popen rather than run(timeout=): a wedged mount leaves this process in
+                # uninterruptible sleep, so run() would hang trying to kill it.  The
+                # connection has to be aborted before the process can be reaped.
+                # pylint: disable-next=consider-using-with
+                proc = subprocess.Popen(create_cmd, env=cmd_env, stdout=subprocess.PIPE,
+                                        stderr=subprocess.STDOUT)
+                try:
+                    stdout, _ = proc.communicate(timeout=60)
+                except subprocess.TimeoutExpired:
+                    print(f'container create {idx} did not return; mount is wedged')
+                    wedged = True
+                    break
+                if proc.returncode == 0:
+                    successes += 1
+                else:
+                    print(f'container create {idx} failed rc={proc.returncode}')
+                    print(stdout.decode('utf-8', errors='replace'))
+                    if create_must_succeed:
+                        self.fail()
+
+            if wedged:
+                # Nothing below can complete against a wedged mount and the processes cannot
+                # be signalled, so break the connection before trying to clean up.
+                _, lines = _abort_fuse_connections()
+                for line in lines:
+                    print(line)
+                aborted = True
+                proc.wait(timeout=60)
+                self.fail()
+
+            if not create_must_succeed and successes == 0:
+                print(f'None of 8 creates against {v_hint} succeeded')
+                self.fail()
+        finally:
+            try:
+                with open(stop_file, 'w'):
+                    pass
+            except OSError:
+                pass
+            for cproc in churn:
+                try:
+                    cproc.wait(timeout=30)
+                except subprocess.TimeoutExpired:
+                    cproc.kill()
+                    try:
+                        cproc.wait(timeout=10)
+                    except subprocess.TimeoutExpired:
+                        print('Churn process is unkillable; mount is still wedged')
+                        stuck = True
+            if stuck and not aborted:
+                # The create loop never saw the wedge, so it was never aborted above; abort
+                # now or dfuse.stop() below hangs the same way the churn children did.
+                _, lines = _abort_fuse_connections()
+                for line in lines:
+                    print(line)
+            if dfuse.stop():
+                self.fatal_errors = True
+        if stuck and not aborted:
+            self.fail()
+
+    def test_uns_create_vs_dir_lock(self):
+        """Race UNS container-create against mkdir churn holding the mount root locked.
+
+        Exercises the setxattr notification path.  Fails by timing out.
+        """
+        churn_sh = ('i=0; while [ ! -e {stop_file} ]; do '
+                    'mkdir {dfuse_dir}/churn_$$_$i 2>/dev/null; i=$((i+1)); done')
+        self._run_uns_wedge_test(
+            'uns_create_vs_dir_lock', churn_sh,
+            lambda dfuse_dir, idx: join(dfuse_dir, f'uns_{idx}'),
+            create_must_succeed=True)
+
+    def test_uns_create_vs_rmdir_churn(self):
+        """Race UNS container-create against rmdir/rename churn on the same paths.
+
+        Exercises the unlink and rename notification paths.  Fails by timing out.
+        """
+        num_paths = 4
+        churn_sh = ('i=0; while [ ! -e {stop_file} ]; do '
+                    f'n=$((i % {num_paths})); '
+                    'mkdir {dfuse_dir}/uns_$n 2>/dev/null; '
+                    'mv {dfuse_dir}/uns_$n {dfuse_dir}/uns_${{n}}_old 2>/dev/null; '
+                    'rmdir {dfuse_dir}/uns_${{n}}_old 2>/dev/null; '
+                    'i=$((i+1)); done')
+        self._run_uns_wedge_test(
+            'uns_create_vs_rmdir_churn', churn_sh,
+            lambda dfuse_dir, idx: join(dfuse_dir, f'uns_{idx % num_paths}'),
+            create_must_succeed=False)
+
+
 class NltStdoutWrapper():
     """Class for capturing stdout from threads"""
 
@@ -4739,12 +4937,14 @@ def __del__(self):
         sys.stderr = self._stderr
 
 
-def run_posix_tests(server, conf, test_list):
+def run_posix_tests(server, conf, test_list, klass=None, sequential=False):
     """Run one or all posix tests
 
     Create a new container per test, to ensure that every test is
     isolated from others.
     """
+    if klass is None:
+        klass = PosixTests
 
     def _run_test(ptl=None, function=None, test_cb=None):
         ptl.call_index = 0
@@ -4801,7 +5001,7 @@ def _run_test(ptl=None, function=None, test_cb=None):
     out_wrapper = NltStdoutWrapper()
     err_wrapper = NltStderrWrapper()
 
-    pto = PosixTests(server, conf, pool=pool)
+    pto = klass(server, conf, pool=pool)
     if len(test_list) == 1:
         obj = getattr(pto, test_list[0])
 
@@ -4815,11 +5015,15 @@ def _run_test(ptl=None, function=None, test_cb=None):
         test_list.sort(key=lambda x: x not in slow_tests)
 
         for function in test_list:
-            ptl = PosixTests(server, conf, pool=pool)
+            ptl = klass(server, conf, pool=pool)
             obj = getattr(ptl, function)
             if not callable(obj):
                 continue
 
+            if sequential:
+                _run_test(ptl=ptl, test_cb=obj, function=function)
+                continue
+
             thread = threading.Thread(None,
                                       target=_run_test,
                                       name=f'test {function}',
@@ -5616,6 +5820,216 @@ def test_pydaos_kv_obj_class(server, conf):
 #
 
 
+def _run_diag(cmd, timeout=DIAG_TIMEOUT):
+    """Run a diagnostic command returning (exit code, output), never raising or blocking
+
+    A command that outlives timeout is killed and reports a nonzero code.
+    """
+    try:
+        # pylint: disable-next=consider-using-with
+        proc = subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
+                                stderr=subprocess.STDOUT, start_new_session=True)
+    except OSError as err:
+        return 127, f'<{cmd[0]} failed: {err}>'
+
+    def output(wait_secs):
+        return proc.communicate(timeout=wait_secs)[0].decode('utf-8', errors='replace').rstrip()
+
+    try:
+        text = output(timeout)
+    except subprocess.TimeoutExpired:
+        # Overdue: kill the process group and salvage whatever it wrote.
+        try:
+            os.killpg(proc.pid, signal.SIGKILL)
+        except OSError:
+            pass
+        try:
+            return -1, f'{output(KILL_GRACE)}\n<{cmd[0]} exceeded {timeout}s>'
+        except subprocess.TimeoutExpired:
+            proc.stdout.close()
+            return -1, f'<{cmd[0]} exceeded {timeout}s and could not be reaped>'
+    return proc.returncode, text
+
+
+def _read_file(path):
+    """Return the stripped content of a small /proc or /sys file, or None"""
+    try:
+        with open(path, encoding='utf-8', errors='replace') as pfile:
+            return pfile.read().strip()
+    except OSError:
+        return None
+
+
+def _proc_state(pid):
+    """Return blocked system-call state for a pid from /proc"""
+    out = []
+    for name in ('wchan', 'syscall'):
+        out.append(f'{name}={_read_file(f"/proc/{pid}/{name}") or "<unreadable>"}')
+    status = _read_file(f'/proc/{pid}/status') or ''
+    for line in status.splitlines():
+        if line.startswith(('State:', 'Threads:')):
+            out.append(line.strip())
+    return ' '.join(out)
+
+
+def _proc_threads(pid, deadline=None):
+    """Report every thread of a process from /proc; debuggers cannot attach in D state"""
+    lines = []
+    stack_err = None
+    try:
+        tids = sorted(os.listdir(f'/proc/{pid}/task'), key=int)
+    except OSError as err:
+        return f'<cannot list threads of {pid}: {err}>'
+    for tid in tids:
+        if deadline is not None and time.monotonic() > deadline:
+            lines.append(f'  <dump deadline reached; skipping remaining of {len(tids)} threads>')
+            break
+        base = f'/proc/{pid}/task/{tid}'
+        fields = []
+        for name in ('comm', 'wchan'):
+            fields.append(f'{name}={_read_file(f"{base}/{name}") or "?"}')
+        # The state field follows the parenthesized command name, which can itself
+        # contain ') ', so anchor on the last occurrence per proc(5).
+        state = (_read_file(f'{base}/stat') or '').rpartition(') ')[2].split()
+        if state:
+            fields.append(f'state={state[0]}')
+        lines.append(f'  TID {tid}: ' + ' '.join(fields))
+        rc, kstack = _run_diag(['sudo', 'cat', f'{base}/stack'], timeout=STACK_READ_TIMEOUT)
+        if rc == 0 and kstack:
+            for kline in kstack.splitlines()[:12]:
+                lines.append(f'      {kline.strip()}')
+        elif rc != 0 and stack_err is None:
+            stack_err = (kstack or '<no output>').splitlines()[0].strip()
+            lines.append(f'  <kernel stack reads failing: rc={rc} {stack_err}>')
+    return '\n'.join(lines)
+
+
+def dump_stalled(active, log_dir=None):
+    """Dump diagnostics for wedged child processes, to stdout and to a file"""
+    dump_file = None
+    if log_dir:
+        try:
+            # The dnt*.log name is what CI's log collection keeps; append across stalls.
+            # pylint: disable-next=consider-using-with
+            dump_file = open(join(log_dir, 'dnt_stall_dump.log'), 'a', encoding='utf-8')
+        except OSError:
+            dump_file = None
+
+    def emit(text):
+        # File first; a stalled stdout consumer must not cost us the dump.
+        if dump_file:
+            try:
+                dump_file.write(f'{text}\n')
+                dump_file.flush()
+            except OSError:
+                pass
+        print(text, flush=True)
+
+    try:
+        deadline = time.monotonic() + DUMP_DEADLINE_SECS
+        emit(f'\n===== NLT STALL DETECTED {time.strftime("%Y-%m-%d %H:%M:%S")} =====')
+        # /proc cannot block, so gather it before anything that can.
+        for child in active:
+            pid = child.pid()
+            emit(f'--- stalled child: loc={child.loc} pid={pid} elapsed={child.elapsed():.0f}s')
+            emit(_proc_threads(pid, deadline=deadline))
+        daemons = _run_diag(['pgrep', '-a', 'daos_engine|daos_agent|dfuse'])[1]
+        emit(f'--- daemons:\n{daemons}')
+        daemon_pids = [int(line.split()[0]) for line in daemons.splitlines()
+                       if line and line.split()[0].isdigit()]
+        for pid in daemon_pids:
+            if time.monotonic() > deadline:
+                emit('<dump deadline reached; skipping remaining daemons>')
+                break
+            emit(f'--- daemon pid={pid}:')
+            emit(_proc_threads(pid, deadline=deadline))
+        emit('--- process tree:')
+        emit(_run_diag(['ps', 'auxwwf'])[1])
+        emit('===== NLT STALL DUMP COMPLETE =====')
+    except Exception as err:  # pylint: disable=broad-except
+        emit(f'===== NLT STALL DUMP ABORTED: {err!r} =====')
+        traceback.print_exc(file=sys.stdout)
+        sys.stdout.flush()
+    finally:
+        if dump_file:
+            dump_file.close()
+
+
+def _dfuse_connection_ids():
+    """Return the FUSE connection ids of dfuse mounts, mapped to their mount points"""
+    ids = {}
+    # Safer than stat, which could block on a wedged mount
+    for line in (_read_file('/proc/self/mountinfo') or '').splitlines():
+        fields = line.split()
+        if '-' in fields and 'fuse.daos' in fields[fields.index('-'):]:
+            ids[fields[2].split(':')[1]] = fields[4]
+    return ids
+
+
+def _abort_fuse_connections():
+    """Abort every backed-up dfuse connection, failing its requests so blocked processes can die"""
+    done = []
+    aborted = 0
+    dfuse_conns = _dfuse_connection_ids()
+    try:
+        conns = sorted(os.listdir('/sys/fs/fuse/connections'))
+    except OSError as err:
+        return 0, [f'<cannot list fuse connections: {err}>']
+    for conn in conns:
+        if conn not in dfuse_conns:
+            continue
+        waiting = _read_file(f'/sys/fs/fuse/connections/{conn}/waiting')
+        if waiting is None:
+            done.append(f'<cannot read waiting count of dfuse connection {conn}>')
+            continue
+        if waiting in ('', '0'):
+            continue
+        path = f'/sys/fs/fuse/connections/{conn}/abort'
+        rc, res = _run_diag(['sudo', 'sh', '-c', f'echo 1 > {path}'])
+        if rc == 0:
+            aborted += 1
+            done.append(f'aborted dfuse connection {conn} (waiting={waiting}): {res or "ok"}')
+        else:
+            done.append(f'<abort of dfuse connection {conn} failed rc={rc}: {res}>')
+    return aborted, (done or ['<no backed-up dfuse connections found>'])
+
+
+def handle_stalled(active, log_dir=None):
+    """Dump diagnostics then clear the wedged children; returns whether a mount was aborted"""
+    dump_stalled(active, log_dir=log_dir)
+    aborted_mount = False
+    for child in active:
+        child.hang_kill()
+
+    def _find_zombies(wait_sec):
+        deadline = time.monotonic() + wait_sec
+        alive = list(active)
+        while alive and time.monotonic() < deadline:
+            alive = [c for c in alive if not c.is_dead()]
+            if alive:
+                time.sleep(1)
+        return alive
+
+    zombies = _find_zombies(KILL_GRACE)
+    if zombies:
+        # A child that outlives the kill grace is likely blocked in the kernel.
+        # We can try to free it up by failing its FUSE requests so that it
+        # can exit.
+        print(f'{len(zombies)} child(ren) survived the kill; '
+              f'aborting backed-up FUSE connections to release them', flush=True)
+        aborted, lines = _abort_fuse_connections()
+        for line in lines:
+            print(line, flush=True)
+        if aborted:
+            aborted_mount = True
+            zombies = _find_zombies(KILL_GRACE)
+
+    for child in zombies:
+        print(f'WARNING: pid {child.pid()} (loc {child.loc}) survived SIGKILL: '
+              f'{_proc_state(child.pid())}', flush=True)
+    return aborted_mount
+
+
 class AllocFailTestRun():
     """Class to run a fault injection command with a single fault"""
 
@@ -5631,6 +6045,8 @@ def __init__(self, aft, cmd, env, loc, cwd):
         self.dir_handle = None
         self.stdout = None
         self.returncode = None
+        self.was_killed = False
+        self._start_time = None
 
         # Set this to disable memory leak checking if the command outputs a DER_BUSY message.  This
         # is to allow tests to leak memory if there are errors during shutdown.
@@ -5715,6 +6131,26 @@ def start(self):
                                     stdin=subprocess.PIPE,
                                     stdout=subprocess.PIPE,
                                     stderr=subprocess.PIPE)
+        self._start_time = time.monotonic()
+
+    def pid(self):
+        """Return the pid of the command"""
+        return self._sp.pid
+
+    def elapsed(self):
+        """Return seconds since the command started"""
+        return time.monotonic() - self._start_time
+
+    def hang_kill(self):
+        """Kill a wedged command; result checks are skipped on reap"""
+        if self._sp.poll() is not None:
+            return
+        self.was_killed = True
+        self._sp.kill()
+
+    def is_dead(self):
+        """Return whether the command has exited, without blocking"""
+        return self._sp.poll() is not None
 
     def has_finished(self):
         """Check if the command has completed"""
@@ -5724,15 +6160,45 @@ def has_finished(self):
         rc = self._sp.poll()
         if rc is None:
             return False
-        self._post(rc)
+        self._reap(rc)
         return True
 
-    def wait(self):
+    def wait(self, timeout=None):
         """Wait for the command to complete"""
         if self.returncode is not None:
             return
 
-        self._post(self._sp.wait())
+        if timeout is not None:
+            try:
+                self._reap(self._sp.wait(timeout=timeout))
+            except subprocess.TimeoutExpired:
+                handle_stalled([self], log_dir=self._aft.log_dir)
+                try:
+                    self._reap(self._sp.wait(timeout=KILL_GRACE))
+                except subprocess.TimeoutExpired:
+                    # Blocked in the kernel; record the kill rather than joining the hang.
+                    self._post_killed(-signal.SIGKILL)
+            return
+
+        self._reap(self._sp.wait())
+
+    def _reap(self, rc):
+        """Process a completed command, bypassing result checks for watchdog kills"""
+        if self.was_killed:
+            self._post_killed(rc)
+        else:
+            self._post(rc)
+
+    def _post_killed(self, rc):
+        """Reap after a watchdog kill; the positive returncode stops a valgrind re-run"""
+        print()
+        self.returncode = 128 - rc if rc < 0 else rc
+        try:
+            self.stdout, self._stderr = self._sp.communicate(timeout=KILL_GRACE)
+        except subprocess.TimeoutExpired:
+            self.stdout = b'<stdout unavailable: pipe held open after kill>'
+            self._stderr = b''
+        self.fault_injected = True
 
     def _post(self, rc):
         """Helper function, called once after command is complete.
@@ -5929,7 +6395,9 @@ def launch(self):
 
         def _prep(self):
             rc = self._run_cmd(None)
-            rc.wait()
+            rc.wait(timeout=STALL_SECS)
+            if rc.was_killed:
+                raise NLTestFail('prep run (no faults enabled) stalled and was killed')
             self.expected_stdout = rc.stdout
             assert not rc.fault_injected
 
@@ -5965,6 +6433,10 @@ def _prep(self):
 
         max_load_avg = 100
 
+        last_progress = time.monotonic()
+        stall_rounds = 0
+        stalled_out = False
+
         # Now run all iterations in parallel up to max_child.  Iterations will be launched
         # in order but may not finish in order, rather they are processed in the order they
         # finish.  After each repetition completes then check for re-launch new processes
@@ -6005,6 +6477,7 @@ def _prep(self):
                 if not ret.has_finished():
                     continue
                 active.remove(ret)
+                last_progress = time.monotonic()
                 print()
                 print(ret)
                 if ret.returncode < 0:
@@ -6016,15 +6489,43 @@ def _prep(self):
                     finished = True
                 break
 
+            if active and time.monotonic() - last_progress > STALL_SECS:
+                stall_rounds += 1
+                fatal_errors = True
+                if handle_stalled(active, log_dir=self.log_dir):
+                    print('Mount was aborted to clear the stall; ending this sweep')
+                    stalled_out = True
+                elif stall_rounds >= 2:
+                    # A second stall means the mount did not really recover; new
+                    # children will only stall again, forever.
+                    print('Sweep stalled again after a cleared stall; giving up on it')
+                    stalled_out = True
+                if stalled_out:
+                    finished = True
+                    # Reap the children that died; waiting for one that is stuck in
+                    # the kernel would hang the run.
+                    for child in active:
+                        if not child.has_finished():
+                            print(f'Abandoning stuck pid {child.pid()} (loc {child.loc})',
+                                  flush=True)
+                    active.clear()
+                last_progress = time.monotonic()
+
         print(f'Completed, fid {fid}')
         print(f'Max in flight {max_count}/{max_child}')
         if to_rerun:
             print(f'Number of indexes to re-run {len(to_rerun)}')
+            if stalled_out:
+                print('Skipping valgrind re-runs; the mount did not survive the sweep')
+                to_rerun = []
 
         for fid in to_rerun:
             rerun = self._run_cmd(fid, valgrind=True)
             print(rerun)
-            rerun.wait()
+            rerun.wait(timeout=STALL_SECS * VALGRIND_SLOWDOWN)
+            if rerun.was_killed and self.conf.args.failfast:
+                print(f'--failfast set; skipping remaining re-runs after stall at {fid}')
+                break
 
         return fatal_errors
 
@@ -6747,6 +7248,8 @@ def run(wf, args):
 
     if args.mode == 'fi':
         fi_test = True
+    elif args.mode == 'stress':
+        pass
     else:
         for rep in range(args.repeat):
             if args.repeat > 1:
@@ -6772,6 +7275,13 @@ def run(wf, args):
                 print(f'--failfast set; stopping after iteration {rep + 1}/{args.repeat}')
                 break
 
+    if args.mode in ('all', 'stress'):
+        with DaosServer(conf, test_class='stress', wf=wf_server,
+                        fatal_errors=fatal_errors) as server:
+            fatal_errors.add_result(
+                run_posix_tests(server, conf, StressTests.generate_test_list(),
+                                klass=StressTests, sequential=True))
+
     if args.mode == 'all':
         with DaosServer(conf, test_class='restart', wf=wf_server,
                         fatal_errors=fatal_errors) as server:
@@ -6805,6 +7315,9 @@ def run(wf, args):
         print(fs)
         if fs.returncode == 0:
             run_fi = True
+        elif fi_test or fi_test_dfuse:
+            raise NLTestFail('Unable to detect fault injection feature '
+                             '- cannot run requested FI tests')
         else:
             print("Unable to detect fault injection feature - skipping FI testing")
 
@@ -6880,11 +7393,19 @@ def run(wf, args):
 
     wf_server.close()
     close_log_test(conf)
+    conf.cleanup()
     print(f'Total time in log analysis: {conf.log_timer.total:.2f} seconds')
     print(f'Total time in log compression: {conf.compress_timer.total:.2f} seconds')
     return fatal_errors
 
 
+def _exit_now(code):
+    """Terminate without interpreter shutdown, whose cleanup can block on a wedged mount"""
+    sys.stdout.flush()
+    sys.stderr.flush()
+    os._exit(code)  # pylint: disable=protected-access
+
+
 def _positive_int(value):
     """argparse type that rejects values below 1."""
     ivalue = int(value)
@@ -6902,6 +7423,10 @@ def main():
     Test names can either be 'pure' or include suffices that encode a particular caching
     regimen (e.g., read_caching_off).
     """
+    # SIGUSR1 dumps every thread's Python stack.  sys.__stderr__ because NLT's stderr
+    # wrapper lacks the fileno() that faulthandler requires.
+    faulthandler.register(signal.SIGUSR1, file=sys.__stderr__, all_threads=True)
+
     parser = argparse.ArgumentParser(description='Run DAOS client on local node')
     parser.add_argument('--server-debug', default=None)
     parser.add_argument('--dfuse-debug', default=None)
@@ -6917,7 +7442,7 @@ def main():
     parser.add_argument('--repeat', type=_positive_int, default=1,
                         help='Repeat the test execution N times (soak/stability testing)')
     parser.add_argument('--failfast', action='store_true',
-                        help='With --repeat, stop after the first failing iteration')
+                        help='Stop after the first failing --repeat iteration or stalled re-run')
     parser.add_argument('--system-ram-reserved', type=int, default=None, help='GiB reserved RAM')
     parser.add_argument('--dfuse-dir', default='/tmp', help='parent directory for all dfuse mounts')
     parser.add_argument('--perf-check', action='store_true')
@@ -6988,7 +7513,8 @@ def main():
 
     if fatal_errors.errors:
         print("Significant errors encountered")
-        sys.exit(1)
+        _exit_now(1)
+    _exit_now(0)
 
 
 if __name__ == '__main__':
diff --git a/utils/utest.yaml b/utils/utest.yaml
index e5077e57202..64ccd82526e 100644
--- a/utils/utest.yaml
+++ b/utils/utest.yaml
@@ -165,6 +165,10 @@
     - cmd: ["bin/eq_tests"]
     - cmd: ["bin/agent_tests"]
     - cmd: ["bin/job_tests"]
+- name: dfuse
+  base: "PREFIX"
+  tests:
+    - cmd: ["bin/notify_tests"]
 - name: cart
   base: "BUILD_DIR"
   tests: