Merge pull request #319 from thand-io/fix/temporal-register-before-start

Fix Temporal worker startup order

Author: hughneale

internal/config/providers.go

@@ -187,6 +187,12 @@ type initResult struct {
 
 // InitializeProviders initializes all providers in parallel using channels
 func (c *Config) InitializeProviders() error {
+	if c.IsServer() {
+		_ = c.GetServices()
+		if err := c.SetupTemporal(); err != nil {
+			return fmt.Errorf("setting up temporal services: %w", err)
+		}
+	}
 
 	defs := c.GetProviders().Definitions
 
@@ -364,6 +370,10 @@ func (c *Config) InitializeProviders() error {
 	c.providerInstances = results
 	c.mu.Unlock()
 
+	if err := c.StartTemporalWorkers(); err != nil {
+		return err
+	}
+
 	logrus.Debugln("All providers initialized successfully")
 
 	return nil

internal/config/services.go

@@ -18,22 +18,14 @@ func (c *Config) GetServices() models.ServicesClientImpl {
 			return
 		}
 		c.servicesClient = newClient
-		go func() {
-			// Post services setup initialization, we need to do some additional setup for certain services that are dependent on the configuration being fully loaded.
-			err = c.SetupTemporal()
-			if err != nil {
-				logrus.WithError(err).Error("Failed to set up temporal services")
-			}
-		}()
 	})
 
 	return c.servicesClient
 
 }
 
 func (c *Config) SetupTemporal() error {
-
-	if c.GetServices() != nil && c.GetServices().GetTemporal() != nil {
+	if c.servicesClient != nil && c.servicesClient.GetTemporal() != nil {
 
 		logrus.Infoln("Setting up temporal services...")
 
@@ -60,3 +52,19 @@ func (c *Config) SetupTemporal() error {
 	return nil
 
 }
+
+func (c *Config) StartTemporalWorkers() error {
+	if c.servicesClient == nil || c.servicesClient.GetTemporal() == nil {
+		return nil
+	}
+
+	if !c.IsServer() {
+		return nil
+	}
+
+	if err := c.servicesClient.GetTemporal().StartWorkers(); err != nil {
+		return fmt.Errorf("starting temporal workers: %w", err)
+	}
+
+	return nil
+}

internal/config/services/temporal/main.go

@@ -33,6 +33,7 @@ type TemporalClient struct {
 	mu             sync.Mutex
 	readyCh        chan struct{}
 	closeReadyOnce sync.Once
+	workersStarted bool
 }
 
 func NewTemporalClient(
@@ -131,7 +132,8 @@ func (a *TemporalClient) Initialize() error {
 		}
 	}
 
-	// Create and start a worker for each identity (task queue)
+	// Create a worker for each identity (task queue).
+	// Registration must happen before workers are started.
 	a.mu.Lock()
 	defer a.mu.Unlock()
 
@@ -147,32 +149,70 @@ func (a *TemporalClient) Initialize() error {
 			workerOptions,
 		)
 
+		a.workers[identity] = newWorker
+	}
+
+	if len(a.workers) == 0 {
+		a.markReady() // Unblock any waiters even on failure
+		return fmt.Errorf("failed to create any Temporal workers")
+	}
+
+	return nil
+}
+
+// StartWorkers starts all registered Temporal workers.
+// This must be called only after workflow/activity registration is complete.
+func (c *TemporalClient) StartWorkers() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.client == nil {
+		return fmt.Errorf("temporal client is not initialized")
+	}
+
+	if len(c.workers) == 0 {
+		c.markReady()
+		return fmt.Errorf("no Temporal workers configured")
+	}
+
+	if c.workersStarted {
+		logrus.Warn("Temporal workers already started, skipping worker startup")
+		return nil
+	}
+
+	buildID := common.GetBuildIdentifier()
+	startedCount := 0
+
+	for identity, w := range c.workers {
 		logrus.WithFields(logrus.Fields{
 			"BuildID":   buildID,
 			"taskQueue": identity,
-		}).Infof("Starting Temporal worker")
+		}).Info("Starting Temporal worker")
 
-		if err := newWorker.Start(); err != nil {
+		if err := w.Start(); err != nil {
 			logrus.WithError(err).
 				WithField("taskQueue", identity).
 				Error("Failed to start temporal worker")
+			delete(c.workers, identity)
 			continue
 		}
 
-		a.workers[identity] = newWorker
+		startedCount++
 	}
 
-	if len(a.workers) == 0 {
-		a.markReady() // Unblock any waiters even on failure
+	if startedCount == 0 {
+		c.markReady()
 		return fmt.Errorf("failed to start any Temporal workers")
 	}
 
+	c.workersStarted = true
+
 	// If versioning is enabled, confirm our deployment version is registered
 	// on the Temporal server before allowing workflow submissions via GetClient().
-	if a.config.DisableVersioning {
-		a.markReady()
+	if c.config.DisableVersioning {
+		c.markReady()
 	} else {
-		go a.awaitVersionRegistration(buildID)
+		go c.awaitVersionRegistration(buildID)
 	}
 
 	return nil
@@ -335,6 +375,7 @@ func (c *TemporalClient) Shutdown() error {
 
 	c.workers = nil
 	c.client = nil
+	c.workersStarted = false
 
 	return nil
 }

internal/config/services/temporal/multi_worker.go

@@ -13,7 +13,7 @@ import (
 // identity-specific task queues.
 //
 // Lifecycle methods (Start/Run/Stop) are no-ops because TemporalClient
-// manages worker lifecycle directly in Initialize() and Shutdown().
+// manages worker lifecycle directly via StartWorkers() and Shutdown().
 type multiWorker struct {
 	workers []worker.Worker
 }

internal/config/temporal.go

@@ -10,8 +10,7 @@ import (
 
 // Register temporal workflows and activities
 func (c *Config) registerTemporalWorkflows() error {
-
-	if c.GetServices() == nil || c.GetServices().GetTemporal() == nil {
+	if c.servicesClient == nil || c.servicesClient.GetTemporal() == nil {
 		return fmt.Errorf("temporal service is not initialized")
 	}
 
@@ -26,8 +25,7 @@ func (c *Config) registerTemporalWorkflows() error {
 }
 
 func (c *Config) registerTemporalActivities() error {
-
-	if c.GetServices() == nil || c.GetServices().GetTemporal() == nil {
+	if c.servicesClient == nil || c.servicesClient.GetTemporal() == nil {
 		return fmt.Errorf("temporal service is not initialized")
 	}
 

internal/models/temporal.go

@@ -61,6 +61,7 @@ func (t *TemporalConfig) HasMtlsConfig() bool {
 
 type TemporalImpl interface {
 	Initialize() error
+	StartWorkers() error
 	Shutdown() error
 
 	GetClient() client.Client

test/integration/services/temporal_test.go

@@ -61,6 +61,7 @@ func initAndRegister(t *testing.T, infra *testinfra.TestInfrastructure, cfg *mod
 	require.NotNil(t, w, "GetWorker must return a worker after Initialize")
 	w.RegisterWorkflow(echoWorkflow)
 	w.RegisterActivity(echoActivity)
+	require.NoError(t, tc.StartWorkers(), "StartWorkers should succeed")
 	return tc
 }