refactor: update lock release worker to handle all nodes of workflow once

Author: SalmaElsoly

backend/cmd/root.go

@@ -206,7 +206,7 @@ func addFlags() error {
 	if err := bindIntFlag(rootCmd, "locks.lock_timeout_in_hours", 24, "Redis lock timeout (hours)"); err != nil {
 		return fmt.Errorf("failed to bind locks.lock_timeout_in_hours flag: %w", err)
 	}
-	if err := bindIntFlag(rootCmd, "locks.locks_release_interval_in_minutes", 5, "Locks release interval (minutes)"); err != nil {
+	if err := bindIntFlag(rootCmd, "locks.locks_release_interval_in_minutes", 2, "Locks release interval (minutes)"); err != nil {
 		return fmt.Errorf("failed to bind locks.locks_release_interval_in_minutes flag: %w", err)
 	}
 

backend/internal/api/app/app.go

@@ -223,6 +223,7 @@ func (app *App) StartBackgroundWorkers() {
 	go app.workers.TrackReservedNodeHealth()
 	go app.workers.CollectGORMMetrics()
 	go app.workers.CollectGoRuntimeMetrics()
+	go app.workers.ReleaseWorkflowLocks()
 }
 
 // Run starts the server

backend/internal/api/handlers/deployment_handler.go

@@ -258,6 +258,10 @@ func (h *DeploymentHandler) HandleDeployCluster(c *gin.Context) {
 	wfUUID, wfStatus, err := h.svc.AsyncDeployCluster(config, cluster)
 	if err != nil {
 		reqLog.Error().Err(err).Msg("failed to start deployment workflow")
+		err = h.locker.ReleaseLock(c.Request.Context(), nodeIDs, wfUUID)
+		if err != nil {
+			reqLog.Error().Err(err).Msg("failed to release nodes locks")
+		}
 		InternalServerError(c)
 		return
 	}
@@ -439,6 +443,10 @@ func (h *DeploymentHandler) HandleAddNode(c *gin.Context) {
 	wfUUID, wfStatus, err := h.svc.AsyncAddNode(config, cl, cluster.Nodes[0])
 	if err != nil {
 		reqLog.Error().Err(err).Msg("failed to start add node workflow")
+		err = h.locker.ReleaseLock(c.Request.Context(), []uint32{cluster.Nodes[0].NodeID}, wfUUID)
+		if err != nil {
+			reqLog.Error().Err(err).Msg("failed to release nodes locks")
+		}
 		InternalServerError(c)
 		return
 	}

backend/internal/api/handlers/node_handler.go

@@ -310,6 +310,10 @@ func (h *NodeHandler) ReserveNodeHandler(c *gin.Context) {
 	wfUUID, err := h.svc.AsyncReserveNode(userID, user.Mnemonic, nodeID)
 	if err != nil {
 		reqLog.Error().Err(err).Msg("failed to start workflow to reserve node")
+		err = h.locker.ReleaseLock(c.Request.Context(), []uint32{nodeID}, wfUUID)
+		if err != nil {
+			reqLog.Error().Err(err).Msg("failed to release nodes locks")
+		}
 		InternalServerError(c)
 		return
 	}

backend/internal/config/config.go

@@ -394,7 +394,7 @@ func applyDefaultValues(config *Configuration) {
 		config.Locks.LockTimeoutInHours = 24
 	}
 	if config.Locks.LocksReleaseIntervalInMinutes == 0 {
-		config.Locks.LocksReleaseIntervalInMinutes = 5
+		config.Locks.LocksReleaseIntervalInMinutes = 2
 	}
 
 	if config.Telemetry.OTLPEndpoint == "" {

backend/internal/core/distributed_locks/distributed_locks.go

@@ -11,7 +11,7 @@ var ErrNodeLocked = errors.New("node is currently locked by another request")
 type DistributedLocks interface {
 	AcquireNodesLocks(ctx context.Context, nodeIDs []uint32) error
 	AcquireWorkflowLock(ctx context.Context, nodeIDs []uint32, workflowID string) error
-	ReleaseLock(ctx context.Context, nodeID uint32, workflowID string) error
+	ReleaseLock(ctx context.Context, nodeIDs []uint32, workflowID string) error
 	GetAllWorkflowsLocks(ctx context.Context) ([]string, error)
 	GetLockedNodes(ctx context.Context) ([]uint32, error)
 }

backend/internal/core/distributed_locks/redis_locker.go

@@ -101,10 +101,13 @@ func (l *RedisLocker) rollbackLocks(ctx context.Context, keys []string) error {
 	return nil
 }
 
-func (l *RedisLocker) ReleaseLock(ctx context.Context, nodeID uint32, workflowID string) error {
-	lockedKey := nodeLockKey(nodeID)
-	usedKey := workflowLockKey(nodeID, workflowID)
-	return l.client.Del(ctx, lockedKey, usedKey).Err()
+func (l *RedisLocker) ReleaseLock(ctx context.Context, nodeIDs []uint32, workflowID string) error {
+	lockedKeys := lockKeys(nodeIDs, nodeLockKey)
+	usedKeys := lockKeys(nodeIDs, func(id uint32) string {
+		return workflowLockKey(id, workflowID)
+	})
+	allWorkflowsLocks := append(lockedKeys, usedKeys...)
+	return l.client.Del(ctx, allWorkflowsLocks...).Err()
 }
 
 // GetAllWorkflowsLocks gets all workflow locks.

backend/internal/core/distributed_locks/redis_locker_test.go

@@ -69,7 +69,7 @@ func TestRedisLocker_ReleaseLock(t *testing.T) {
 	require.NoError(t, client.Set(context.Background(), "locked:1", 1, 0).Err())
 	require.NoError(t, client.Set(context.Background(), "used:1:wf-1", 1, 0).Err())
 
-	err := locker.ReleaseLock(context.Background(), 1, "wf-1")
+	err := locker.ReleaseLock(context.Background(), []uint32{1}, "wf-1")
 
 	require.NoError(t, err)
 	require.Equal(t, int64(0), client.Exists(context.Background(), "locked:1").Val())

backend/internal/core/services/node_service_test.go

@@ -36,8 +36,8 @@ func (m *MockDistributedLocks) AcquireWorkflowLock(ctx context.Context, nodeIDs
 	return args.Error(0)
 }
 
-func (m *MockDistributedLocks) ReleaseLock(ctx context.Context, nodeID uint32, workflowID string) error {
-	args := m.Called(ctx, nodeID, workflowID)
+func (m *MockDistributedLocks) ReleaseLock(ctx context.Context, nodeIDs []uint32, workflowID string) error {
+	args := m.Called(ctx, nodeIDs, workflowID)
 	return args.Error(0)
 }
 

backend/internal/core/services/workers_service.go

@@ -474,32 +474,41 @@ func (svc WorkerService) GetAllWorkflowsLocks() ([]string, error) {
 	return svc.locker.GetAllWorkflowsLocks(svc.ctx)
 }
 
-func (svc WorkerService) ReleaseLocks(key string) error {
-	keyParts := strings.Split(key, ":")
-	if len(keyParts) != 3 {
-		return fmt.Errorf("invalid lock key format: %s", key)
-	}
-
-	nodeID, err := strconv.ParseUint(keyParts[1], 10, 32)
-	if err != nil {
-		return fmt.Errorf("invalid node ID: %w", err)
-	}
-
-	workflowID := keyParts[2]
-	workflow, err := svc.ewfEngine.Store().LoadWorkflowByUUID(svc.ctx, workflowID)
-	if err != nil {
-		return fmt.Errorf("failed to load workflow by UUID: %w", err)
-	}
+func (svc WorkerService) ReleaseLocks(keys []string) {
+	log := logger.ForOperation("locks_worker", "release_locks")
+	workflowsNodes := map[string][]uint32{}
+	for _, key := range keys {
+		parts := strings.Split(key, ":")
+		if len(parts) != 3 {
+			log.Error().Str("key", key).Msg("invalid lock key format")
+			continue
+		}
 
-	if !slices.Contains([]ewf.WorkflowStatus{ewf.StatusCompleted, ewf.StatusFailed}, workflow.Status) {
-		return nil
+		workflowID := parts[2]
+		nodeID, err := strconv.ParseUint(parts[1], 10, 32)
+		if err != nil {
+			log.Error().Str("key", key).Msg("invalid node ID")
+			continue
+		}
+		workflowsNodes[workflowID] = append(workflowsNodes[workflowID], uint32(nodeID))
 	}
 
-	if err := svc.locker.ReleaseLock(svc.ctx, uint32(nodeID), workflowID); err != nil {
-		return fmt.Errorf("failed to release nodes locks: %w", err)
+	for workflowID := range workflowsNodes {
+		workflow, err := svc.ewfEngine.Store().LoadWorkflowByUUID(svc.ctx, workflowID)
+		if err != nil {
+			log.Error().Str("workflow_id", workflowID).Msg("failed to load workflow")
+			continue
+		}
+		if !slices.Contains([]ewf.WorkflowStatus{ewf.StatusCompleted, ewf.StatusFailed}, workflow.Status) {
+			continue
+		}
+		nodeIDs := workflowsNodes[workflowID]
+		if err := svc.locker.ReleaseLock(svc.ctx, nodeIDs, workflowID); err != nil {
+			log.Error().Str("workflow_id", workflow.UUID).Msg("failed to release locks")
+			continue
+		}
 	}
 
-	return nil
 }
 
 func getHoursOfGivenPeriod(startDate, endDate time.Time) int {