feat(updater): FIFO queue with single worker and per-job results
This commit is contained in:
@@ -0,0 +1,124 @@
|
||||
package updater
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/shcizo/package-updater/internal/discovery"
|
||||
)
|
||||
|
||||
// Status describes the outcome of executing a single Job.
|
||||
type Status string
|
||||
|
||||
const (
|
||||
StatusUpdated Status = "updated"
|
||||
StatusFailed Status = "failed"
|
||||
StatusRefused Status = "refused"
|
||||
StatusTimeout Status = "timeout"
|
||||
)
|
||||
|
||||
// Result captures a per-Job outcome to be surfaced in the HTTP response.
|
||||
type Result struct {
|
||||
Job discovery.Job
|
||||
Status Status
|
||||
Error string
|
||||
DurationMs int64
|
||||
}
|
||||
|
||||
// Queue serialises Job execution through a single worker so we never
|
||||
// run two `docker compose` commands against the same stack concurrently.
|
||||
// It also satisfies the "single global FIFO worker" design choice in
|
||||
// spec section 5.7.
|
||||
type Queue struct {
|
||||
exec Executor
|
||||
ch chan submission
|
||||
stop chan struct{}
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
type submission struct {
|
||||
ctx context.Context
|
||||
jobs []discovery.Job
|
||||
results chan []Result
|
||||
}
|
||||
|
||||
// NewQueue constructs a queue bound to the given executor.
|
||||
func NewQueue(exec Executor) *Queue {
|
||||
return &Queue{
|
||||
exec: exec,
|
||||
ch: make(chan submission, 16),
|
||||
stop: make(chan struct{}),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// Start launches the single background worker. Call Stop to terminate.
|
||||
func (q *Queue) Start(ctx context.Context) {
|
||||
go q.run(ctx)
|
||||
}
|
||||
|
||||
// Stop signals the worker to exit and waits for it to finish.
|
||||
func (q *Queue) Stop() {
|
||||
close(q.stop)
|
||||
<-q.done
|
||||
}
|
||||
|
||||
// Submit blocks until all jobs in the batch have been processed by the
|
||||
// worker, then returns per-job results in the same order. The ctx
|
||||
// timeout (if any) applies to each individual Job's execution and is
|
||||
// surfaced as StatusTimeout.
|
||||
func (q *Queue) Submit(ctx context.Context, jobs []discovery.Job) []Result {
|
||||
resCh := make(chan []Result, 1)
|
||||
q.ch <- submission{ctx: ctx, jobs: jobs, results: resCh}
|
||||
return <-resCh
|
||||
}
|
||||
|
||||
func (q *Queue) run(_ context.Context) {
|
||||
defer close(q.done)
|
||||
for {
|
||||
select {
|
||||
case <-q.stop:
|
||||
return
|
||||
case s := <-q.ch:
|
||||
results := make([]Result, len(s.jobs))
|
||||
for i, j := range s.jobs {
|
||||
results[i] = q.runOne(s.ctx, j)
|
||||
}
|
||||
s.results <- results
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (q *Queue) runOne(ctx context.Context, job discovery.Job) Result {
|
||||
start := time.Now()
|
||||
r := Result{Job: job}
|
||||
|
||||
if job.Refused {
|
||||
r.Status = StatusRefused
|
||||
r.Error = job.RefusedReason
|
||||
r.DurationMs = time.Since(start).Milliseconds()
|
||||
return r
|
||||
}
|
||||
|
||||
err := q.exec.Execute(ctx, job)
|
||||
r.DurationMs = time.Since(start).Milliseconds()
|
||||
switch {
|
||||
case err == nil:
|
||||
r.Status = StatusUpdated
|
||||
case errorsIsContextDeadline(err) || ctxDeadlineExceeded(ctx):
|
||||
r.Status = StatusTimeout
|
||||
r.Error = err.Error()
|
||||
default:
|
||||
r.Status = StatusFailed
|
||||
r.Error = err.Error()
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
func errorsIsContextDeadline(err error) bool {
|
||||
return err == context.DeadlineExceeded
|
||||
}
|
||||
|
||||
func ctxDeadlineExceeded(ctx context.Context) bool {
|
||||
return ctx.Err() == context.DeadlineExceeded
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
package updater_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/shcizo/package-updater/internal/discovery"
|
||||
"github.com/shcizo/package-updater/internal/updater"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type fakeExec struct {
|
||||
mu sync.Mutex
|
||||
calls []discovery.Job
|
||||
delay time.Duration
|
||||
errForSvc map[string]error
|
||||
}
|
||||
|
||||
func (f *fakeExec) Execute(ctx context.Context, job discovery.Job) error {
|
||||
f.mu.Lock()
|
||||
f.calls = append(f.calls, job)
|
||||
f.mu.Unlock()
|
||||
if f.delay > 0 {
|
||||
select {
|
||||
case <-time.After(f.delay):
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
if err, ok := f.errForSvc[job.Service]; ok {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeExec) callsCopy() []discovery.Job {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return append([]discovery.Job(nil), f.calls...)
|
||||
}
|
||||
|
||||
func TestQueue_ProcessesFIFO(t *testing.T) {
|
||||
exec := &fakeExec{delay: 20 * time.Millisecond}
|
||||
q := updater.NewQueue(exec)
|
||||
q.Start(context.Background())
|
||||
defer q.Stop()
|
||||
|
||||
job1 := discovery.Job{Project: "a", Service: "svc"}
|
||||
job2 := discovery.Job{Project: "b", Service: "svc"}
|
||||
job3 := discovery.Job{Project: "c", Service: "svc"}
|
||||
|
||||
results := make(chan []updater.Result, 3)
|
||||
go func() { results <- q.Submit(context.Background(), []discovery.Job{job1}) }()
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
go func() { results <- q.Submit(context.Background(), []discovery.Job{job2}) }()
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
go func() { results <- q.Submit(context.Background(), []discovery.Job{job3}) }()
|
||||
|
||||
for i := 0; i < 3; i++ {
|
||||
<-results
|
||||
}
|
||||
|
||||
calls := exec.callsCopy()
|
||||
require.Equal(t, []string{"a", "b", "c"}, []string{calls[0].Project, calls[1].Project, calls[2].Project})
|
||||
}
|
||||
|
||||
func TestQueue_ReturnsPerJobResults(t *testing.T) {
|
||||
exec := &fakeExec{errForSvc: map[string]error{"failing": errors.New("boom")}}
|
||||
q := updater.NewQueue(exec)
|
||||
q.Start(context.Background())
|
||||
defer q.Stop()
|
||||
|
||||
jobs := []discovery.Job{
|
||||
{Project: "p1", Service: "ok"},
|
||||
{Project: "p2", Service: "failing"},
|
||||
{Project: "p3", Service: "refused-svc", Refused: true, RefusedReason: "outside root"},
|
||||
}
|
||||
results := q.Submit(context.Background(), jobs)
|
||||
|
||||
require.Len(t, results, 3)
|
||||
require.Equal(t, updater.StatusUpdated, results[0].Status)
|
||||
require.Equal(t, updater.StatusFailed, results[1].Status)
|
||||
require.Contains(t, results[1].Error, "boom")
|
||||
require.Equal(t, updater.StatusRefused, results[2].Status)
|
||||
require.Contains(t, results[2].Error, "outside root")
|
||||
}
|
||||
|
||||
func TestQueue_Timeout(t *testing.T) {
|
||||
exec := &fakeExec{delay: 200 * time.Millisecond}
|
||||
q := updater.NewQueue(exec)
|
||||
q.Start(context.Background())
|
||||
defer q.Stop()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
jobs := []discovery.Job{{Project: "slow", Service: "svc"}}
|
||||
results := q.Submit(ctx, jobs)
|
||||
|
||||
require.Len(t, results, 1)
|
||||
require.Equal(t, updater.StatusTimeout, results[0].Status)
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
package updater
|
||||
|
||||
// Worker logic currently lives in queue.go (single in-process worker
|
||||
// goroutine). This file is reserved for the eventual per-stack-mutex
|
||||
// implementation called out in spec section 15 ("Multi-worker concurrency
|
||||
// with per-stack mutex"). Intentionally empty for v1.
|
||||
@@ -0,0 +1,38 @@
|
||||
package updater_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/shcizo/package-updater/internal/discovery"
|
||||
"github.com/shcizo/package-updater/internal/updater"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type counterExec struct{ n atomic.Int32 }
|
||||
|
||||
func (c *counterExec) Execute(_ context.Context, _ discovery.Job) error {
|
||||
c.n.Add(1)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestWorker_RunsExactlyOneAtATime(t *testing.T) {
|
||||
exec := &counterExec{}
|
||||
q := updater.NewQueue(exec)
|
||||
q.Start(context.Background())
|
||||
defer q.Stop()
|
||||
|
||||
jobs := make([]discovery.Job, 10)
|
||||
for i := range jobs {
|
||||
jobs[i] = discovery.Job{Project: "p", Service: "svc"}
|
||||
}
|
||||
start := time.Now()
|
||||
q.Submit(context.Background(), jobs)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
require.GreaterOrEqual(t, elapsed, 90*time.Millisecond)
|
||||
require.Equal(t, int32(10), exec.n.Load())
|
||||
}
|
||||
Reference in New Issue
Block a user