package engine
import (
"context"
"log/slog"
"os"
"sync"
"testing"
"time"
"tangled.org/core/spindle/config"
"tangled.org/core/spindle/models"
)
type slottedEngine struct {
*mockEngine
slots *SemaphoreSlotter
}
func (e *slottedEngine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode AcquireMode) (WorkflowSlot, error) {
return e.slots.AcquireWorkflowSlot(ctx, wid, wf, mode)
}
func TestQueuedWorkflowHasNoLogFileUntilItHoldsASlot(t *testing.T) {
testDB := newTestDB(t)
logDir := t.TempDir()
started := make(chan models.WorkflowId, 2)
release := make(chan struct{})
eng := &slottedEngine{mockEngine: &mockEngine{}, slots: NewSemaphoreSlotter(1)}
eng.runStepFunc = func(_ context.Context, wid models.WorkflowId, _ int, _ models.WorkflowLogger) error {
started <- wid
<-release
return nil
}
pipelineId := models.PipelineId("slot-log")
step := []models.Step{mockStep{name: "step"}}
pipeline := &models.Pipeline{Workflows: map[models.Engine][]models.Workflow{
eng: {{Name: "a", Steps: step}, {Name: "b", Steps: step}},
}}
cfg := &config.Config{Server: config.Server{LogDir: logDir}}
var wg sync.WaitGroup
wg.Go(func() {
StartWorkflows(slog.New(slog.NewTextHandler(os.Stderr, nil)), nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
})
holder := <-started
waiting := "a"
if holder.Name == "a" {
waiting = "b"
}
// nothing signals the waiter parked on the slot, this is just a wait
time.Sleep(300 * time.Millisecond)
if _, err := os.Stat(models.LogFilePath(logDir, models.WorkflowId{PipelineId: pipelineId, Name: waiting})); err == nil {
t.Errorf("workflow %q is waiting for a slot and already has a log file open", waiting)
}
close(release)
wg.Wait()
}