diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md
index 2bf74c5f8..b1c63bb22 100644
--- a/PendingReleaseNotes.md
+++ b/PendingReleaseNotes.md
@@ -4,9 +4,9 @@
 
 - Rook now validates node topology during CephCluster creation to prevent misconfigured CRUSH hierarchies. If child labels like `topology.rook.io/rack` are duplicated across zones, cluster creation will fail. The check applies only to new clusters without OSDs; existing clusters will log a warning and continue. To bypass, set `ROOK_SKIP_OSD_TOPOLOGY_CHECK=true` in the operator configmap. See [#16017](https://github.com/rook/rook/pull/16017) for details.
 
-
 ## Features
 
-
 - Previously, only the latest version of helm was tested and the docs stated only version 3.x of helm as a prerequisite. Now rook supports the six most recent minor versions of helm along with their their patch updates. Explicitly, helm versions 3.13 and newer are supported.
 - Add support for specifying the clusterID in the CephBlockPoolRadosNamespace and the CephFilesystemSubVolumeGroup CR.
+- If a mon is being failed over, if the assigned node no longer exists, failover immediately instead of waiting for a
+  20 minute timeout.
diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go
index 44142ac55..056849d75 100644
--- a/pkg/operator/ceph/cluster/mon/health.go
+++ b/pkg/operator/ceph/cluster/mon/health.go
@@ -273,33 +273,39 @@ func (c *Cluster) checkHealth(ctx context.Context) error {
 		}
 
 		// If not yet set, add the current time, for the timeout
-		// calculation, to the list
+		// calculation, to the list.
+		rescheduleImmediately := false
 		if _, ok := c.monTimeoutList[mon.Name]; !ok {
 			c.monTimeoutList[mon.Name] = time.Now()
+
+			// If the assigned node is not found, skip the timeout wait
+			rescheduleImmediately = c.shouldFailoverMonImmediately(ctx, mon.Name)
 		}
 
 		// when the timeout for the mon has been reached, continue to the
 		// normal failover mon pod part of the code
-		if time.Since(c.monTimeoutList[mon.Name]) <= MonOutTimeout {
-			timeToFailover := int(MonOutTimeout.Seconds() - time.Since(c.monTimeoutList[mon.Name]).Seconds())
-			logger.Warningf("mon %q not found in quorum, waiting for timeout (%d seconds left) before failover", mon.Name, timeToFailover)
-			continue
+		if !rescheduleImmediately {
+			if time.Since(c.monTimeoutList[mon.Name]) <= MonOutTimeout {
+				timeToFailover := int(MonOutTimeout.Seconds() - time.Since(c.monTimeoutList[mon.Name]).Seconds())
+				logger.Warningf("mon %q not found in quorum, waiting for timeout (%d seconds left) before failover", mon.Name, timeToFailover)
+				continue
+			}
+
+			// retry only once before the mon failover if the mon pod is not scheduled
+			monLabelSelector := fmt.Sprintf("%s=%s,%s=%s", k8sutil.AppAttr, AppName, controller.DaemonIDLabel, mon.Name)
+			isScheduled, err := k8sutil.IsPodScheduled(ctx, c.context.Clientset, c.Namespace, monLabelSelector)
+			if err != nil {
+				logger.Warningf("failed to check if mon %q is assigned to a node, continuing with mon failover. %v", mon.Name, err)
+			} else if !isScheduled && retriesBeforeNodeDrainFailover > 0 {
+				logger.Warningf("mon %q NOT found in quorum after timeout. Mon pod is not scheduled. Retrying with a timeout of %.2f seconds before failover", mon.Name, MonOutTimeout.Seconds())
+				delete(c.monTimeoutList, mon.Name)
+				retriesBeforeNodeDrainFailover = retriesBeforeNodeDrainFailover - 1
+				return nil
+			}
+			retriesBeforeNodeDrainFailover = 1
+			logger.Warningf("mon %q NOT found in quorum and timeout exceeded, mon will be failed over", mon.Name)
 		}
 
-		// retry only once before the mon failover if the mon pod is not scheduled
-		monLabelSelector := fmt.Sprintf("%s=%s,%s=%s", k8sutil.AppAttr, AppName, controller.DaemonIDLabel, mon.Name)
-		isScheduled, err := k8sutil.IsPodScheduled(ctx, c.context.Clientset, c.Namespace, monLabelSelector)
-		if err != nil {
-			logger.Warningf("failed to check if mon %q is assigned to a node, continuing with mon failover. %v", mon.Name, err)
-		} else if !isScheduled && retriesBeforeNodeDrainFailover > 0 {
-			logger.Warningf("mon %q NOT found in quorum after timeout. Mon pod is not scheduled. Retrying with a timeout of %.2f seconds before failover", mon.Name, MonOutTimeout.Seconds())
-			delete(c.monTimeoutList, mon.Name)
-			retriesBeforeNodeDrainFailover = retriesBeforeNodeDrainFailover - 1
-			return nil
-		}
-		retriesBeforeNodeDrainFailover = 1
-
-		logger.Warningf("mon %q NOT found in quorum and timeout exceeded, mon will be failed over", mon.Name)
 		if !c.failMon(len(quorumStatus.MonMap.Mons), desiredMonCount, mon.Name) {
 			// The failover was skipped, so we continue to see if another mon needs to failover
 			continue
@@ -383,6 +389,28 @@ func (c *Cluster) checkHealth(ctx context.Context) error {
 	return nil
 }
 
+func (c *Cluster) shouldFailoverMonImmediately(ctx context.Context, monName string) bool {
+	// If the assigned node does not exist, reschedule immediately
+	logger.Debugf("checking if mon %q is scheduled on a node", monName)
+	assignedNode := ""
+	if mapping, ok := c.mapping.Schedule[monName]; ok {
+		assignedNode = mapping.Hostname
+	}
+	if assignedNode == "" {
+		logger.Debugf("mon %q is not assigned to a node", monName)
+		return false
+	}
+
+	logger.Debugf("mon %q is scheduled on node %q", monName, assignedNode)
+	nodeExists, err := k8sutil.NodeWithHostnameExists(ctx, c.context.Clientset, assignedNode)
+	if err == nil && !nodeExists {
+		logger.Warningf("mon %q NOT found in quorum and its assigned node %q is not found, failing over immediately", monName, assignedNode)
+		return true
+	}
+
+	return false
+}
+
 // reconcileExternalMons handling external monitors defined in CephCluster.spec.mon.externalMonIDs when Rook managing local cluster.
 func (c *Cluster) reconcileExternalMons(ctx context.Context, quorumStatus cephclient.MonStatusResponse) (cephclient.MonStatusResponse, error) {
 	if len(c.spec.Mon.ExternalMonIDs) != 0 {
diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go
index 3e462a225..4e92de553 100644
--- a/pkg/operator/ceph/cluster/mon/health_test.go
+++ b/pkg/operator/ceph/cluster/mon/health_test.go
@@ -206,6 +206,32 @@ func TestRemoveExtraMon(t *testing.T) {
 	}
 }
 
+func TestScheduleFailoverImmediately(t *testing.T) {
+	clientset := test.New(t, 1)
+	c := &Cluster{mapping: &opcontroller.Mapping{}, context: &clusterd.Context{Clientset: clientset}}
+	c.mapping.Schedule = map[string]*opcontroller.MonScheduleInfo{
+		"a": {Name: "a", Hostname: "nodea"},
+		"b": {Name: "b"},
+		"c": {Name: "c", Hostname: "nodec"},
+	}
+
+	node := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "a",
+			Labels: map[string]string{"kubernetes.io/hostname": "nodea"},
+		},
+	}
+	_, err := clientset.CoreV1().Nodes().Create(context.TODO(), node, metav1.CreateOptions{})
+	assert.NoError(t, err)
+
+	// mon a is assigned to a node, so it should not failover immediately
+	assert.False(t, c.shouldFailoverMonImmediately((context.TODO()), "a"))
+	// mon b is not assigned to a node, so it should not failover immediately
+	assert.False(t, c.shouldFailoverMonImmediately((context.TODO()), "b"))
+	// mon c is assigned to a non-existent node, so it should failover immediately
+	assert.True(t, c.shouldFailoverMonImmediately((context.TODO()), "c"))
+}
+
 func TestTrackMonsOutOfQuorum(t *testing.T) {
 	endpoint := "1.2.3.4:6789"
 	clientset := test.New(t, 1)
diff --git a/pkg/operator/k8sutil/node.go b/pkg/operator/k8sutil/node.go
index 07b3f660e..f903e844b 100644
--- a/pkg/operator/k8sutil/node.go
+++ b/pkg/operator/k8sutil/node.go
@@ -20,10 +20,11 @@ package k8sutil
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"strings"
 
+	"github.com/pkg/errors"
+
 	"sigs.k8s.io/yaml"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -97,6 +98,16 @@ func GetValidNodes(ctx context.Context, rookStorage cephv1.StorageScopeSpec, cli
 	return RookNodesMatchingKubernetesNodes(rookStorage, validK8sNodes)
 }
 
+func NodeWithHostnameExists(ctx context.Context, clientset kubernetes.Interface, hostName string) (bool, error) {
+	options := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", LabelHostname(), hostName)}
+	nodes, err := clientset.CoreV1().Nodes().List(ctx, options)
+	if err != nil {
+		return false, errors.Wrapf(err, "failed to query node %q", hostName)
+	}
+	logger.Debugf("found %d nodes with hostname %q", len(nodes.Items), hostName)
+	return len(nodes.Items) > 0, nil
+}
+
 // GetNodeNameFromHostname returns the name of the node resource looked up by the hostname label
 // Typically these will be the same name, but sometimes they are not such as when nodes have a longer
 // dns name, but the hostname is short.