mirror of
https://github.com/Dvorinka/Containr.git
synced 2026-10-28 07:41:27 +00:00
feat(deploy): per-service deploy queue, cancellable deployments, no-cache builds
- internal/deployqueue: serialized FIFO per service — concurrent deploys of one service no longer race (the container-name-conflict failure mode); different services still run in parallel. - Deploys and rollbacks enqueue through the queue; a busy service returns status 'queued' and promotes when the active job finishes. - POST /deployments/:id/cancel — queued jobs drop immediately (200 cancelled); active jobs get ctx cancelled so build/reconcile abort (202 cancelling, settles 'cancelled'); finished deployments return 409 NOT_CANCELLABLE. - Redeploy and variables save now route through the same queue so they can't overlap a build; PUT /services/:id/variables accepts redeploy:true to apply env changes to running containers. - no_cache on deployment create plumbs BuildRequest -> BuildOptions for nixpacks, dockerfile and railpack builders. - CLI: deployments cancel, deploy --no-cache, variables set/unset --redeploy. MCP: containr_deployments_cancel; deploy/variables_set document no_cache/redeploy. UI: Cancel button + queued/cancelled badges on the deployments table. OpenAPI + types regenerated. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
84d58cf808
commit
7839dc48a3
25 files changed
+721
-32
No files matched your search
@@ -0,0 +1,146 @@
|
||||
package deployqueue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// Job is a unit of serialized deployment work. DeploymentID is the
|
||||
// deployments row it mirrors (uuid.Nil for runtime ops with no row).
|
||||
// Run is invoked with a cancellable context — cancelled by Cancel for
|
||||
// active jobs, or the job is dropped before starting when queued.
|
||||
type Job struct {
|
||||
DeploymentID uuid.UUID
|
||||
Run func(ctx context.Context)
|
||||
}
|
||||
|
||||
// Queue serializes deploy work per service: one active job, the rest FIFO.
|
||||
// A build for service A never races a redeploy or second build of A, while
|
||||
// services B, C proceed in parallel. Not a global worker pool — per-service
|
||||
// serialization is what prevents container thrash.
|
||||
type Queue struct {
|
||||
mu sync.Mutex
|
||||
states map[uuid.UUID]*serviceState
|
||||
cancels map[uuid.UUID]context.CancelFunc
|
||||
}
|
||||
|
||||
type serviceState struct {
|
||||
running bool
|
||||
pending []Job
|
||||
}
|
||||
|
||||
func New() *Queue {
|
||||
return &Queue{
|
||||
states: make(map[uuid.UUID]*serviceState),
|
||||
cancels: make(map[uuid.UUID]context.CancelFunc),
|
||||
}
|
||||
}
|
||||
|
||||
// Enqueue schedules a job. Returns the position: 0 means it starts
|
||||
// immediately, >0 means that many jobs run first.
|
||||
func (q *Queue) Enqueue(serviceID uuid.UUID, job Job) int {
|
||||
q.mu.Lock()
|
||||
st := q.states[serviceID]
|
||||
if st == nil {
|
||||
st = &serviceState{}
|
||||
q.states[serviceID] = st
|
||||
}
|
||||
if st.running {
|
||||
st.pending = append(st.pending, job)
|
||||
pos := len(st.pending)
|
||||
q.mu.Unlock()
|
||||
return pos
|
||||
}
|
||||
st.running = true
|
||||
q.mu.Unlock()
|
||||
|
||||
q.start(serviceID, job)
|
||||
return 0
|
||||
}
|
||||
|
||||
func (q *Queue) start(serviceID uuid.UUID, job Job) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
if job.DeploymentID != uuid.Nil {
|
||||
q.mu.Lock()
|
||||
q.cancels[job.DeploymentID] = cancel
|
||||
q.mu.Unlock()
|
||||
}
|
||||
|
||||
go func() {
|
||||
defer func() {
|
||||
if job.DeploymentID != uuid.Nil {
|
||||
q.mu.Lock()
|
||||
delete(q.cancels, job.DeploymentID)
|
||||
q.mu.Unlock()
|
||||
}
|
||||
cancel()
|
||||
q.promote(serviceID)
|
||||
}()
|
||||
job.Run(ctx)
|
||||
}()
|
||||
}
|
||||
|
||||
// promote marks the service free and starts the next queued job, if any.
|
||||
func (q *Queue) promote(serviceID uuid.UUID) {
|
||||
q.mu.Lock()
|
||||
st := q.states[serviceID]
|
||||
if st == nil || len(st.pending) == 0 {
|
||||
if st != nil {
|
||||
st.running = false
|
||||
}
|
||||
q.mu.Unlock()
|
||||
return
|
||||
}
|
||||
next := st.pending[0]
|
||||
st.pending = st.pending[1:]
|
||||
q.mu.Unlock()
|
||||
q.start(serviceID, next)
|
||||
}
|
||||
|
||||
// CancelState distinguishes what Cancel found.
|
||||
type CancelState int
|
||||
|
||||
const (
|
||||
NotFound CancelState = iota
|
||||
WasQueued // removed before it started — caller marks DB cancelled
|
||||
WasActive // ctx cancelled — the running job marks DB itself
|
||||
)
|
||||
|
||||
// Cancel cancels a job by its deployment row id.
|
||||
func (q *Queue) Cancel(serviceID, deploymentID uuid.UUID) CancelState {
|
||||
q.mu.Lock()
|
||||
if st := q.states[serviceID]; st != nil {
|
||||
for i, j := range st.pending {
|
||||
if j.DeploymentID == deploymentID {
|
||||
st.pending = append(st.pending[:i], st.pending[i+1:]...)
|
||||
q.mu.Unlock()
|
||||
return WasQueued
|
||||
}
|
||||
}
|
||||
}
|
||||
cancel, ok := q.cancels[deploymentID]
|
||||
q.mu.Unlock()
|
||||
if ok {
|
||||
cancel()
|
||||
return WasActive
|
||||
}
|
||||
return NotFound
|
||||
}
|
||||
|
||||
// Do runs fn through the queue and waits for completion — for synchronous
|
||||
// endpoints (redeploy) that still must not overlap an in-flight deployment.
|
||||
// ctx bounds the wait, not the job; on timeout the job still runs to
|
||||
// completion to avoid orphaned reconcile state.
|
||||
func (q *Queue) Do(ctx context.Context, serviceID uuid.UUID, fn func(ctx context.Context)) {
|
||||
done := make(chan struct{})
|
||||
q.Enqueue(serviceID, Job{Run: func(jctx context.Context) {
|
||||
defer close(done)
|
||||
fn(jctx)
|
||||
}})
|
||||
select {
|
||||
case <-done:
|
||||
case <-ctx.Done():
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,142 @@
|
||||
package deployqueue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
func TestSerializesPerService(t *testing.T) {
|
||||
q := New()
|
||||
svc := uuid.New()
|
||||
|
||||
var active, maxActive int32
|
||||
finish := make(chan struct{})
|
||||
for i := 0; i < 3; i++ {
|
||||
q.Enqueue(svc, Job{Run: func(ctx context.Context) {
|
||||
if n := atomic.AddInt32(&active, 1); n > atomic.LoadInt32(&maxActive) {
|
||||
atomic.StoreInt32(&maxActive, n)
|
||||
}
|
||||
<-finish // hold until released
|
||||
atomic.AddInt32(&active, -1)
|
||||
}})
|
||||
}
|
||||
|
||||
deadline := time.After(2 * time.Second)
|
||||
for atomic.LoadInt32(&active) < 1 {
|
||||
select {
|
||||
case <-deadline:
|
||||
t.Fatal("first job never started")
|
||||
default:
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
close(finish)
|
||||
deadline = time.After(2 * time.Second)
|
||||
for atomic.LoadInt32(&active) > 0 {
|
||||
select {
|
||||
case <-deadline:
|
||||
t.Fatal("jobs did not drain")
|
||||
default:
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
if got := atomic.LoadInt32(&maxActive); got != 1 {
|
||||
t.Fatalf("expected max concurrency 1, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDifferentServicesRunInParallel(t *testing.T) {
|
||||
q := New()
|
||||
started := make(chan struct{}, 2)
|
||||
release := make(chan struct{})
|
||||
for i := 0; i < 2; i++ {
|
||||
q.Enqueue(uuid.New(), Job{Run: func(ctx context.Context) {
|
||||
started <- struct{}{}
|
||||
<-release
|
||||
}})
|
||||
}
|
||||
for i := 0; i < 2; i++ {
|
||||
select {
|
||||
case <-started:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("parallel services did not both start")
|
||||
}
|
||||
}
|
||||
close(release)
|
||||
}
|
||||
|
||||
func TestCancelQueuedRemovesJob(t *testing.T) {
|
||||
q := New()
|
||||
svc := uuid.New()
|
||||
release := make(chan struct{})
|
||||
q.Enqueue(svc, Job{DeploymentID: uuid.New(), Run: func(ctx context.Context) { <-release }})
|
||||
|
||||
queuedID := uuid.New()
|
||||
ran := make(chan struct{})
|
||||
q.Enqueue(svc, Job{DeploymentID: queuedID, Run: func(ctx context.Context) { close(ran) }})
|
||||
|
||||
if got := q.Cancel(svc, queuedID); got != WasQueued {
|
||||
t.Fatalf("expected WasQueued, got %v", got)
|
||||
}
|
||||
close(release)
|
||||
select {
|
||||
case <-ran:
|
||||
t.Fatal("cancelled queued job still ran")
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
|
||||
func TestCancelActiveCancelsContext(t *testing.T) {
|
||||
q := New()
|
||||
svc := uuid.New()
|
||||
id := uuid.New()
|
||||
ctxDone := make(chan struct{})
|
||||
q.Enqueue(svc, Job{DeploymentID: id, Run: func(ctx context.Context) {
|
||||
<-ctx.Done()
|
||||
close(ctxDone)
|
||||
}})
|
||||
|
||||
deadline := time.After(2 * time.Second)
|
||||
for {
|
||||
q.mu.Lock()
|
||||
_, ok := q.cancels[id]
|
||||
q.mu.Unlock()
|
||||
if ok {
|
||||
break
|
||||
}
|
||||
select {
|
||||
case <-deadline:
|
||||
t.Fatal("job never registered")
|
||||
default:
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
if got := q.Cancel(svc, id); got != WasActive {
|
||||
t.Fatalf("expected WasActive, got %v", got)
|
||||
}
|
||||
select {
|
||||
case <-ctxDone:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("active job context not cancelled")
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnqueuePosition(t *testing.T) {
|
||||
q := New()
|
||||
svc := uuid.New()
|
||||
release := make(chan struct{})
|
||||
if pos := q.Enqueue(svc, Job{Run: func(ctx context.Context) { <-release }}); pos != 0 {
|
||||
t.Fatalf("first job position = %d, want 0", pos)
|
||||
}
|
||||
if pos := q.Enqueue(svc, Job{Run: func(ctx context.Context) {}}); pos != 1 {
|
||||
t.Fatalf("second job position = %d, want 1", pos)
|
||||
}
|
||||
if pos := q.Enqueue(svc, Job{Run: func(ctx context.Context) {}}); pos != 2 {
|
||||
t.Fatalf("third job position = %d, want 2", pos)
|
||||
}
|
||||
close(release)
|
||||
}
|
||||
Reference in new issue
Block a user