refactor(runtime): remove queue shutdown loop

Author: MarcStdt

apps/druid/core/services/runtime_lifecycle.go

@@ -12,7 +12,7 @@ func (s *RuntimeSupervisor) DeleteWithPolicy(id string, purgeData bool) error {
 	delete(s.sessions, id)
 	s.mu.Unlock()
 	if session != nil {
-		session.Shutdown()
+		session.stopDeploymentQueue()
 	}
 
 	runtimeScroll, err := s.store.GetScroll(id)
@@ -60,6 +60,6 @@ func (s *RuntimeSupervisor) Stop(id string) (*domain.RuntimeScroll, error) {
 		session.markError(err)
 		return nil, err
 	}
-	session.Shutdown()
+	session.stopDeploymentQueue()
 	return s.store.GetScroll(id)
 }

apps/druid/core/services/runtime_session.go

@@ -22,11 +22,6 @@ type RuntimeSession struct {
 	runtimeBackend    ports.RuntimeBackendInterface
 	procedures        ports.ProcedureLauchnerInterface
 	queue             map[string]*runtimeQueueItem
-	taskChan          chan string
-	taskDoneChan      chan struct{}
-	shutdownChan      chan struct{}
-	shutdownDoneChan  chan struct{}
-	shutdownOnce      sync.Once
 	workWg            sync.WaitGroup
 	notifierChan      []chan []string
 	devWatchPaths     []string
@@ -84,19 +79,7 @@ func (s *RuntimeSession) Start() {
 	}
 	s.started = true
 	s.mu.Unlock()
-	s.startQueue()
-}
-
-func (s *RuntimeSession) Shutdown() {
-	s.shutdownQueueLoop()
-	s.mu.Lock()
-	s.started = false
-	s.mu.Unlock()
-}
-
-func (s *RuntimeSession) startQueue() {
-	s.resetQueueState()
-	go s.Work()
+	s.triggerRunQueue()
 }
 
 func (s *RuntimeSession) newQueue(scrollService *coreservices.ScrollService, root string, scrollName string) (*coreservices.QueueManager, *coreservices.ProcedureLauncher, error) {
@@ -133,7 +116,11 @@ func (s *RuntimeSession) replaceQueue(start bool) error {
 	s.mu.Unlock()
 
 	if start {
-		s.startQueue()
+		s.resetQueueState()
+		s.mu.Lock()
+		s.started = true
+		s.mu.Unlock()
+		s.triggerRunQueue()
 	} else {
 		s.resetQueueState()
 	}

apps/druid/core/services/runtime_session_cache.go

@@ -59,7 +59,7 @@ func (s *RuntimeSupervisor) startSession(runtimeScroll *domain.RuntimeScroll) (*
 	s.mu.Lock()
 	if existing := s.sessions[runtimeScroll.ID]; existing != nil {
 		s.mu.Unlock()
-		session.Shutdown()
+		session.stopDeploymentQueue()
 		return existing, nil
 	}
 	s.sessions[runtimeScroll.ID] = session

apps/druid/core/services/runtime_session_queue.go

@@ -3,8 +3,6 @@ package services
 import (
 	"context"
 	"errors"
-	"fmt"
-	"sync"
 	"time"
 
 	"github.com/highcard-dev/daemon/internal/core/domain"
@@ -76,10 +74,9 @@ func (s *RuntimeSession) addQueueItem(cmd string, options coreservices.AddItemOp
 	item = &runtimeQueueItem{doneChan: doneChan}
 	s.queue[cmd] = item
 	s.setQueueStatusLocked(cmd, item, domain.ScrollLockStatusWaiting, nil)
-	taskChan := s.taskChan
 	s.queueMu.Unlock()
 
-	taskChan <- cmd
+	s.triggerRunQueue()
 
 	if options.Wait {
 		<-doneChan
@@ -127,29 +124,14 @@ func (s *RuntimeSession) HydrateFromState(statuses domain.ProcedureStatusMap) er
 	return nil
 }
 
-func (s *RuntimeSession) Work() {
-	s.queueMu.Lock()
-	taskChan := s.taskChan
-	taskDoneChan := s.taskDoneChan
-	shutdownChan := s.shutdownChan
-	shutdownDoneChan := s.shutdownDoneChan
-	s.queueMu.Unlock()
-	defer close(shutdownDoneChan)
-
-	for {
-		select {
-		case <-taskChan:
-			s.startRunQueue()
-		case <-taskDoneChan:
-			s.startRunQueue()
-		case <-shutdownChan:
-			s.workWg.Wait()
-			s.queueMu.Lock()
-			s.queue = make(map[string]*runtimeQueueItem)
-			s.queueMu.Unlock()
-			return
-		}
+func (s *RuntimeSession) triggerRunQueue() {
+	s.mu.Lock()
+	started := s.started
+	s.mu.Unlock()
+	if !started {
+		return
 	}
+	s.startRunQueue()
 }
 
 func (s *RuntimeSession) startRunQueue() {
@@ -241,15 +223,7 @@ func (s *RuntimeSession) RunQueue() {
 				if i.doneChan != nil {
 					close(i.doneChan)
 				}
-
-				s.queueMu.Lock()
-				taskDoneChan := s.taskDoneChan
-				shutdownChan := s.shutdownChan
-				s.queueMu.Unlock()
-				select {
-				case taskDoneChan <- struct{}{}:
-				case <-shutdownChan:
-				}
+				s.triggerRunQueue()
 			}()
 
 			startedAt := time.Now()
@@ -301,7 +275,6 @@ func (s *RuntimeSession) WaitUntilEmptyContext(ctx context.Context) error {
 
 	s.queueMu.Lock()
 	s.notifierChan = append(s.notifierChan, notifier)
-	shutdownChan := s.shutdownChan
 	if !s.hasActiveItemsLocked() {
 		s.removeNotifierLocked(notifier)
 		s.queueMu.Unlock()
@@ -322,8 +295,6 @@ func (s *RuntimeSession) WaitUntilEmptyContext(ctx context.Context) error {
 			}
 		case <-ctx.Done():
 			return ctx.Err()
-		case <-shutdownChan:
-			return fmt.Errorf("queue manager shut down while waiting for commands")
 		}
 	}
 }
@@ -340,34 +311,32 @@ func (s *RuntimeSession) GetQueue() map[string]domain.ScrollLockStatus {
 	return queue
 }
 
-func (s *RuntimeSession) shutdownQueueLoop() {
-	s.queueMu.Lock()
-	shutdownChan := s.shutdownChan
-	shutdownDoneChan := s.shutdownDoneChan
-	s.queueMu.Unlock()
+func (s *RuntimeSession) stopDeploymentQueue() {
+	s.mu.Lock()
+	s.started = false
+	s.mu.Unlock()
+	s.drainQueueWork()
+	s.resetQueueState()
+}
 
-	s.shutdownOnce.Do(func() {
-		close(shutdownChan)
-		s.notify()
-	})
+func (s *RuntimeSession) drainQueueWork() {
+	done := make(chan struct{})
+	go func() {
+		s.workWg.Wait()
+		close(done)
+	}()
 
 	select {
-	case <-shutdownDoneChan:
+	case <-done:
 	case <-time.After(5 * time.Second):
-		logger.Log().Warn("Timed out waiting for queue manager shutdown")
+		logger.Log().Warn("Timed out waiting for queue work to finish")
 	}
 }
 
 func (s *RuntimeSession) resetQueueState() {
 	s.queueMu.Lock()
 	defer s.queueMu.Unlock()
 	s.queue = make(map[string]*runtimeQueueItem)
-	s.taskChan = make(chan string, 100)
-	s.taskDoneChan = make(chan struct{}, 1)
-	s.shutdownChan = make(chan struct{})
-	s.shutdownDoneChan = make(chan struct{})
-	s.shutdownOnce = sync.Once{}
-	s.workWg = sync.WaitGroup{}
 	s.notifierChan = make([]chan []string, 0)
 }
 

apps/druid/core/services/runtime_session_runtime.go

@@ -48,14 +48,20 @@ func (s *RuntimeSession) StopRuntime() error {
 	started := s.started
 	s.mu.Unlock()
 	if started {
-		s.shutdownQueueLoop()
+		s.mu.Lock()
+		s.started = false
+		s.mu.Unlock()
+		s.drainQueueWork()
 	}
 	if err := s.replaceQueue(false); err != nil {
 		return err
 	}
 	if err := s.runtimeBackend.StopRuntime(root); err != nil {
 		if started {
-			s.startQueue()
+			s.mu.Lock()
+			s.started = true
+			s.mu.Unlock()
+			s.triggerRunQueue()
 		}
 		return err
 	}
@@ -74,7 +80,10 @@ func (s *RuntimeSession) StopRuntime() error {
 	err := s.store.UpdateScroll(s.runtimeScroll)
 	s.mu.Unlock()
 	if err == nil && started {
-		s.startQueue()
+		s.mu.Lock()
+		s.started = true
+		s.mu.Unlock()
+		s.triggerRunQueue()
 	}
 	return err
 }
@@ -117,7 +126,10 @@ func (s *RuntimeSession) ApplyRestore(materialized *ports.RuntimeMaterialization
 	started := s.started
 	s.mu.Unlock()
 	if started {
-		s.shutdownQueueLoop()
+		s.mu.Lock()
+		s.started = false
+		s.mu.Unlock()
+		s.drainQueueWork()
 	}
 
 	s.mu.Lock()
@@ -142,7 +154,11 @@ func (s *RuntimeSession) ApplyRestore(materialized *ports.RuntimeMaterialization
 	err = s.store.UpdateScroll(s.runtimeScroll)
 	s.mu.Unlock()
 	if err == nil && started {
-		s.startQueue()
+		s.resetQueueState()
+		s.mu.Lock()
+		s.started = true
+		s.mu.Unlock()
+		s.triggerRunQueue()
 	}
 	return err
 }

apps/druid/core/services/runtime_supervisor_test.go

@@ -324,7 +324,7 @@ func TestRuntimeSupervisorStartHydratesRunningScroll(t *testing.T) {
 	}
 	defer func() {
 		if session := supervisor.sessions["running-scroll"]; session != nil {
-			session.Shutdown()
+			session.stopDeploymentQueue()
 		}
 	}()
 

apps/druid/core/services/runtime_update.go

@@ -32,7 +32,7 @@ func (s *RuntimeSupervisor) updateExistingScroll(runtimeScroll *domain.RuntimeSc
 	delete(s.sessions, runtimeScroll.ID)
 	s.mu.Unlock()
 	if session != nil {
-		session.Shutdown()
+		session.stopDeploymentQueue()
 	}
 
 	if wasRunning {

test/integration/example_test.go

@@ -175,7 +175,6 @@ func TestExamples(t *testing.T) {
 				t.Error(err)
 				return
 			}
-			session.Shutdown()
 
 			if config.TestAddress != "" {
 				err = test_utils.ConnectionTest(config.TestAddress, false)