fix(server): evict recovering engine frontends on Create

When EngineFrontendCreate encounters an existing frontend in Pending
state (still being recovered asynchronously), evict it instead of
returning AlreadyExists. This allows the new Create to proceed
immediately without waiting for the potentially slow recovery to
complete or fail.

Longhorn 13185

Signed-off-by: Derek Su <derek.su@suse.com>
(cherry picked from commit dfac075)

Author: derekbit

pkg/spdk/enginefrontend.go

@@ -155,6 +155,22 @@ type UblkFrontend struct {
 	UblkID int32
 }
 
+func (ef *EngineFrontend) setMetadataDirLocked(metadataDir string) {
+	ef.metadataDir = metadataDir
+}
+
+func (ef *EngineFrontend) setMetadataDir(metadataDir string) {
+	ef.Lock()
+	defer ef.Unlock()
+	ef.setMetadataDirLocked(metadataDir)
+}
+
+func (ef *EngineFrontend) getMetadataDir() string {
+	ef.RLock()
+	defer ef.RUnlock()
+	return ef.metadataDir
+}
+
 func getUblkQueueDepth(ublkQueueDepth int32) int32 {
 	if ublkQueueDepth == 0 {
 		return types.DefaultUblkQueueDepth
@@ -2620,6 +2636,17 @@ func (ef *EngineFrontend) RecoverFromHost(spdkClient *spdkclient.Client) error {
 		return nil
 
 	case types.FrontendSPDKTCPBlockdev:
+		// Early cancellation check before creating the NVMe-TCP initiator.
+		// If a concurrent EngineFrontendCreate already completed for this
+		// volume (evicted us and connected its own NVMe controller), we must
+		// not proceed — creating an initiator and then calling Delete/Stop
+		// would disconnect the NEW ef's controller via DisconnectTarget
+		// (which disconnects ALL controllers for the subsystem NQN).
+		if ef.isRecoveryCancelled() {
+			recoverErr = ErrRecoveryCancelled
+			return recoverErr
+		}
+
 		// Recover the NVMe-oF initiator (blockdev frontend with dm-device).
 		i, nqn, nguid, err := ef.newNvmeTcpInitiator()
 		if err != nil {
@@ -2679,7 +2706,7 @@ func (ef *EngineFrontend) RecoverFromHost(spdkClient *spdkclient.Client) error {
 					reconnected = true
 				} else {
 					ef.log.WithError(loadErr).Warnf("NVMe device not found on host during recovery of engine frontend %s, removing persisted record", ef.Name)
-					if removeErr := removeEngineFrontendRecord(ef.metadataDir, ef.VolumeName); removeErr != nil {
+					if removeErr := removeEngineFrontendRecord(ef.getMetadataDir(), ef.VolumeName); removeErr != nil {
 						ef.log.WithError(removeErr).Warn("Failed to remove engine frontend record")
 					}
 					deviceNotFound = true

pkg/spdk/server.go

@@ -1132,6 +1132,11 @@ func (s *Server) recoverEngineFrontends(ctx context.Context) {
 
 			if current != ef {
 				logrus.Warnf("Engine frontend %s was superseded during recovery, cleaning up recovered resources", record.Name)
+				// The eviction path in EngineFrontendCreate already decided
+				// whether metadataDir should be kept (pre-create eviction,
+				// where no new record exists yet) or cleared (post-create
+				// eviction with successful Create, where a new record was
+				// written). Respect that decision — do not override here.
 				if deleteErr := ef.Delete(spdkClient); deleteErr != nil {
 					logrus.WithError(deleteErr).Warnf("Failed to clean up superseded engine frontend %s", record.Name)
 				}

pkg/spdk/server_enginefrontend.go

@@ -232,24 +232,9 @@ func (s *Server) EngineFrontendCreate(ctx context.Context, req *spdkrpc.EngineFr
 		return nil, grpcstatus.Errorf(grpccodes.InvalidArgument, "frontend %v is not supported", req.Frontend)
 	}
 
-	s.Lock()
-	_, ok := s.engineFrontendMap[req.Name]
-	if ok {
-		s.Unlock()
-		return nil, grpcstatus.Errorf(grpccodes.AlreadyExists, "engine frontend %v already exists", req.Name)
-	}
-	if existing := s.engineFrontendByVolumeName(req.VolumeName); existing != nil {
-		s.Unlock()
-		return nil, grpcstatus.Errorf(grpccodes.AlreadyExists, "engine frontend %v already exists for volume %v", existing.Name, req.VolumeName)
-	}
-
-	ef := NewEngineFrontend(req.Name, req.EngineName, req.VolumeName, req.Frontend, req.SpecSize,
-		req.UblkQueueDepth, req.UblkNumberOfQueue, s.updateChs[types.InstanceTypeEngineFrontend])
-	ef.metadataDir = s.metadataDir
-
-	spdkClient := s.spdkClient
-	s.Unlock()
-
+	// Derive and validate targetAddress BEFORE evicting any Pending recovery
+	// ef, so that a malformed address does not permanently cancel a valid
+	// ongoing recovery.
 	targetAddress := req.TargetAddress
 	// When disableFrontend=True, the manager passes an empty target address
 	// because the engine's NVMe-oF target port is 0 (no listener exposed).
@@ -262,8 +247,80 @@ func (s *Server) EngineFrontendCreate(ctx context.Context, req *spdkrpc.EngineFr
 			targetAddress = net.JoinHostPort(podIP, strconv.Itoa(types.SPDKServicePort))
 		}
 	}
+	// Validate the address format. ef.Create() would reject this with a hard
+	// error (ErrEngineFrontendCreateInvalidArgument), but at that point we
+	// may have already evicted a Pending recovery ef.
+	if _, _, splitErr := splitHostPort(targetAddress); splitErr != nil {
+		return nil, grpcstatus.Errorf(grpccodes.InvalidArgument, "invalid target address %v: %v", targetAddress, splitErr)
+	}
+
+	s.Lock()
+
+	var evictedEfs []*EngineFrontend
+
+	if existing, ok := s.engineFrontendMap[req.Name]; ok {
+		// If the conflicting ef is still recovering (Pending state from
+		// async recovery), the new Create takes priority — evict it.
+		// We only mark state and remove from map here; the recovery
+		// goroutine will detect the state/map change and handle Delete
+		// sequentially after RecoverFromHost returns, avoiding concurrent
+		// access to the initiator.
+		//
+		// Keep metadataDir intact for now. After Create() we decide:
+		// - Create succeeded → clear metadataDir so the old ef's Delete
+		//   does not remove the NEW persistence record (same volumeName key).
+		// - Create failed with runtime error → keep metadataDir so Delete
+		//   removes the stale old record that would cause bad recovery.
+		existing.Lock()
+		if existing.State == types.InstanceStatePending {
+			existing.State = types.InstanceStateTerminating
+			existing.Unlock()
+			delete(s.engineFrontendMap, req.Name)
+			evictedEfs = append(evictedEfs, existing)
+		} else {
+			existing.Unlock()
+			s.Unlock()
+			return nil, grpcstatus.Errorf(grpccodes.AlreadyExists, "engine frontend %v already exists", req.Name)
+		}
+	}
+	if existing := s.engineFrontendByVolumeName(req.VolumeName); existing != nil {
+		existing.Lock()
+		if existing.State == types.InstanceStatePending {
+			existing.State = types.InstanceStateTerminating
+			existing.Unlock()
+			delete(s.engineFrontendMap, existing.Name)
+			evictedEfs = append(evictedEfs, existing)
+		} else {
+			existing.Unlock()
+			s.Unlock()
+			return nil, grpcstatus.Errorf(grpccodes.AlreadyExists, "engine frontend %v already exists for volume %v", existing.Name, req.VolumeName)
+		}
+	}
+
+	ef := NewEngineFrontend(req.Name, req.EngineName, req.VolumeName, req.Frontend, req.SpecSize,
+		req.UblkQueueDepth, req.UblkNumberOfQueue, s.updateChs[types.InstanceTypeEngineFrontend])
+	ef.metadataDir = s.metadataDir
+
+	s.Unlock()
+
+	// Hold the per-volume host lock so that if the async recovery goroutine
+	// is still performing host NVMe/dm operations for this volume, we wait
+	// for it to finish before starting our own. The lock is held through
+	// map registration so that a second concurrent Create for the same
+	// volume cannot start host operations before we register.
+	unlockVolumeHost := s.acquireVolumeHostLock(req.VolumeName)
+	defer unlockVolumeHost()
+
+	s.RLock()
+	spdkClient := s.spdkClient
+	s.RUnlock()
 
 	ret, createErr := ef.Create(spdkClient, targetAddress)
+	for _, evicted := range evictedEfs {
+		if createErr == nil && evicted.VolumeName == req.VolumeName {
+			evicted.setMetadataDir("")
+		}
+	}
 
 	// Distinguish hard errors (validation / precondition) from runtime
 	// failures (e.g. NVMe initiator can't connect).  Hard errors are
@@ -273,6 +330,26 @@ func (s *Server) EngineFrontendCreate(ctx context.Context, req *spdkrpc.EngineFr
 	if createErr != nil &&
 		(errors.Is(createErr, ErrEngineFrontendCreateInvalidArgument) ||
 			errors.Is(createErr, ErrEngineFrontendCreatePrecondition)) {
+		// Hard error — Create did not mutate anything.  Restore the
+		// evicted efs so valid ongoing recoveries are not permanently lost.
+		if len(evictedEfs) > 0 {
+			s.Lock()
+			for _, evicted := range evictedEfs {
+				evicted.Lock()
+				evicted.State = types.InstanceStatePending
+				evicted.Unlock()
+				// Only restore if both the name and volume slots are still free.
+				// A concurrent Create for the same volume (different name) could
+				// have registered while we were in-flight, and blindly restoring
+				// would violate the one-ef-per-volume map invariant.
+				if _, taken := s.engineFrontendMap[evicted.Name]; !taken {
+					if s.engineFrontendByVolumeName(evicted.VolumeName) == nil {
+						s.engineFrontendMap[evicted.Name] = evicted
+					}
+				}
+			}
+			s.Unlock()
+		}
 		return nil, toEngineFrontendCreateGRPCError(createErr, "failed to create engine frontend %v", req.Name)
 	}
 
@@ -290,6 +367,35 @@ func (s *Server) EngineFrontendCreate(ctx context.Context, req *spdkrpc.EngineFr
 		winner = existing
 	}
 	if duplicateName || duplicateVolume {
+		// If the conflicting ef is still recovering (inserted by async
+		// recovery between our first check and now), the new Create wins.
+		// Only mark state and remove from map; the recovery goroutine
+		// will handle Delete sequentially after RecoverFromHost returns.
+		if winner != nil {
+			winner.Lock()
+			if winner.State == types.InstanceStatePending {
+				winner.State = types.InstanceStateTerminating
+				// Only clear metadataDir when Create succeeded AND the winner
+				// shares the same volumeName. Records are keyed by volumeName,
+				// so only same-volume records collide. Different-volume winners
+				// must keep metadataDir so Delete removes their own record.
+				if createErr == nil && winner.VolumeName == ef.VolumeName {
+					winner.setMetadataDirLocked("")
+				}
+				winner.Unlock()
+
+				delete(s.engineFrontendMap, winner.Name)
+				s.engineFrontendMap[req.Name] = ef
+				s.Unlock()
+
+				if createErr != nil {
+					return ef.Get(), nil
+				}
+				return ret, nil
+			}
+			winner.Unlock()
+		}
+
 		s.Unlock()
 		// The race loser holds a fully-created frontend with real SPDK
 		// resources (bdevs, NVMe controllers, etc.). Clean them up so
@@ -300,7 +406,7 @@ func (s *Server) EngineFrontendCreate(ctx context.Context, req *spdkrpc.EngineFr
 		// When volumeNames differ, each has its own directory and the
 		// loser should clean up its own record.
 		if winner != nil && ef.VolumeName == winner.VolumeName {
-			ef.metadataDir = ""
+			ef.setMetadataDir("")
 		}
 		if deleteErr := ef.Delete(spdkClient); deleteErr != nil {
 			logrus.WithError(deleteErr).Warnf("Failed to clean up race-loser engine frontend %v", req.Name)
@@ -311,10 +417,12 @@ func (s *Server) EngineFrontendCreate(ctx context.Context, req *spdkrpc.EngineFr
 		// Hold the winner's read lock to prevent concurrent mutations
 		// (e.g. Delete, switchover) from racing with the field reads
 		// inside saveEngineFrontendRecord.
-		if winner != nil && winner.metadataDir != "" && ef.VolumeName == winner.VolumeName {
+		if winner != nil && ef.VolumeName == winner.VolumeName {
 			winner.RLock()
-			if err := saveEngineFrontendRecord(winner.metadataDir, winner); err != nil {
-				logrus.WithError(err).Warnf("Failed to re-persist winner engine frontend %v record after race", winner.Name)
+			if metadataDir := winner.metadataDir; metadataDir != "" {
+				if err := saveEngineFrontendRecord(metadataDir, winner); err != nil {
+					logrus.WithError(err).Warnf("Failed to re-persist winner engine frontend %v record after race", winner.Name)
+				}
 			}
 			winner.RUnlock()
 		}

pkg/spdk/server_verify_test.go

@@ -374,6 +374,7 @@ func (s *TestSuite) TestEngineFrontendCreateReturnsAlreadyExistsForDuplicate(c *
 	updateCh := make(chan interface{}, 1)
 
 	existing := NewEngineFrontend("ef-dup", "engine-a", "vol-a", lhtypes.FrontendSPDKTCPNvmf, 1024, 0, 0, updateCh)
+	existing.State = lhtypes.InstanceStateRunning
 
 	srv := &Server{
 		engineFrontendMap: map[string]*EngineFrontend{
@@ -587,3 +588,94 @@ func (s *TestSuite) TestEngineFrontendDeleteWaitsForVolumeHostLock(c *C) {
 		c.Fatal("delete did not resume after releasing the per-volume host lock")
 	}
 }
+
+func (s *TestSuite) TestEngineFrontendCreateInvalidAddressDoesNotEvictPendingEf(c *C) {
+	fmt.Println("Testing EngineFrontendCreate with invalid address does not evict a Pending frontend")
+
+	updateCh := make(chan interface{}, 1)
+	existing := NewEngineFrontend("ef-recovering", "engine-a", "vol-a", lhtypes.FrontendSPDKTCPNvmf, 1024, 0, 0, updateCh)
+	// State is Pending (the default from NewEngineFrontend), simulating an
+	// in-progress async recovery.
+
+	srv := &Server{
+		engineFrontendMap: map[string]*EngineFrontend{
+			"ef-recovering": existing,
+		},
+		volumeHostLocks: map[string]*volumeHostLockEntry{},
+		updateChs: map[lhtypes.InstanceType]chan interface{}{
+			lhtypes.InstanceTypeEngineFrontend: updateCh,
+		},
+	}
+
+	// Use an un-bracketed IPv6 address that splitHostPort rejects.
+	_, err := srv.EngineFrontendCreate(context.Background(), &spdkrpc.EngineFrontendCreateRequest{
+		Name:          "ef-recovering",
+		EngineName:    "engine-b",
+		VolumeName:    "vol-a",
+		Frontend:      lhtypes.FrontendSPDKTCPNvmf,
+		SpecSize:      2048,
+		TargetAddress: "2001:db8::1:9502",
+	})
+	c.Assert(err, NotNil)
+	st, ok := grpcstatus.FromError(err)
+	c.Assert(ok, Equals, true)
+	c.Assert(st.Code(), Equals, grpccodes.InvalidArgument)
+
+	// The existing Pending frontend must still be in the map, untouched.
+	srv.RLock()
+	ef := srv.engineFrontendMap["ef-recovering"]
+	srv.RUnlock()
+	c.Assert(ef, Equals, existing)
+	c.Assert(ef.State, Equals, lhtypes.InstanceState(lhtypes.InstanceStatePending))
+}
+
+func (s *TestSuite) TestEngineFrontendCreateEvictsMultiplePendingRecoveries(c *C) {
+	fmt.Println("Testing EngineFrontendCreate evicts both name and volume Pending recoveries in one request")
+
+	updateCh := make(chan interface{}, 1)
+	sameName := NewEngineFrontend("ef-new", "engine-old-name", "vol-old", lhtypes.FrontendSPDKTCPNvmf, 1024, 0, 0, updateCh)
+	sameName.metadataDir = "/tmp/name-recovery"
+	sameVolume := NewEngineFrontend("ef-old-volume", "engine-old-volume", "vol-new", lhtypes.FrontendSPDKTCPNvmf, 1024, 0, 0, updateCh)
+	sameVolume.metadataDir = "/tmp/volume-recovery"
+
+	srv := &Server{
+		engineFrontendMap: map[string]*EngineFrontend{
+			sameName.Name:   sameName,
+			sameVolume.Name: sameVolume,
+		},
+		volumeHostLocks: map[string]*volumeHostLockEntry{},
+		updateChs: map[lhtypes.InstanceType]chan interface{}{
+			lhtypes.InstanceTypeEngineFrontend: updateCh,
+		},
+	}
+
+	resp, err := srv.EngineFrontendCreate(context.Background(), &spdkrpc.EngineFrontendCreateRequest{
+		Name:       "ef-new",
+		EngineName: "engine-new",
+		VolumeName: "vol-new",
+		Frontend:   lhtypes.FrontendSPDKTCPNvmf,
+		SpecSize:   2048,
+	})
+	c.Assert(err, IsNil)
+	c.Assert(resp, NotNil)
+
+	srv.RLock()
+	created := srv.engineFrontendMap["ef-new"]
+	_, sameVolumeStillMapped := srv.engineFrontendMap["ef-old-volume"]
+	srv.RUnlock()
+
+	c.Assert(created, NotNil)
+	c.Assert(created, Not(Equals), sameName)
+	c.Assert(created.VolumeName, Equals, "vol-new")
+	c.Assert(sameVolumeStillMapped, Equals, false)
+
+	sameName.RLock()
+	c.Assert(sameName.State, Equals, lhtypes.InstanceState(lhtypes.InstanceStateTerminating))
+	c.Assert(sameName.metadataDir, Equals, "/tmp/name-recovery")
+	sameName.RUnlock()
+
+	sameVolume.RLock()
+	c.Assert(sameVolume.State, Equals, lhtypes.InstanceState(lhtypes.InstanceStateTerminating))
+	c.Assert(sameVolume.metadataDir, Equals, "")
+	sameVolume.RUnlock()
+}