package engine
import (
"context"
"errors"
"io"
"log/slog"
"strings"
"testing"
"time"
"tangled.org/core/spindle/config"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/models"
"tangled.org/core/spindle/observability"
"tangled.org/core/spindle/quota"
)
const (
minutesOwner = "did:web:alice"
minutesRepo = "did:web:alice/repo"
)
func minutesQuotaManager(t *testing.T, database *db.DB, userLimit int64) (*quota.Manager, *db.QuotaStore) {
t.Helper()
qs := db.NewQuotaStore(database, quota.Defaults{
quota.ScopeUser: quota.Resources{quota.ResourceCIMinutesMonthly: userLimit},
})
qm := quota.NewManager(qs, 50*time.Millisecond, nil)
t.Cleanup(qm.Close)
return qm, qs
}
func runMinutesWorkflow(t *testing.T, qm *quota.Manager, database *db.DB, ctx context.Context, eng models.Engine, name string) models.WorkflowId {
t.Helper()
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
pipelineID := models.PipelineId("rkey")
pipeline := &models.Pipeline{
RepoDid: minutesRepo,
Workflows: map[models.Engine][]models.Workflow{
eng: {{
Name: name,
Steps: []models.Step{mockStep{name: "step1"}},
OwnerDID: minutesOwner,
RepoDID: minutesRepo,
}},
},
}
StartWorkflows(logger, nil, cfg, qm, nil, database, nil, nil, nil, ctx, pipeline, pipelineID)
return models.WorkflowId{PipelineId: pipelineID, Name: name}
}
func workflowError(t *testing.T, database *db.DB, wid models.WorkflowId) string {
t.Helper()
var msg string
err := database.QueryRow(`
select coalesce(error, '') from workflow_statuses
where pipeline_id = ? and workflow = ? and status = 'failed'
`, string(wid.PipelineId), wid.Name).Scan(&msg)
if err != nil {
t.Fatalf("no failed status for %s: %v", wid, err)
}
return msg
}
func TestStartWorkflows_DeniedWhenMonthlyMinutesAreUsedUp(t *testing.T) {
database := newTestDB(t)
qm, qs := minutesQuotaManager(t, database, 3)
if err := qs.RecordMinutes(context.Background(), "earlier", quota.Identity{OwnerDID: minutesOwner, RepoDID: minutesRepo}, 3); err != nil {
t.Fatal(err)
}
metrics := observability.NewMetrics()
ctx := observability.WithMetrics(context.Background(), metrics)
eng := &mockEngine{}
wid := runMinutesWorkflow(t, qm, database, ctx, eng, "job1")
eng.mu.Lock()
ran := len(eng.setupCalls) + len(eng.runStepCalls)
eng.mu.Unlock()
if ran != 0 {
t.Fatalf("denied workflow still reached the engine %d times", ran)
}
if st, err := database.GetStatus(wid); err != nil || st != models.StatusKindFailed {
t.Fatalf("status = %q, %v, want failed", st, err)
}
msg := workflowError(t, database, wid)
if !strings.HasPrefix(msg, "monthly CI minutes used up (owner): 3 of 3, resets ") || !strings.HasSuffix(msg, " 00:00 UTC") {
t.Fatalf("status message = %q", msg)
}
if !quotaDeniedRecorded(t, metrics) {
t.Fatal("denial was not attributed to quota_denied")
}
}
func quotaDeniedRecorded(t *testing.T, metrics *observability.Metrics) bool {
t.Helper()
families, err := metrics.Registry().Gather()
if err != nil {
t.Fatal(err)
}
for _, family := range families {
if family.GetName() != "spindle_workflow_terminations_total" {
continue
}
for _, metric := range family.GetMetric() {
labels := map[string]string{}
for _, label := range metric.GetLabel() {
labels[label.GetName()] = label.GetValue()
}
if labels["failure_class"] == string(FailureClassPolicy) && labels["reason"] == string(FailureReasonQuotaDenied) && metric.GetCounter().GetValue() == 1 {
return true
}
}
}
return false
}
type remoteSlotEngine struct {
*mockEngine
released chan struct{}
}
func (*remoteSlotEngine) AuthorsRemoteStatus() {}
func (e *remoteSlotEngine) AcquireWorkflowSlot(context.Context, models.WorkflowId, *models.Workflow, AcquireMode) (WorkflowSlot, error) {
return releaseFunc(func() { close(e.released) }), nil
}
func TestStartWorkflows_RemoteEngineDenialStillRecordsTheTerminalMetric(t *testing.T) {
database := newTestDB(t)
qm, qs := minutesQuotaManager(t, database, 3)
if err := qs.RecordMinutes(context.Background(), "earlier", quota.Identity{OwnerDID: minutesOwner, RepoDID: minutesRepo}, 3); err != nil {
t.Fatal(err)
}
metrics := observability.NewMetrics()
ctx := observability.WithMetrics(context.Background(), metrics)
eng := &remoteSlotEngine{mockEngine: &mockEngine{}, released: make(chan struct{})}
runMinutesWorkflow(t, qm, database, ctx, eng, "job1")
select {
case <-eng.released:
default:
t.Fatal("the remote slot was not released after the denial")
}
if !quotaDeniedRecorded(t, metrics) {
t.Fatal("a denial after the remote slot was taken recorded no quota_denied termination")
}
}
func TestStartWorkflows_RecordsMinutesWhenAWorkflowFinishes(t *testing.T) {
database := newTestDB(t)
qm, qs := minutesQuotaManager(t, database, 10)
eng := &mockEngine{}
wid := runMinutesWorkflow(t, qm, database, context.Background(), eng, "job1")
if st, err := database.GetStatus(wid); err != nil || st != models.StatusKindSuccess {
t.Fatalf("status = %q, %v, want success", st, err)
}
failing := &mockEngine{runStepFunc: func(context.Context, models.WorkflowId, int, models.WorkflowLogger) error {
return io.ErrUnexpectedEOF
}}
runMinutesWorkflow(t, qm, database, context.Background(), failing, "job2")
usages, err := qs.ListUsage(context.Background())
if err != nil {
t.Fatal(err)
}
got := map[quota.Scope]int64{}
for _, u := range usages {
if u.Resource == quota.ResourceCIMinutesMonthly {
got[u.Scope] = u.Used
}
}
// instant workflows still bill the one-minute floor
if got[quota.ScopeUser] != 2 || got[quota.ScopeRepo] != 2 {
t.Fatalf("minutes used = %v, want 2 on both the owner and the repo", got)
}
}
type flakyMinutesStore struct {
spyStore
failures int
recorded []string
}
func (s *flakyMinutesStore) CheckMinutes(context.Context, quota.Identity) error { return nil }
func (s *flakyMinutesStore) RecordMinutes(_ context.Context, usageID string, _ quota.Identity, _ int64) error {
if s.failures > 0 {
s.failures--
return errors.New("database is locked")
}
s.recorded = append(s.recorded, usageID)
return nil
}
func TestRecordMinutesRetriesAFailedRecord(t *testing.T) {
id := quota.Identity{OwnerDID: minutesOwner, RepoDID: minutesRepo}
store := &flakyMinutesStore{failures: recordMinutesAttempts - 1}
qm := quota.NewManager(store, 0, nil)
t.Cleanup(qm.Close)
if err := recordMinutes(context.Background(), qm, "wf", id, 1); err != nil || len(store.recorded) != 1 {
t.Fatalf("recordMinutes = %v, recorded %v, want it to succeed on the last attempt", err, store.recorded)
}
store = &flakyMinutesStore{failures: recordMinutesAttempts}
qm = quota.NewManager(store, 0, nil)
t.Cleanup(qm.Close)
if err := recordMinutes(context.Background(), qm, "wf", id, 1); err == nil || len(store.recorded) != 0 {
t.Fatalf("recordMinutes = %v, recorded %v, want the error once the attempts run out", err, store.recorded)
}
}