package spindle
import (
"context"
"testing"
"time"
"tangled.org/core/spindle/models"
"tangled.org/core/spindle/observability"
"tangled.org/core/workflow"
)
func TestJobWorkerRetriesADequeueThatFailed(t *testing.T) {
s := newWipeTestSpindle(t)
s.metrics = observability.NewMetrics()
s.jobWake = make(chan struct{}, 1)
s.cfg.Server.MaxJobCount = 1
record := &models.PipelineRecord{ID: "queued", RepoDID: "did:plc:repo"}
if err := s.db.EnqueuePipeline(context.Background(), record.RepoDID, record, nil, workflow.PipelineSpec{}, false, "", "", nil); err != nil {
t.Fatal(err)
}
// renaming the table makes the first dequeue fail, like sqlite answering busy
if _, err := s.db.Exec(`alter table jobs rename to jobs_away`); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(func() {
cancel()
s.jobWorkers.Wait()
})
s.StartJobWorkers(ctx)
time.Sleep(100 * time.Millisecond)
if _, err := s.db.Exec(`alter table jobs_away rename to jobs`); err != nil {
t.Fatal(err)
}
eventually(t, 5*time.Second, "the queued job to be dequeued", func() bool {
var queued int
if err := s.db.QueryRow(`select count(*) from jobs`).Scan(&queued); err != nil {
t.Fatal(err)
}
return queued == 0
})
}