diff --git a/internal/updater/queue.go b/internal/updater/queue.go new file mode 100644 index 0000000..9f02da6 --- /dev/null +++ b/internal/updater/queue.go @@ -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 +} diff --git a/internal/updater/queue_test.go b/internal/updater/queue_test.go new file mode 100644 index 0000000..d346a46 --- /dev/null +++ b/internal/updater/queue_test.go @@ -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) +} diff --git a/internal/updater/worker.go b/internal/updater/worker.go new file mode 100644 index 0000000..7705bae --- /dev/null +++ b/internal/updater/worker.go @@ -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. diff --git a/internal/updater/worker_test.go b/internal/updater/worker_test.go new file mode 100644 index 0000000..7dcc8d4 --- /dev/null +++ b/internal/updater/worker_test.go @@ -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()) +}