[PATCH 04/22] ksmbd: retry break notifications on another connection

[Namjae Jeon <linkinjeon@kernel.org>]

An oplock or lease break belongs to an open that can outlive its original
channel. A draining channel or stalled TCP writer can otherwise leave the
break pending until timeout or invalidate a durable open.

Snapshot the owner's live channels and try eligible connections with the
same notification. Track TCP delivery separately from the SMB ACK, retry
when a channel stops progressing, and share one break deadline. Run
ordinary break waits on a dedicated workqueue so blocked request
responses do not stall breaks.

Publish NONE before sending a level II break, which requires no ACK, so
an unsolicited ACK observes the completed level while TCP delivery is
pending.

Keep the first rename lease notification in the request context and
enqueue the interim writer only after the notification is accepted. Join
that work before returning so a blocked interim response does not stop
notification retries or release the request early.

Reported-by: Pan Guihua <pan.guihua@datatom.com>
Signed-off-by: Namjae Jeon <linkinjeon@kernel.org>
---
 fs/smb/server/ksmbd_work.c        |  19 +
 fs/smb/server/ksmbd_work.h        |   2 +
 fs/smb/server/mgmt/user_session.c |  31 ++
 fs/smb/server/mgmt/user_session.h |   3 +
 fs/smb/server/oplock.c            | 855 +++++++++++++++++-------------
 fs/smb/server/oplock.h            |  14 +-
 fs/smb/server/vfs_cache.c         |   2 +
 7 files changed, 539 insertions(+), 387 deletions(-)

diff --git a/fs/smb/server/ksmbd_work.c b/fs/smb/server/ksmbd_work.c
index 52ea1a674a67..6e5122e2d6fc 100644
--- a/fs/smb/server/ksmbd_work.c
+++ b/fs/smb/server/ksmbd_work.c
@@ -16,6 +16,7 @@
 
 static struct kmem_cache *work_cache;
 static struct workqueue_struct *ksmbd_wq;
+static struct workqueue_struct *ksmbd_break_wq;
 
 static int ksmbd_reserve_iov(struct ksmbd_work *work, int need_iov_cnt)
 {
@@ -110,6 +111,12 @@ int ksmbd_workqueue_init(void)
 	ksmbd_wq = alloc_workqueue("ksmbd-io", WQ_PERCPU, 0);
 	if (!ksmbd_wq)
 		return -ENOMEM;
+	ksmbd_break_wq = alloc_workqueue("ksmbd-break", WQ_UNBOUND, 0);
+	if (!ksmbd_break_wq) {
+		destroy_workqueue(ksmbd_wq);
+		ksmbd_wq = NULL;
+		return -ENOMEM;
+	}
 	return 0;
 }
 
@@ -117,6 +124,8 @@ void ksmbd_workqueue_destroy(void)
 {
 	destroy_workqueue(ksmbd_wq);
 	ksmbd_wq = NULL;
+	destroy_workqueue(ksmbd_break_wq);
+	ksmbd_break_wq = NULL;
 }
 
 bool ksmbd_queue_work(struct ksmbd_work *work)
@@ -124,6 +133,16 @@ bool ksmbd_queue_work(struct ksmbd_work *work)
 	return queue_work(ksmbd_wq, &work->work);
 }
 
+void ksmbd_queue_io_work(struct work_struct *work)
+{
+	queue_work(ksmbd_wq, work);
+}
+
+void ksmbd_queue_break_work(struct work_struct *work)
+{
+	queue_work(ksmbd_break_wq, work);
+}
+
 static inline void __ksmbd_iov_pin(struct ksmbd_work *work, void *ib,
 				   unsigned int ib_len)
 {
diff --git a/fs/smb/server/ksmbd_work.h b/fs/smb/server/ksmbd_work.h
index 40f70a5d08b5..b51dea69cb0e 100644
--- a/fs/smb/server/ksmbd_work.h
+++ b/fs/smb/server/ksmbd_work.h
@@ -155,6 +155,8 @@ int ksmbd_work_pool_init(void);
 int ksmbd_workqueue_init(void);
 void ksmbd_workqueue_destroy(void);
 bool ksmbd_queue_work(struct ksmbd_work *work);
+void ksmbd_queue_io_work(struct work_struct *work);
+void ksmbd_queue_break_work(struct work_struct *work);
 int ksmbd_iov_pin_rsp_read(struct ksmbd_work *work, void *ib, int len,
 			   void *aux_buf, unsigned int aux_size);
 int ksmbd_iov_pin_rsp(struct ksmbd_work *work, void *ib, int len);
diff --git a/fs/smb/server/mgmt/user_session.c b/fs/smb/server/mgmt/user_session.c
index d814a99ff519..554434c7673a 100644
--- a/fs/smb/server/mgmt/user_session.c
+++ b/fs/smb/server/mgmt/user_session.c
@@ -674,6 +674,37 @@ struct ksmbd_session *ksmbd_session_lookup_slowpath(unsigned long long id)
 	return sess;
 }
 
