Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions internal/config/providers.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Comment thread
hughneale marked this conversation as resolved.
if err := c.SetupTemporal(); err != nil {
return fmt.Errorf("setting up temporal services: %w", err)
}
}

defs := c.GetProviders().Definitions

Expand Down Expand Up @@ -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
Expand Down
26 changes: 17 additions & 9 deletions internal/config/services.go
Original file line number Diff line number Diff line change
Expand Up @@ -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...")

Expand All @@ -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
}
59 changes: 50 additions & 9 deletions internal/config/services/temporal/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ type TemporalClient struct {
mu sync.Mutex
readyCh chan struct{}
closeReadyOnce sync.Once
workersStarted bool
}

func NewTemporalClient(
Expand Down Expand Up @@ -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()

Expand All @@ -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
Expand Down Expand Up @@ -335,6 +375,7 @@ func (c *TemporalClient) Shutdown() error {

c.workers = nil
c.client = nil
c.workersStarted = false

return nil
}
Expand Down
2 changes: 1 addition & 1 deletion internal/config/services/temporal/multi_worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down
6 changes: 2 additions & 4 deletions internal/config/temporal.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
}

Expand All @@ -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")
}

Expand Down
1 change: 1 addition & 0 deletions internal/models/temporal.go
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ func (t *TemporalConfig) HasMtlsConfig() bool {

type TemporalImpl interface {
Initialize() error
StartWorkers() error
Shutdown() error

GetClient() client.Client
Expand Down
1 change: 1 addition & 0 deletions test/integration/services/temporal_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand Down
Loading