From 36a59c30c03d97a87cd81364ecb3044519619294 Mon Sep 17 00:00:00 2001
From: Anirudh Oppiliappan <anirudh@tangled.org>
Date: Sun, 3 May 2026 23:57:51 +0300
Subject: [PATCH] spindle: limit concurrent workflows to cap total memory usage

Add a channel-based semaphore (SPINDLE_SERVER_MAX_CONCURRENT_WORKFLOWS,
default 8) that blocks workflow goroutines from starting a container until
a slot is free. Each workflow acquires a slot before SetupWorkflow and
releases it on exit via defer.

Combined with the 6 GiB per-container limit (MAX_JOB_MEMORY_MB), this
bounds total container memory to ~48 GiB on the current 64 GiB host.
MAX_JOB_COUNT still controls pipeline-level concurrency at the queue.
Without this semaphore, max workflows were effectively unbounded and
(wrongly) assumed to be capped at MAX_JOB_COUNT.

Signed-off-by: Anirudh Oppiliappan <anirudh@tangled.org>
---
 spindle/config/config.go |  3 ++-
 spindle/engine/engine.go |  6 +++++-
 spindle/server.go        | 35 ++++++++++++++++++++---------------
 3 files changed, 27 insertions(+), 17 deletions(-)

diff --git a/spindle/config/config.go b/spindle/config/config.go
index 02301e50..38493d04 100644
--- a/spindle/config/config.go
+++ b/spindle/config/config.go
@@ -19,7 +19,8 @@ type Server struct {
 	Secrets           Secrets `env:",prefix=SECRETS_"`
 	LogDir            string  `env:"LOG_DIR, default=/var/log/spindle"`
 	QueueSize         int     `env:"QUEUE_SIZE, default=100"`
-	MaxJobCount       int     `env:"MAX_JOB_COUNT, default=2"` // max number of jobs that run at a time
+	MaxJobCount              int `env:"MAX_JOB_COUNT, default=2"`               // max number of pipelines that run at a time
+	MaxConcurrentWorkflows   int `env:"MAX_CONCURRENT_WORKFLOWS, default=8"`    // max number of workflow containers running at once (memory cap)
 }
 
 func (s Server) Did() syntax.DID {
diff --git a/spindle/engine/engine.go b/spindle/engine/engine.go
index 5a5d7fcc..d09cb77f 100644
--- a/spindle/engine/engine.go
+++ b/spindle/engine/engine.go
@@ -21,7 +21,7 @@ var (
 	ErrWorkflowFailed = errors.New("workflow failed")
 )
 
-func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) {
+func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, workflowSem chan struct{}, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) {
 	l.Info("starting all workflows in parallel", "pipeline", pipelineId)
 
 	// extract secrets
@@ -81,6 +81,10 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, d
 					return
 				}
 
+				// acquire semaphore slot before starting the container
+				workflowSem <- struct{}{}
+				defer func() { <-workflowSem }()
+
 				err = eng.SetupWorkflow(ctx, wid, &w, wfLogger)
 				if err != nil {
 					// TODO(winter): Should this always set StatusFailed?
diff --git a/spindle/server.go b/spindle/server.go
index 09f35ebe..c2560920 100644
--- a/spindle/server.go
+++ b/spindle/server.go
@@ -48,9 +48,10 @@ type Spindle struct {
 	cfg    *config.Config
 	ks     *eventconsumer.Consumer
 	res    *idresolver.Resolver
-	vault  secrets.Manager
-	motd   []byte
-	motdMu sync.RWMutex
+	vault       secrets.Manager
+	motd        []byte
+	motdMu      sync.RWMutex
+	workflowSem chan struct{}
 }
 
 // New creates a new Spindle server with the provided configuration and engines.
@@ -98,6 +99,9 @@ func New(ctx context.Context, cfg *config.Config, engines map[string]models.Engi
 	jq := queue.NewQueue(cfg.Server.QueueSize, cfg.Server.MaxJobCount)
 	logger.Info("initialized queue", "queueSize", cfg.Server.QueueSize, "numWorkers", cfg.Server.MaxJobCount)
 
+	workflowSem := make(chan struct{}, cfg.Server.MaxConcurrentWorkflows)
+	logger.Info("initialized workflow semaphore", "maxConcurrentWorkflows", cfg.Server.MaxConcurrentWorkflows)
+
 	collections := []string{
 		tangled.SpindleMemberNSID,
 		tangled.RepoNSID,
@@ -121,17 +125,18 @@ func New(ctx context.Context, cfg *config.Config, engines map[string]models.Engi
 	resolver := idresolver.DefaultResolver(cfg.Server.PlcUrl)
 
 	spindle := &Spindle{
-		jc:    jc,
-		e:     e,
-		db:    d,
-		l:     logger,
-		n:     &n,
-		engs:  engines,
-		jq:    jq,
-		cfg:   cfg,
-		res:   resolver,
-		vault: vault,
-		motd:  defaultMotd,
+		jc:          jc,
+		e:           e,
+		db:          d,
+		l:           logger,
+		n:           &n,
+		engs:        engines,
+		jq:          jq,
+		cfg:         cfg,
+		res:         resolver,
+		vault:       vault,
+		motd:        defaultMotd,
+		workflowSem: workflowSem,
 	}
 
 	err = e.AddSpindle(rbacDomain)
@@ -393,7 +398,7 @@ func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source,
 
 		ok := s.jq.Enqueue(queue.Job{
 			Run: func() error {
-				engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, ctx, &models.Pipeline{
+				engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.workflowSem, ctx, &models.Pipeline{
 					RepoOwner: tpl.TriggerMetadata.Repo.Did,
 					RepoName:  repoName,
 					Workflows: workflows,
-- 
2.51.2