+/* The caller holds a session reference and must put each returned connection. */
+unsigned int ksmbd_session_get_channels(struct ksmbd_session *sess,
+					struct ksmbd_conn **conns,
+					unsigned int max_conns)
+{
+	struct channel *chann;
+	unsigned long index;
+	unsigned int nr_conns = 0;
+
+	/*
+	 * Channel connections are borrowed. Unregister can leave the channel
+	 * list in a pinned session after removing it from the registry, so
+	 * chann_lock alone does not keep those connections alive. Only collect
+	 * registered sessions, excluding both unregister and connection cleanup
+	 * until the connection references have been taken.
+	 */
+	down_read(&sessions_table_lock);
+	if (hlist_unhashed(&sess->hlist))
+		goto out;
+	down_read(&sess->chann_lock);
+	xa_for_each(&sess->ksmbd_chann_list, index, chann) {
+		if (nr_conns == max_conns)
+			break;
+		conns[nr_conns++] = ksmbd_conn_get(chann->conn);
+	}
+	up_read(&sess->chann_lock);
+out:
+	up_read(&sessions_table_lock);
+	return nr_conns;
+}
+
 struct ksmbd_session *ksmbd_session_lookup_all_states(struct ksmbd_conn *conn,
 						      unsigned long long id)
 {
diff --git a/fs/smb/server/mgmt/user_session.h b/fs/smb/server/mgmt/user_session.h
index 217258551d6d..aff31f1694db 100644
--- a/fs/smb/server/mgmt/user_session.h
+++ b/fs/smb/server/mgmt/user_session.h
@@ -94,6 +94,9 @@ struct ksmbd_session *ksmbd_smb2_session_create(void);
 void ksmbd_session_destroy(struct ksmbd_session *sess);
 
 struct ksmbd_session *ksmbd_session_lookup_slowpath(unsigned long long id);
+unsigned int ksmbd_session_get_channels(struct ksmbd_session *sess,
+					struct ksmbd_conn **conns,
+					unsigned int max_conns);
 struct ksmbd_session *ksmbd_session_lookup(struct ksmbd_conn *conn,
 					   unsigned long long id);
 bool is_ksmbd_session_in_connection(struct ksmbd_conn *conn,
diff --git a/fs/smb/server/oplock.c b/fs/smb/server/oplock.c
index 25ff8408f0ba..313bf01ad350 100644
--- a/fs/smb/server/oplock.c
+++ b/fs/smb/server/oplock.c
@@ -87,6 +87,7 @@ static struct oplock_info *alloc_opinfo(struct ksmbd_work *work,
 		return NULL;
 
 	opinfo->sess = sess;
+	opinfo->session_id = sess->id;
 	opinfo->conn = ksmbd_conn_get(work->conn);
 	opinfo->level = SMB2_OPLOCK_LEVEL_NONE;
 	opinfo->op_state = OPLOCK_STATE_NONE;
@@ -302,15 +303,9 @@ static struct oplock_info *opinfo_get_list(struct ksmbd_inode *ci,
 	opinfo = list_first_entry_or_null(&ci->m_op_list, struct oplock_info,
 					  op_entry);
 	if (opinfo) {
-		if (opinfo->conn == NULL ||
-		    !atomic_inc_not_zero(&opinfo->refcount))
+		/* A live open may still be bound to a disconnected channel. */
+		if (!atomic_inc_not_zero(&opinfo->refcount))
 			opinfo = NULL;
-		else {
-			if (ksmbd_conn_releasing(opinfo->conn)) {
-				atomic_dec(&opinfo->refcount);
-				opinfo = NULL;
-			}
-		}
 
 		if (opinfo && snapshot && opinfo->o_fp &&
 		    opinfo->o_fp != skip_fp &&
@@ -367,7 +362,7 @@ static void opinfo_del(struct oplock_info *opinfo)
 		lease_del_open(opinfo);
 
 	down_write(&ci->m_lock);
-	list_del(&opinfo->op_entry);
+	list_del_init(&opinfo->op_entry);
 	up_write(&ci->m_lock);
 }
 
@@ -639,19 +634,27 @@ static void grant_none_oplock(struct oplock_info *opinfo_new,
 static inline int compare_guid_key(struct oplock_info *opinfo,
 				   const char *guid1, const char *key1)
 {
-	const char *guid2, *key2;
+	struct lease *lease = opinfo->o_lease;
+	struct lease_table *lb;
 	struct ksmbd_conn *conn;
+	int ret = 0;
 
-	conn = READ_ONCE(opinfo->conn);
-	if (!conn)
+	if (memcmp(key1, lease->lease_key, SMB2_LEASE_KEY_SIZE))
 		return 0;
-	guid2 = conn->ClientGUID;
-	key2 = opinfo->o_lease->lease_key;
-	if (!memcmp(guid1, guid2, SMB2_CLIENT_GUID_SIZE) &&
-	    !memcmp(key1, key2, SMB2_LEASE_KEY_SIZE))
-		return 1;
 
-	return 0;
+	/* The caller holds ci->m_lock, which protects opinfo->conn. */
+	conn = READ_ONCE(opinfo->conn);
+	if (conn)
+		return !memcmp(guid1, conn->ClientGUID, SMB2_CLIENT_GUID_SIZE);
+
+	/* A detached durable lease retains its client identity in the table. */
+	read_lock(&lease_list_lock);
+	lb = lease->l_lb;
+	if (lb)
+		ret = !memcmp(guid1, lb->client_guid, SMB2_CLIENT_GUID_SIZE);
+	read_unlock(&lease_list_lock);
+
+	return ret;
 }
 
 /**
@@ -681,7 +684,7 @@ static struct oplock_info *same_client_has_lease(struct ksmbd_inode *ci,
 	 */
 	down_read(&ci->m_lock);
 	list_for_each_entry(opinfo, &ci->m_op_list, op_entry) {
-		if (!opinfo->is_lease || !opinfo->conn)
+		if (!opinfo->is_lease)
 			continue;
 		lease = opinfo->o_lease;
 
@@ -734,35 +737,6 @@ static struct oplock_info *same_client_has_lease(struct ksmbd_inode *ci,
 	return m_opinfo;
 }
 
-static bool wait_for_break_ack(struct oplock_info *opinfo)
-{
-	int rc = 0;
-
-	rc = wait_event_interruptible_timeout(opinfo->oplock_q,
-					      opinfo->op_state == OPLOCK_STATE_NONE ||
-					      opinfo->op_state == OPLOCK_CLOSING,
-					      OPLOCK_WAIT_TIME);
-
-	/* is this a timeout ? */
-	if (!rc) {
-		spin_lock(&opinfo->state_lock);
-		if (opinfo->op_state == OPLOCK_CLOSING) {
-			spin_unlock(&opinfo->state_lock);
-			return false;
-		}
-		if (opinfo->is_lease) {
-			opinfo->o_lease->state = SMB2_LEASE_NONE_LE;
-			lease_update_oplock_levels(opinfo->o_lease);
-		}
-		opinfo->level = SMB2_OPLOCK_LEVEL_NONE;
-		opinfo->op_state = OPLOCK_STATE_NONE;
-		spin_unlock(&opinfo->state_lock);
-		return true;
-	}
-
-	return false;
-}
-
 static bool oplock_break_set_ack_wait(struct oplock_info *opinfo)
 {
 	bool ret = false;
@@ -842,330 +816,492 @@ static bool lease_break_needed(struct oplock_info *opinfo, int req_op_level,
 	return opinfo->level > req_op_level;
 }
 
-/**
- * __smb2_oplock_break_noti() - send smb2 oplock break cmd from conn
- * to client
- * @wk:     smb work object
- *
- * There are two ways this function can be called. 1- while file open we break
- * from exclusive/batch lock to levelII oplock and 2- while file write/truncate
- * we break from levelII oplock no oplock.
- * work->request_buf contains oplock_info.
- */
-static void __smb2_oplock_break_noti(struct work_struct *wk)
+struct smb2_break_channel {
+	struct list_head list;
+	struct ksmbd_conn *conn;
+	struct ksmbd_tx_tracking tracking;
+	bool queued;
+	bool delivered;
+	bool failed;
+};
+
+static bool smb2_break_complete(struct oplock_info *opinfo, bool expect_ack)
 {
-	struct smb2_oplock_break *rsp = NULL;
-	struct ksmbd_work *work = container_of(wk, struct ksmbd_work, work);
-	struct ksmbd_conn *conn = work->conn;
-	struct oplock_break_info *br_info = work->request_buf;
-	struct smb2_hdr *rsp_hdr;
-	struct ksmbd_file *fp;
+	bool complete;
 
-	fp = ksmbd_lookup_global_fd(br_info->fid);
-	if (!fp)
-		goto out;
+	spin_lock(&opinfo->state_lock);
+	complete = opinfo->op_state == OPLOCK_CLOSING ||
+		(expect_ack && opinfo->op_state == OPLOCK_STATE_NONE);
+	spin_unlock(&opinfo->state_lock);
+	return complete;
+}
 
-	if (allocate_interim_rsp_buf(work)) {
-		pr_err("smb2_allocate_rsp_buf failed! ");
-		ksmbd_fd_put(work, fp);
-		goto out;
+static int smb2_break_failed(struct oplock_info *opinfo, bool expect_ack)
+{
+	spin_lock(&opinfo->state_lock);
+	/* An ACK or close which won the race has already completed the break. */
+	if (opinfo->op_state == OPLOCK_CLOSING ||
+	    (expect_ack && opinfo->op_state == OPLOCK_STATE_NONE)) {
+		spin_unlock(&opinfo->state_lock);
+		return 0;
 	}
+	if (opinfo->is_lease) {
+		opinfo->o_lease->state = SMB2_LEASE_NONE_LE;
+		opinfo->o_lease->new_state = SMB2_LEASE_NONE_LE;
+		lease_update_oplock_levels(opinfo->o_lease);
+		atomic_set(&opinfo->breaking_cnt, 0);
+	}
+	opinfo->level = SMB2_OPLOCK_LEVEL_NONE;
+	opinfo->op_state = OPLOCK_STATE_NONE;
+	spin_unlock(&opinfo->state_lock);
+	wake_up_interruptible_all(&opinfo->oplock_q);
+	wake_up_interruptible_all(&opinfo->oplock_brk);
+	return ksmbd_invalidate_durable_fd(opinfo->fid);
+}
 
-	rsp_hdr = smb_get_msg(work->response_buf);
-	memset(rsp_hdr, 0, sizeof(struct smb2_hdr) + 2);
-	rsp_hdr->ProtocolId = SMB2_PROTO_NUMBER;
-	rsp_hdr->StructureSize = SMB2_HEADER_STRUCTURE_SIZE;
-	rsp_hdr->CreditRequest = cpu_to_le16(0);
-	rsp_hdr->Command = SMB2_OPLOCK_BREAK;
-	rsp_hdr->Flags = (SMB2_FLAGS_SERVER_TO_REDIR);
-	rsp_hdr->NextCommand = 0;
-	rsp_hdr->MessageId = cpu_to_le64(-1);
-	rsp_hdr->Id.SyncId.ProcessId = 0;
-	rsp_hdr->Id.SyncId.TreeId = 0;
-	rsp_hdr->SessionId = 0;
-	memset(rsp_hdr->Signature, 0, 16);
-
-	rsp = smb_get_msg(work->response_buf);
-
-	rsp->StructureSize = cpu_to_le16(24);
-	if (!br_info->open_trunc &&
-	    (br_info->level == SMB2_OPLOCK_LEVEL_BATCH ||
-	     br_info->level == SMB2_OPLOCK_LEVEL_EXCLUSIVE))
-		rsp->OplockLevel = SMB2_OPLOCK_LEVEL_II;
-	else
-		rsp->OplockLevel = SMB2_OPLOCK_LEVEL_NONE;
-	rsp->Reserved = 0;
-	rsp->Reserved2 = 0;
-	rsp->PersistentFid = fp->persistent_id;
-	rsp->VolatileFid = fp->volatile_id;
-
-	ksmbd_fd_put(work, fp);
-	if (ksmbd_iov_pin_rsp(work, (void *)rsp,
-			      sizeof(struct smb2_oplock_break)))
-		goto out;
-
-	ksmbd_debug(OPLOCK,
-		    "sending oplock break v_id %llu p_id = %llu lock level = %d\n",
-		    rsp->VolatileFid, rsp->PersistentFid, rsp->OplockLevel);
-
-	ksmbd_conn_write(work);
+static int smb2_break_add_channel(struct list_head *head, struct ksmbd_conn *conn)
+{
+	struct smb2_break_channel *channel;
 
-out:
-	ksmbd_free_work_struct(work);
-	ksmbd_conn_r_count_dec(conn);
-	ksmbd_conn_put(conn);
+	channel = kzalloc_obj(*channel, KSMBD_DEFAULT_GFP);
+	if (!channel)
+		return -ENOMEM;
+	channel->conn = ksmbd_conn_get(conn);
+	list_add_tail(&channel->list, head);
+	return 0;
 }
 
 /*
- * Select and pin the connection used for an oplock break before doing any
- * allocations which may sleep.  The caller of oplock_break() holds a live
- * reference on ci (a file being opened, a file being operated on, or an
- * explicit ksmbd_inode_lookup_lock() reference in the parent lease break
- * paths), so the inode cannot be freed during the call and its lock is
- * reachable without dereferencing opinfo->o_fp, which is not pinned by
- * the oplock reference and may be freed by a concurrent close.
- *
- * opinfo->conn is cleared under ci->m_lock by session_fd_check() when the
- * durable handle owning the oplock is disconnected, reassigned by
- * ksmbd_reopen_durable_fd() under the same lock, and the last
- * ksmbd_conn_put() of the old connection frees it.  Holding the read lock
- * excludes both writers, so the connection cannot be freed while it is
- * selected.
+ * Snapshot connections before sending. References keep their transports alive,
+ * while r_count is taken separately for each write, under request_lock.
  */
-static struct ksmbd_conn *smb2_oplock_break_conn_get(struct oplock_info *opinfo,
-						     struct ksmbd_inode *ci)
-{
+static int smb2_break_collect_channels(struct oplock_info *opinfo,
+				       struct ksmbd_inode *ci,
+				       struct smb2_break_channel *owner,
+				       struct list_head *head,
+				       struct smb2_oplock_break *rsp)
+{
+	struct ksmbd_session *sess = NULL;
+	struct ksmbd_conn *session_conns[KSMBD_MAX_CHANNELS];
+	struct ksmbd_file_table *ft;
+	struct ksmbd_file *fp;
 	struct ksmbd_conn *conn;
+	struct lease_table *lb;
+	char guid[SMB2_CLIENT_GUID_SIZE];
+	u64 session_id = 0;
+	unsigned int nr_conns, i;
+	bool have_guid = false;
+	int bkt, ret = 0;
 
+	/* m_lock protects membership and binding, but not the session's lifetime. */
 	down_read(&ci->m_lock);
-	conn = READ_ONCE(opinfo->conn);
-	if (conn && !ksmbd_conn_releasing(conn))
-		conn = ksmbd_conn_get(conn);
-	else
-		conn = NULL;
+	if (!list_empty(&opinfo->op_entry)) {
+		if (!opinfo->is_lease) {
+			rsp->PersistentFid = READ_ONCE(opinfo->o_fp->persistent_id);
+			rsp->VolatileFid = READ_ONCE(opinfo->o_fp->volatile_id);
+			if (!has_file_id(rsp->PersistentFid) ||
+			    !has_file_id(rsp->VolatileFid)) {
+				/* A detached oplock must take the failed-break path. */
+				ret = opinfo->conn ? -ENOENT : -ENOTCONN;
+				goto unlock;
+			}
+			session_id = opinfo->session_id;
+		}
+		conn = opinfo->conn;
+		if (conn) {
+			owner->conn = ksmbd_conn_get(conn);
+			memcpy(guid, conn->ClientGUID, sizeof(guid));
+			have_guid = true;
+		}
+	} else {
+		ret = -ENOENT;
+	}
+unlock:
 	up_read(&ci->m_lock);
+	if (ret)
+		return ret;
+
+	if (opinfo->is_lease && !have_guid) {
+		read_lock(&lease_list_lock);
+		lb = opinfo->o_lease->l_lb;
+		if (lb) {
+			memcpy(guid, lb->client_guid, sizeof(guid));
+			have_guid = true;
+			if (lb->conn)
+				owner->conn = ksmbd_conn_get(lb->conn);
+		}
+		read_unlock(&lease_list_lock);
+	}
+	if (owner->conn)
+		list_add_tail(&owner->list, head);
+
+	if (!opinfo->is_lease) {
+		/* Session teardown takes m_lock while holding the session table lock. */
+		sess = ksmbd_session_lookup_slowpath(session_id);
+		if (!sess)
+			return -ENOENT;
+
+		/* Revalidate the binding and IDs after taking a safe session reference. */
+		down_read(&ci->m_lock);
+		if (list_empty(&opinfo->op_entry) || opinfo->sess != sess) {
+			ret = -ENOENT;
+		} else {
+			ft = &sess->file_table;
+			read_lock(&ft->lock);
+			fp = opinfo->o_fp;
+			if (fp->f_state == FP_CLOSED ||
+			    !has_file_id(fp->volatile_id) ||
+			    idr_find(ft->idr, fp->volatile_id) != fp) {
+				ret = -ENOENT;
+			} else {
+				rsp->PersistentFid = READ_ONCE(fp->persistent_id);
+				rsp->VolatileFid = fp->volatile_id;
+			}
+			read_unlock(&ft->lock);
+		}
+		up_read(&ci->m_lock);
+		if (ret) {
+			ksmbd_user_session_put(sess);
+			return ret;
+		}
+	}
 
-	return conn;
+	if (opinfo->is_lease && have_guid) {
+		down_read(&conn_list_lock);
+		hash_for_each(conn_list, bkt, conn, hlist) {
+			if (conn == owner->conn || !ksmbd_conn_good(conn) ||
+			    memcmp(conn->ClientGUID, guid, sizeof(guid)))
+				continue;
+			ret = smb2_break_add_channel(head, conn);
+			if (ret)
+				break;
+		}
+		up_read(&conn_list_lock);
+	} else if (!opinfo->is_lease && sess && sess->dialect >= SMB30_PROT_ID) {
+		nr_conns = ksmbd_session_get_channels(sess, session_conns,
+						      ARRAY_SIZE(session_conns));
+		for (i = 0; i < nr_conns; i++) {
+			conn = session_conns[i];
+			if (!ret && conn && conn != owner->conn && ksmbd_conn_good(conn))
+				ret = smb2_break_add_channel(head, conn);
+			/* Drop all snapshot references, also on allocation failure. */
+			ksmbd_conn_put(conn);
+		}
+	}
+	ksmbd_user_session_put(sess);
+
+	/* Allocation pressure may limit retries, but must not discard the owner. */
+	return list_empty(head) ? (ret ?: -ENOTCONN) : 0;
 }
 
-/**
- * smb2_oplock_break_noti() - send smb2 exclusive/batch to level2 oplock
- *		break command from server to client
- * @opinfo:		oplock info object
- * @ci:		inode owning the break target's oplock list, pinned by
- *		the caller
- *
- * Return:      0 on success, otherwise error
- */
-static int smb2_oplock_break_noti(struct oplock_info *opinfo,
-				  struct ksmbd_inode *ci)
+static bool smb2_break_delivered(struct list_head *head, bool require_live)
 {
+	struct smb2_break_channel *channel;
 	struct ksmbd_conn *conn;
-	struct oplock_break_info *br_info;
-	int ret = 0;
-	struct ksmbd_work *work;
-
-	conn = smb2_oplock_break_conn_get(opinfo, ci);
-	if (!conn)
-		return ksmbd_invalidate_durable_fd(opinfo->fid);
 
-	work = ksmbd_alloc_work_struct();
-	if (!work) {
-		ksmbd_conn_put(conn);
-		return -ENOMEM;
+	list_for_each_entry(channel, head, list) {
+		conn = channel->conn;
+		if (!channel->queued || (require_live && !ksmbd_conn_good(conn)))
+			continue;
+		/* A synchronous send needs no separate delivery probe. */
+		if (channel->delivered || !channel->tracking.valid ||
+		    conn->transport->ops->is_acked(conn->transport,
+						  &channel->tracking)) {
+			channel->delivered = true;
+			return true;
+		}
 	}
+	return false;
+}
 
-	br_info = kmalloc_obj(struct oplock_break_info, KSMBD_DEFAULT_GFP);
-	if (!br_info) {
-		ksmbd_free_work_struct(work);
-		ksmbd_conn_put(conn);
-		return -ENOMEM;
-	}
+static bool smb2_break_has_unsent_channels(struct list_head *head)
+{
+	struct smb2_break_channel *channel;
 
-	br_info->level = opinfo->level;
-	br_info->fid = opinfo->fid;
-	br_info->open_trunc = opinfo->open_trunc;
+	list_for_each_entry(channel, head, list) {
+		if (!channel->queued && !channel->failed)
+			return true;
+	}
+	return false;
+}
 
-	work->request_buf = (char *)br_info;
-	/* Transfer the reference acquired by smb2_oplock_break_conn_get(). */
-	work->conn = conn;
-	work->sess = opinfo->sess;
+static int smb2_break_wait(struct oplock_info *opinfo, struct list_head *head,
+			   struct smb2_break_channel *sent_channel,
+			   bool expect_ack, bool wait_ack, bool can_retry,
+			   unsigned long retry_at, unsigned long deadline)
+{
+	unsigned long remaining;
+	long ret;
 
-	ksmbd_conn_r_count_inc(conn);
-	if (opinfo->op_state == OPLOCK_ACK_WAIT) {
-		INIT_WORK(&work->work, __smb2_oplock_break_noti);
-		ksmbd_queue_work(work);
+	for (;;) {
+		if (smb2_break_complete(opinfo, expect_ack))
+			return 0;
+		if (!wait_ack && smb2_break_delivered(head, false))
+			return 0;
+		if (time_after_eq(jiffies, deadline))
+			return -ETIMEDOUT;
+		/* TCP delivery followed by a slow SMB ACK is not a send failure. */
+		if (can_retry && !smb2_break_delivered(head, true) &&
+		    ((sent_channel && !ksmbd_conn_good(sent_channel->conn)) ||
+		     time_after_eq(jiffies, retry_at)))
+			return -EAGAIN;
 
-		if (wait_for_break_ack(opinfo))
-			ret = ksmbd_invalidate_durable_fd(opinfo->fid);
-	} else {
-		__smb2_oplock_break_noti(&work->work);
-		if (opinfo->level == SMB2_OPLOCK_LEVEL_II)
-			opinfo->level = SMB2_OPLOCK_LEVEL_NONE;
+		remaining = min_t(unsigned long, deadline - jiffies,
+				  max_t(unsigned long, 1, HZ / 10));
+		ret = wait_event_interruptible_timeout(opinfo->oplock_q,
+						       smb2_break_complete(opinfo, expect_ack),
+						       remaining);
+		if (ret < 0)
+			return ret;
 	}
-	return ret;
 }
 
-/**
- * __smb2_lease_break_noti() - send lease break command from server
- * to client
- * @wk:     smb work object
- */
-static void __smb2_lease_break_noti(struct work_struct *wk)
+static void smb2_break_init_header(struct smb2_hdr *hdr)
 {
-	struct smb2_lease_break *rsp = NULL;
-	struct ksmbd_work *work = container_of(wk, struct ksmbd_work, work);
-	struct ksmbd_conn *conn = work->conn;
-	struct lease_break_info *br_info = work->request_buf;
-	struct smb2_hdr *rsp_hdr;
+	memset(hdr, 0, sizeof(*hdr));
+	hdr->ProtocolId = SMB2_PROTO_NUMBER;
+	hdr->StructureSize = SMB2_HEADER_STRUCTURE_SIZE;
+	hdr->Command = SMB2_OPLOCK_BREAK;
+	hdr->Flags = SMB2_FLAGS_SERVER_TO_REDIR;
+	hdr->MessageId = cpu_to_le64(-1);
+}
 
-	if (allocate_interim_rsp_buf(work)) {
-		ksmbd_debug(OPLOCK, "smb2_allocate_rsp_buf failed! ");
-		goto out;
+/* Build one notification and reuse its payload and epoch on each connection. */
+static int smb2_break_notify(struct oplock_info *opinfo, struct ksmbd_inode *ci,
+			     bool expect_ack, bool wait_ack, bool inc_epoch,
+			     struct work_struct *interim_work)
+{
+	struct smb2_break_channel owner = {}, *channel, *tmp, *last = NULL;
+	struct ksmbd_work *work;
+	struct smb2_oplock_break *oplock_rsp;
+	struct smb2_lease_break *lease_rsp;
+	struct lease *lease = opinfo->o_lease;
+	unsigned long deadline, retry_at;
+	int ret, size;
+	LIST_HEAD(channels);
+
+	if (smb2_break_complete(opinfo, expect_ack)) {
+		ret = 0;
+		goto out_interim;
+	}
+	wait_ack = wait_ack && expect_ack;
+	work = ksmbd_alloc_work_struct();
+	if (!work) {
+		ret = smb2_break_failed(opinfo, expect_ack);
+		goto out_interim;
 	}
+	ret = allocate_interim_rsp_buf(work);
+	if (ret)
+		goto failed;
+	oplock_rsp = smb_get_msg(work->response_buf);
+	smb2_break_init_header(&oplock_rsp->hdr);
 
-	rsp_hdr = smb_get_msg(work->response_buf);
-	memset(rsp_hdr, 0, sizeof(struct smb2_hdr) + 2);
-	rsp_hdr->ProtocolId = SMB2_PROTO_NUMBER;
-	rsp_hdr->StructureSize = SMB2_HEADER_STRUCTURE_SIZE;
-	rsp_hdr->CreditRequest = cpu_to_le16(0);
-	rsp_hdr->Command = SMB2_OPLOCK_BREAK;
-	rsp_hdr->Flags = (SMB2_FLAGS_SERVER_TO_REDIR);
-	rsp_hdr->NextCommand = 0;
-	rsp_hdr->MessageId = cpu_to_le64(-1);
-	rsp_hdr->Id.SyncId.ProcessId = 0;
-	rsp_hdr->Id.SyncId.TreeId = 0;
-	rsp_hdr->SessionId = 0;
-	memset(rsp_hdr->Signature, 0, 16);
-
-	rsp = smb_get_msg(work->response_buf);
-	rsp->StructureSize = cpu_to_le16(44);
-	rsp->Epoch = br_info->epoch;
-	rsp->Flags = 0;
-
-	if (br_info->curr_state & (SMB2_LEASE_WRITE_CACHING_LE |
-			SMB2_LEASE_HANDLE_CACHING_LE))
-		rsp->Flags = SMB2_NOTIFY_BREAK_LEASE_FLAG_ACK_REQUIRED;
-
-	memcpy(rsp->LeaseKey, br_info->lease_key, SMB2_LEASE_KEY_SIZE);
-	rsp->CurrentLeaseState = br_info->curr_state;
-	rsp->NewLeaseState = br_info->new_state;
-	rsp->BreakReason = 0;
-	rsp->AccessMaskHint = 0;
-	rsp->ShareMaskHint = 0;
-
-	if (ksmbd_iov_pin_rsp(work, (void *)rsp,
-			      sizeof(struct smb2_lease_break)))
+	ret = smb2_break_collect_channels(opinfo, ci, &owner, &channels, oplock_rsp);
+	if (ret == -ENOENT)
 		goto out;
+	if (ret)
+		goto failed;
 
-	ksmbd_conn_write(work);
-
+	spin_lock(&opinfo->state_lock);
+	if (opinfo->op_state == OPLOCK_CLOSING ||
+	    (expect_ack && opinfo->op_state == OPLOCK_STATE_NONE)) {
+		spin_unlock(&opinfo->state_lock);
+		ret = 0;
+		goto out;
+	}
+	if (opinfo->is_lease) {
+		lease_rsp = smb_get_msg(work->response_buf);
+		lease_rsp->StructureSize = cpu_to_le16(44);
+		if (lease->version == 2) {
+			if (inc_epoch)
+				lease->epoch++;
+			lease_rsp->Epoch = cpu_to_le16(lease->epoch);
+		}
+		if (lease->state & (SMB2_LEASE_WRITE_CACHING_LE |
+				    SMB2_LEASE_HANDLE_CACHING_LE))
+			lease_rsp->Flags = SMB2_NOTIFY_BREAK_LEASE_FLAG_ACK_REQUIRED;
+		memcpy(lease_rsp->LeaseKey, lease->lease_key, SMB2_LEASE_KEY_SIZE);
+		lease_rsp->CurrentLeaseState = lease->state;
+		lease_rsp->NewLeaseState = lease->new_state;
+		size = sizeof(*lease_rsp);
+	} else {
+		oplock_rsp->StructureSize = cpu_to_le16(24);
+		oplock_rsp->OplockLevel = !opinfo->open_trunc &&
+			(opinfo->level == SMB2_OPLOCK_LEVEL_BATCH ||
+			 opinfo->level == SMB2_OPLOCK_LEVEL_EXCLUSIVE) ?
+			SMB2_OPLOCK_LEVEL_II : SMB2_OPLOCK_LEVEL_NONE;
+		/*
+		 * A level II break needs no ACK. Publish NONE before sending so
+		 * an unsolicited ACK cannot observe II while delivery is pending.
+		 */
+		if (!expect_ack)
+			opinfo->level = SMB2_OPLOCK_LEVEL_NONE;
+		size = sizeof(*oplock_rsp);
+	}
+	spin_unlock(&opinfo->state_lock);
+	ret = ksmbd_iov_pin_rsp(work, oplock_rsp, size);
+	if (ret)
+		goto failed;
+
+	deadline = jiffies + OPLOCK_WAIT_TIME;
+retry:
+	list_for_each_entry(channel, &channels, list) {
+		if (channel->queued || channel->failed)
+			continue;
+		if (smb2_break_complete(opinfo, expect_ack)) {
+			ret = 0;
+			goto out;
+		}
+		if (time_after_eq(jiffies, deadline)) {
+			ret = -ETIMEDOUT;
+			break;
+		}
+		if (!ksmbd_conn_r_count_inc_if_good(channel->conn)) {
+			channel->failed = true;
+			ret = -ENOTCONN;
+			continue;
+		}
+		work->conn = channel->conn;
+		ret = ksmbd_conn_write_tracked(work, &channel->tracking);
+		ksmbd_conn_r_count_dec(channel->conn);
+		if (ret) {
+			/* Only a nonblocking send guarantees no bytes on -EAGAIN. */
+			if (ret != -EAGAIN || !channel->conn->transport->ops->is_acked)
+				channel->failed = true;
+			continue;
+		}
+		channel->queued = true;
+		if (interim_work) {
+			ksmbd_queue_io_work(interim_work);
+			interim_work = NULL;
+		}
+		last = channel;
+		retry_at = jiffies + channel->tracking.retry_after;
+		ret = smb2_break_wait(opinfo, &channels, channel, expect_ack,
+				      wait_ack, smb2_break_has_unsent_channels(&channels),
+				      retry_at, deadline);
+		if (ret != -EAGAIN)
+			goto done;
+	}
+	if (smb2_break_has_unsent_channels(&channels)) {
+		/* Allow busy writers to make progress before another nonblocking pass. */
+		retry_at = jiffies + max_t(unsigned long, 1, HZ / 10);
+		ret = smb2_break_wait(opinfo, &channels, NULL, expect_ack,
+				      wait_ack, true, retry_at, deadline);
+		if (ret == -EAGAIN)
+			goto retry;
+	} else if (last) {
+		/* Give already queued notifications the rest of the shared wait. */
+		ret = smb2_break_wait(opinfo, &channels, last, expect_ack,
+				      wait_ack, false, deadline, deadline);
+	}
+
+done:
+	if (ret)
+		goto failed;
+	if (!expect_ack && opinfo->is_lease) {
+		spin_lock(&opinfo->state_lock);
+		if (opinfo->op_state != OPLOCK_CLOSING) {
+			lease->state = lease->new_state;
+			lease_update_oplock_levels(lease);
+		}
+		spin_unlock(&opinfo->state_lock);
+	}
+	goto out;
+failed:
+	ret = smb2_break_failed(opinfo, expect_ack);
 out:
 	ksmbd_free_work_struct(work);
-	ksmbd_conn_r_count_dec(conn);
-	ksmbd_conn_put(conn);
+	list_for_each_entry_safe(channel, tmp, &channels, list) {
+		list_del(&channel->list);
+		ksmbd_conn_put(channel->conn);
+		if (channel != &owner)
+			kfree(channel);
+	}
+out_interim:
+	if (interim_work)
+		ksmbd_queue_io_work(interim_work);
+	return ret;
 }
 
-/*
- * Select and pin the connection used for a lease break before doing any
- * allocations which may sleep. opinfo->conn is cleared under ci->m_lock,
- * while lease->l_lb and the lease table lifetime are protected by
- * lease_list_lock.
- */
-static struct ksmbd_conn *smb2_lease_break_conn_get(struct oplock_info *opinfo)
+struct smb2_interim_work {
+	struct work_struct work;
+	struct ksmbd_work *in_work;
+};
+
+static void smb2_break_interim_work(struct work_struct *work)
 {
-	struct lease *lease = opinfo->o_lease;
-	struct lease_table *lb;
-	struct ksmbd_conn *conn;
+	struct smb2_interim_work *iw = container_of(work, struct smb2_interim_work, work);
 
-	/* Keep the connection which owns the open, when it is still active. */
-	down_read(&lease->ci->m_lock);
-	conn = READ_ONCE(opinfo->conn);
-	if (conn && !ksmbd_conn_releasing(conn))
-		conn = ksmbd_conn_get(conn);
-	else
-		conn = NULL;
-	up_read(&lease->ci->m_lock);
+	if (!setup_async_work(iw->in_work, NULL, NULL)) {
+		smb2_send_interim_resp(iw->in_work, STATUS_PENDING);
+		release_async_work(iw->in_work);
+	}
+}
 
-	if (conn || lease->version != 2)
-		return conn;
+struct smb2_break_work {
+	struct work_struct work;
+	struct oplock_info *opinfo;
+	struct ksmbd_inode *ci;
+	bool expect_ack;
+	bool wait_ack;
+	bool inc_epoch;
+	int ret;
+};
 
-	/* Otherwise route v2 lease breaks through the shared lease channel. */
-	read_lock(&lease_list_lock);
-	lb = lease->l_lb;
-	if (lb && lb->conn && !ksmbd_conn_releasing(lb->conn))
-		conn = ksmbd_conn_get(lb->conn);
-	read_unlock(&lease_list_lock);
+static void smb2_break_notify_work(struct work_struct *work)
+{
+	struct smb2_break_work *bw = container_of(work, struct smb2_break_work, work);
 
-	return conn;
+	bw->ret = smb2_break_notify(bw->opinfo, bw->ci, bw->expect_ack,
+				   bw->wait_ack, bw->inc_epoch, NULL);
 }
 
-/**
- * smb2_lease_break_noti() - break lease when a new client request
- *			write lease
- * @opinfo:		contains lease state information
- * @sync:		send the lease break notification synchronously
- * @inc_epoch:		increment the lease epoch before sending the break
- *
- * Return:	0 on success, otherwise error
- */
-static int smb2_lease_break_noti(struct oplock_info *opinfo, bool sync,
-				 bool inc_epoch)
+static int smb2_break_notify_with_interim(struct oplock_info *opinfo,
+					struct ksmbd_inode *ci,
+					bool expect_ack, bool wait_ack,
+					bool inc_epoch, struct ksmbd_work *in_work,
+					bool notify_first)
 {
-	struct ksmbd_conn *conn;
-	struct ksmbd_work *work;
-	struct lease_break_info *br_info;
-	struct lease *lease = opinfo->o_lease;
+	struct smb2_break_work bw = {
+		.opinfo = opinfo,
+		.ci = ci,
+		.expect_ack = expect_ack,
+		.wait_ack = wait_ack,
+		.inc_epoch = inc_epoch,
+	};
 
-	conn = smb2_lease_break_conn_get(opinfo);
-	if (!conn)
-		return ksmbd_invalidate_durable_fd(opinfo->fid);
+	if (!in_work)
+		return smb2_break_notify(opinfo, ci, expect_ack, wait_ack, inc_epoch, NULL);
 
-	work = ksmbd_alloc_work_struct();
-	if (!work) {
-		ksmbd_conn_put(conn);
-		return -ENOMEM;
-	}
+	if (notify_first) {
+		struct smb2_interim_work iw = { .in_work = in_work };
+		int ret;
 
-	br_info = kmalloc_obj(struct lease_break_info, KSMBD_DEFAULT_GFP);
-	if (!br_info) {
-		ksmbd_free_work_struct(work);
-		ksmbd_conn_put(conn);
-		return -ENOMEM;
-	}
-
-	br_info->curr_state = lease->state;
-	br_info->new_state = lease->new_state;
-	if (lease->version == 2) {
-		if (inc_epoch)
-			lease->epoch++;
-		br_info->epoch = cpu_to_le16(lease->epoch);
-	} else {
-		br_info->epoch = 0;
+		/*
+		 * Rename sends the first notification in the request context.
+		 * Queue the interim writer after that send so a blocked response
+		 * cannot stop notification retries or the break ACK wait.
+		 */
+		INIT_WORK_ONSTACK(&iw.work, smb2_break_interim_work);
+		ret = smb2_break_notify(opinfo, ci, expect_ack, wait_ack,
+					inc_epoch, &iw.work);
+		flush_work(&iw.work);
+		destroy_work_on_stack(&iw.work);
+		return ret;
 	}
-	memcpy(br_info->lease_key, lease->lease_key, SMB2_LEASE_KEY_SIZE);
 
-	work->request_buf = (char *)br_info;
-	/* Transfer the reference acquired by smb2_lease_break_conn_get(). */
-	work->conn = conn;
-	work->sess = opinfo->sess;
-
-	ksmbd_conn_r_count_inc(conn);
-	if (opinfo->op_state == OPLOCK_ACK_WAIT) {
-		if (sync) {
-			__smb2_lease_break_noti(&work->work);
-		} else {
-			INIT_WORK(&work->work, __smb2_lease_break_noti);
-			ksmbd_queue_work(work);
-		}
-	} else {
-		__smb2_lease_break_noti(&work->work);
-		if (opinfo->o_lease->new_state == SMB2_LEASE_NONE_LE) {
-			opinfo->o_lease->state = SMB2_LEASE_NONE_LE;
-			lease_update_oplock_levels(opinfo->o_lease);
-		}
+	/* A blocked interim writer must not stop notification sends or retries. */
+	INIT_WORK_ONSTACK(&bw.work, smb2_break_notify_work);
+	ksmbd_queue_break_work(&bw.work);
+	if (!setup_async_work(in_work, NULL, NULL)) {
+		smb2_send_interim_resp(in_work, STATUS_PENDING);
+		release_async_work(in_work);
 	}
-	return 0;
+	/* The caller keeps opinfo and ci alive until the notifier has finished. */
+	flush_work(&bw.work);
+	destroy_work_on_stack(&bw.work);
+	return bw.ret;
 }
 
 static void wait_lease_breaking(struct oplock_info *opinfo)
@@ -1187,10 +1323,11 @@ static void wait_lease_breaking(struct oplock_info *opinfo)
 
 static int oplock_break(struct oplock_info *brk_opinfo, struct ksmbd_inode *ci,
 			int req_op_level, struct ksmbd_work *in_work,
-			bool share_break, bool sync_lease_break)
+			bool share_break, bool notify_first)
 {
 	int err = 0;
 	bool sent_interim = false;
+	bool expect_ack = false;
 
 	/* Need to break exclusive/batch oplock, write lease or overwrite_if */
 	ksmbd_debug(OPLOCK,
@@ -1248,8 +1385,9 @@ static int oplock_break(struct oplock_info *brk_opinfo, struct ksmbd_inode *ci,
 			}
 		}
 
-		if (lease->state & (SMB2_LEASE_WRITE_CACHING_LE |
-				SMB2_LEASE_HANDLE_CACHING_LE)) {
+		expect_ack = lease->state & (SMB2_LEASE_WRITE_CACHING_LE |
+					     SMB2_LEASE_HANDLE_CACHING_LE);
+		if (expect_ack) {
 			if (!oplock_break_set_ack_wait(brk_opinfo)) {
 				atomic_dec_if_positive(&brk_opinfo->breaking_cnt);
 				clear_and_wake_up_bit(0, &brk_opinfo->pending_break);
@@ -1265,16 +1403,12 @@ static int oplock_break(struct oplock_info *brk_opinfo, struct ksmbd_inode *ci,
 			inc_epoch = false;
 			lease->reuse_epoch = false;
 		}
-		err = smb2_lease_break_noti(brk_opinfo, sync_lease_break, inc_epoch);
+		err = smb2_break_notify_with_interim(brk_opinfo, ci, expect_ack,
+						    wait_ack, inc_epoch,
+						    sent_interim ? NULL : in_work,
+						    notify_first);
+		sent_interim = true;
 		inc_epoch = false;
-		if (in_work && !sent_interim) {
-			setup_async_work(in_work, NULL, NULL);
-			smb2_send_interim_resp(in_work, STATUS_PENDING);
-			release_async_work(in_work);
-			sent_interim = true;
-		}
-		if (wait_ack && !err && wait_for_break_ack(brk_opinfo))
-			err = ksmbd_invalidate_durable_fd(brk_opinfo->fid);
 
 		ksmbd_debug(OPLOCK, "oplock granted = %d\n", brk_opinfo->level);
 		if (brk_opinfo->op_state == OPLOCK_CLOSING)
@@ -1310,26 +1444,16 @@ static int oplock_break(struct oplock_info *brk_opinfo, struct ksmbd_inode *ci,
 
 		if (brk_opinfo->level == SMB2_OPLOCK_LEVEL_BATCH ||
 		    brk_opinfo->level == SMB2_OPLOCK_LEVEL_EXCLUSIVE) {
+			expect_ack = true;
 			if (!oplock_break_set_ack_wait(brk_opinfo)) {
 				clear_and_wake_up_bit(0, &brk_opinfo->pending_break);
 				return -ENOENT;
 			}
 		}
-
-		/*
-		 * Keep a conflicting CREATE asynchronous while waiting for an
-		 * oplock-break acknowledgement.  Besides avoiding a blocked client
-		 * request, this lets a replay arrive while the original CREATE is
-		 * still pending and be rejected with FILE_NOT_AVAILABLE.
-		 */
-		if (in_work) {
-			setup_async_work(in_work, NULL, NULL);
-			smb2_send_interim_resp(in_work, STATUS_PENDING);
-			release_async_work(in_work);
-		}
 	}
 
-	err = smb2_oplock_break_noti(brk_opinfo, ci);
+	err = smb2_break_notify_with_interim(brk_opinfo, ci, expect_ack,
+					    true, false, in_work, false);
 
 	ksmbd_debug(OPLOCK, "oplock granted = %d\n", brk_opinfo->level);
 	if (brk_opinfo->op_state == OPLOCK_CLOSING)
@@ -1491,7 +1615,7 @@ void smb_send_parent_lease_break_noti(struct ksmbd_file *fp,
 
 	down_read(&p_ci->m_lock);
 	list_for_each_entry(opinfo, &p_ci->m_op_list, op_entry) {
-		if (opinfo->conn == NULL || !opinfo->is_lease)
+		if (!opinfo->is_lease)
 			continue;
 
 		if (opinfo->o_lease->state != SMB2_OPLOCK_LEVEL_NONE &&
@@ -1502,11 +1626,6 @@ void smb_send_parent_lease_break_noti(struct ksmbd_file *fp,
 			if (!atomic_inc_not_zero(&opinfo->refcount))
 				continue;
 
-			if (ksmbd_conn_releasing(opinfo->conn)) {
-				opinfo_put(opinfo);
-				continue;
-			}
-
 			if (oplock_break_add(&brk_list, opinfo))
 				opinfo_put(opinfo);
 		}
@@ -1539,18 +1658,13 @@ void smb_lazy_parent_lease_break_close(struct ksmbd_file *fp)
 
 	down_read(&p_ci->m_lock);
 	list_for_each_entry(opinfo, &p_ci->m_op_list, op_entry) {
-		if (opinfo->conn == NULL || !opinfo->is_lease)
+		if (!opinfo->is_lease)
 			continue;
 
 		if (opinfo->o_lease->state != SMB2_OPLOCK_LEVEL_NONE) {
 			if (!atomic_inc_not_zero(&opinfo->refcount))
 				continue;
 
-			if (ksmbd_conn_releasing(opinfo->conn)) {
-				opinfo_put(opinfo);
-				continue;
-			}
-
 			if (oplock_break_add(&brk_list, opinfo))
 				opinfo_put(opinfo);
 		}
@@ -1699,7 +1813,11 @@ int smb_grant_oplock(struct ksmbd_work *work, int req_op_level, u64 pid,
 
 	err = oplock_break(prev_opinfo, ci, break_level, work,
 			   share_ret < 0 && prev_opinfo->is_lease, false);
-	if (prev_durable_detached || (prev_durable_open && err == -ENOENT))
+	/* Keep a detached durable lease if its break retains handle caching. */
+	if ((prev_durable_detached &&
+	     (!prev_op_has_lease ||
+	      !(prev_opinfo->o_lease->state & SMB2_LEASE_HANDLE_CACHING_LE))) ||
+	    (prev_durable_open && err == -ENOENT))
 		ksmbd_invalidate_durable_fd(prev_fid);
 	opinfo_put(prev_opinfo);
 	if (err == -EAGAIN) {
@@ -1803,8 +1921,7 @@ static bool smb_break_all_write_oplock(struct ksmbd_work *work,
 	}
 
 	brk_opinfo->open_trunc = is_trunc;
-	oplock_break(brk_opinfo, fp->f_ci, SMB2_OPLOCK_LEVEL_II, work, false,
-		     false);
+	oplock_break(brk_opinfo, fp->f_ci, SMB2_OPLOCK_LEVEL_II, work, false, false);
 	sent_break = true;
 	opinfo_put(brk_opinfo);
 
@@ -1819,12 +1936,12 @@ static bool smb_break_all_write_oplock(struct ksmbd_work *work,
  * @is_trunc:		truncate on open
  * @send_interim:	send interim response to the client
  * @send_oplock_break:	send oplock break notification to the client
- * @sync_lease_break:	send the lease break notification synchronously
+ * @notify_first:	send the lease break notification before the interim response
  */
 static void __smb_break_all_levII_oplock(struct ksmbd_work *work,
 					 struct ksmbd_file *fp, int is_trunc,
 					 bool send_interim, bool send_oplock_break,
-					 bool sync_lease_break)
+					 bool notify_first)
 {
 	struct oplock_info *op, *brk_op;
 	struct oplock_break_entry *ent, *tmp;
@@ -1842,17 +1959,9 @@ static void __smb_break_all_levII_oplock(struct ksmbd_work *work,
 
 	down_read(&ci->m_lock);
 	list_for_each_entry(brk_op, &ci->m_op_list, op_entry) {
-		if (brk_op->conn == NULL)
-			continue;
-
 		if (!atomic_inc_not_zero(&brk_op->refcount))
 			continue;
 
-		if (ksmbd_conn_releasing(brk_op->conn)) {
-			opinfo_put(brk_op);
-			continue;
-		}
-
 		if (!brk_op->is_lease &&
 		    brk_op->level != SMB2_OPLOCK_LEVEL_II) {
 			ksmbd_debug(OPLOCK, "unexpected oplock(0x%x)\n",
@@ -1867,10 +1976,8 @@ static void __smb_break_all_levII_oplock(struct ksmbd_work *work,
 			goto next;
 
 		if (op && op->is_lease && brk_op->is_lease &&
-		    !memcmp(conn->ClientGUID, brk_op->conn->ClientGUID,
-			    SMB2_CLIENT_GUID_SIZE) &&
-		    !memcmp(op->o_lease->lease_key, brk_op->o_lease->lease_key,
-			    SMB2_LEASE_KEY_SIZE))
+		    compare_guid_key(brk_op, conn->ClientGUID,
+				     op->o_lease->lease_key))
 			goto next;
 		brk_op->open_trunc = is_trunc;
 
@@ -1900,7 +2007,7 @@ static void __smb_break_all_levII_oplock(struct ksmbd_work *work,
 				     brk_op->is_lease && !is_trunc ?
 				     SMB2_OPLOCK_LEVEL_II : SMB2_OPLOCK_LEVEL_NONE,
 				     send_interim && !sent_interim ? work : NULL,
-				     false, sync_lease_break);
+				     false, notify_first);
 		}
 		sent_interim = true;
 		list_del(&ent->list);
diff --git a/fs/smb/server/oplock.h b/fs/smb/server/oplock.h
index b08d21758e07..e114e2148c30 100644
--- a/fs/smb/server/oplock.h
+++ b/fs/smb/server/oplock.h
@@ -62,6 +62,7 @@ struct lease {
 struct oplock_info {
 	struct ksmbd_conn	*conn;
 	struct ksmbd_session	*sess;
+	u64			session_id;
 	struct ksmbd_work	*work;
 	struct ksmbd_file	*o_fp;
 	int                     level;
@@ -82,19 +83,6 @@ struct oplock_info {
 	struct rcu_head		rcu;
 };
 
-struct lease_break_info {
-	__le32			curr_state;
-	__le32			new_state;
-	__le16			epoch;
-	char			lease_key[SMB2_LEASE_KEY_SIZE];
-};
-
-struct oplock_break_info {
-	int level;
-	int open_trunc;
-	int fid;
-};
-
 int smb_grant_oplock(struct ksmbd_work *work, int req_op_level,
 		     u64 pid, struct ksmbd_file *fp, __u16 tid,
 		     struct lease_ctx_info *lctx, int share_ret, bool replay);
diff --git a/fs/smb/server/vfs_cache.c b/fs/smb/server/vfs_cache.c
index 496055ce4a4d..11c123673f9a 100644
--- a/fs/smb/server/vfs_cache.c
+++ b/fs/smb/server/vfs_cache.c
@@ -1767,6 +1767,7 @@ static bool session_fd_check(struct ksmbd_tree_connect *tcon,
 		ksmbd_conn_put(op->conn);
 		op->conn = NULL;
 		op->sess = NULL;
+		op->session_id = 0;
 	}
 	up_write(&ci->m_lock);
 
@@ -1953,6 +1954,7 @@ int ksmbd_reopen_durable_fd(struct ksmbd_work *work, struct ksmbd_file *fp)
 			continue;
 		op->conn = ksmbd_conn_get(fp->conn);
 		op->sess = work->sess;
+		op->session_id = work->sess->id;
 	}
 	up_write(&ci->m_lock);
 
-- 
2.25.1