From 7839dc48a3d0293ed2230042c6bc11a677febbdb Mon Sep 17 00:00:00 2001 From: Tomas Dvorak Date: Thu, 8 Oct 2026 16:23:28 +0200 Subject: [PATCH 1/2] feat(deploy): per-service deploy queue, cancellable deployments, no-cache builds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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> --- AGENTS.md | 1 + app/backend/internal/api/deployments.go | 127 +++++++++++++-- app/backend/internal/api/routes.go | 7 + app/backend/internal/api/runtime.go | 75 ++++++++- app/backend/internal/api/system.go | 4 +- app/backend/internal/api/variables.go | 20 ++- app/backend/internal/build/dockerfile.go | 1 + app/backend/internal/build/nixpacks.go | 3 +- app/backend/internal/build/railpack.go | 1 + app/backend/internal/cli/commands/deploy.go | 29 +++- .../internal/cli/commands/variables.go | 23 ++- app/backend/internal/deployment/engine.go | 2 + app/backend/internal/deployqueue/queue.go | 146 ++++++++++++++++++ .../internal/deployqueue/queue_test.go | 142 +++++++++++++++++ app/backend/internal/docker/client.go | 1 + app/backend/internal/docker/types.go | 1 + app/backend/internal/mcpserver/server.go | 5 +- app/backend/internal/types/build.go | 1 + .../service/pages/ServiceDetailPage.tsx | 27 +++- app/frontend/src/generated/api-types.ts | 70 +++++++++ app/frontend/src/generated/docs-snapshot.ts | 2 +- app/frontend/src/lib/api-client.ts | 6 + docs/api/openapi.yaml | 45 ++++++ docs/guides/CLI.md | 11 +- llms.txt | 3 +- 25 files changed, 721 insertions(+), 32 deletions(-) create mode 100644 app/backend/internal/deployqueue/queue.go create mode 100644 app/backend/internal/deployqueue/queue_test.go diff --git a/AGENTS.md b/AGENTS.md index e13ea76..b3611d9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -28,6 +28,7 @@ Dev stack: Postgres on `:15432` (`containr-postgres-1`), Dragonfly on `:16379` ( ## Conventions that matter - New endpoints go in the right route group; mutations on `authed`/`admin` get idempotency for free. +- Deploy work goes through `deploy_queue` (`internal/deployqueue`) — per-service FIFO; enqueue jobs instead of `go`-ing deployment/reconcile goroutines so they never race a build. - New list endpoints use `pageWindow`/`paginationMeta` (`internal/api/pagination.go`) — cursor + page, `next_cursor` in the envelope. - New errors use stable codes via `respondError(c, status, code, msg)` in `internal/api/errors.go`. - New CLI command → also register an MCP tool in `internal/mcpserver/server.go` and document in OpenAPI. All three surfaces ship together. diff --git a/app/backend/internal/api/deployments.go b/app/backend/internal/api/deployments.go index fb2f71c..6c320b3 100644 --- a/app/backend/internal/api/deployments.go +++ b/app/backend/internal/api/deployments.go @@ -3,7 +3,9 @@ package api import ( "containr/internal/database" "containr/internal/deployment" + "containr/internal/deployqueue" "context" + "errors" "fmt" "net/http" "strings" @@ -34,6 +36,7 @@ type CreateDeploymentRequest struct { Branch string `json:"branch"` Trigger string `json:"trigger"` EnvVars map[string]string `json:"env_vars"` + NoCache bool `json:"no_cache"` } type DeploymentResponse struct { @@ -309,7 +312,23 @@ func handleCreateDeployment(c *gin.Context) { } engineInstance := engine.(*deployment.DeploymentEngine) - go runDeploymentAndSync(context.Background(), db.(*database.DB), engineInstance, &d, service, req, userID.(string)) + pos := getDeployQueue(c).Enqueue(serviceID, deployqueue.Job{ + DeploymentID: d.ID, + Run: func(jctx context.Context) { + runDeploymentAndSync(jctx, db.(*database.DB), engineInstance, &d, service, req, userID.(string)) + }, + }) + if pos > 0 { + d.Status = "queued" + _, _ = db.(*database.DB).Exec( + `UPDATE deployments SET status = 'queued', updated_at = $1 WHERE id = $2`, + time.Now(), d.ID, + ) + _, _ = db.(*database.DB).Exec( + `UPDATE services SET status = 'queued', updated_at = $1 WHERE id = $2`, + time.Now(), serviceID, + ) + } } c.JSON(http.StatusCreated, DeploymentResponse{ @@ -427,6 +446,7 @@ func runDeploymentAndSyncWithImage( SourcePath: sourcePath, Branch: req.Branch, Commit: req.CommitHash, + NoCache: req.NoCache, } } @@ -454,16 +474,19 @@ func runDeploymentAndSyncWithImage( select { case <-ctx.Done(): failedAt := time.Now() - timeoutErr := "Deployment timed out before completion" + finalStatus, finalErr := "failed", "Deployment timed out before completion" + if errors.Is(ctx.Err(), context.Canceled) { + finalStatus, finalErr = "cancelled", "Deployment cancelled" + } _, _ = db.Exec( `UPDATE deployments - SET status = 'failed', error = $1, completed_at = $2, updated_at = $2 - WHERE id = $3`, - timeoutErr, failedAt, dbDeployment.ID, + SET status = $1, error = $2, completed_at = $3, updated_at = $3 + WHERE id = $4`, + finalStatus, finalErr, failedAt, dbDeployment.ID, ) _, _ = db.Exec( - `UPDATE services SET status = 'failed', updated_at = $1 WHERE id = $2`, - failedAt, service.ID, + `UPDATE services SET status = $1, updated_at = $2 WHERE id = $3`, + finalStatus, failedAt, service.ID, ) return case <-syncTicker.C: @@ -530,13 +553,86 @@ func runDeploymentAndSyncWithImage( } func mapEngineStatusToDBStatus(status string) string { - switch status { - case "running": + if status == "running" { return "deployed" - case "cancelled": - return "failed" + } + return status +} + +// getDeployQueue returns the per-service deploy queue. SetupRoutes always +// sets it; the fallback keeps handler tests that skip SetupRoutes working. +func getDeployQueue(c *gin.Context) *deployqueue.Queue { + if v, exists := c.Get("deploy_queue"); exists { + if q, ok := v.(*deployqueue.Queue); ok && q != nil { + return q + } + } + return deployqueue.New() +} + +// handleCancelDeployment cancels a queued or in-flight deployment. Queued +// jobs are dropped and marked cancelled immediately; active jobs get their +// context cancelled so the build/reconcile aborts and the sync loop writes +// the terminal 'cancelled' status. +func handleCancelDeployment(c *gin.Context) { + db, exists := c.Get("db") + if !exists { + c.JSON(http.StatusInternalServerError, gin.H{"error": "Database connection not available", "code": "DEPENDENCY_UNAVAILABLE"}) + return + } + + deploymentID, err := uuid.Parse(c.Param("id")) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid deployment ID", "code": "VALIDATION"}) + return + } + userID, exists := c.Get("user_id") + if !exists { + c.JSON(http.StatusUnauthorized, gin.H{"error": "User not authenticated", "code": "UNAUTHENTICATED"}) + return + } + + var status, owner string + var serviceID uuid.UUID + err = db.(*database.DB).QueryRow( + `SELECT d.status, d.service_id, p.owner_id + FROM deployments d + JOIN services s ON d.service_id = s.id + JOIN projects p ON s.project_id = p.id + WHERE d.id = $1`, + deploymentID, + ).Scan(&status, &serviceID, &owner) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "Deployment not found", "code": "NOT_FOUND"}) + return + } + if owner != userID.(string) && !contextIsAdmin(c) { + c.JSON(http.StatusForbidden, gin.H{"error": "Access denied", "code": "FORBIDDEN"}) + return + } + + switch status { + case "deployed", "failed", "cancelled": + c.JSON(http.StatusConflict, gin.H{"error": "Deployment already finished", "code": "NOT_CANCELLABLE"}) + return + } + + switch getDeployQueue(c).Cancel(serviceID, deploymentID) { + case deployqueue.WasQueued: + now := time.Now() + _, _ = db.(*database.DB).Exec( + `UPDATE deployments SET status = 'cancelled', error = 'Deployment cancelled', completed_at = $1, updated_at = $1 WHERE id = $2`, + now, deploymentID, + ) + _, _ = db.(*database.DB).Exec( + `UPDATE services SET status = 'cancelled', updated_at = $1 WHERE id = $2`, + now, serviceID, + ) + c.JSON(http.StatusOK, gin.H{"status": "cancelled"}) + case deployqueue.WasActive: + c.JSON(http.StatusAccepted, gin.H{"status": "cancelling"}) default: - return status + c.JSON(http.StatusConflict, gin.H{"error": "Deployment is not running on this node", "code": "NOT_CANCELLABLE"}) } } @@ -735,7 +831,12 @@ func handleRollbackDeployment(c *gin.Context) { Trigger: "rollback", } // Reuse the stored image directly — no rebuild. - go runDeploymentAndSyncWithImage(context.Background(), db.(*database.DB), engine, &rollback, service, rollbackReq, userID.(string), targetImage) + getDeployQueue(c).Enqueue(serviceID, deployqueue.Job{ + DeploymentID: rollback.ID, + Run: func(jctx context.Context) { + runDeploymentAndSyncWithImage(jctx, db.(*database.DB), engine, &rollback, service, rollbackReq, userID.(string), targetImage) + }, + }) } c.JSON(http.StatusCreated, gin.H{ diff --git a/app/backend/internal/api/routes.go b/app/backend/internal/api/routes.go index 18ccd1c..960cd89 100644 --- a/app/backend/internal/api/routes.go +++ b/app/backend/internal/api/routes.go @@ -10,6 +10,7 @@ import ( "containr/internal/config" "containr/internal/database" "containr/internal/deployment" + "containr/internal/deployqueue" "containr/internal/docker" "containr/internal/ha" "containr/internal/metrics" @@ -98,6 +99,10 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg // Initialize database handler databaseHandler := NewDatabaseHandler(db.DB, dockerClient) + // Per-service serialized deploy queue — one active deployment per + // service, the rest wait as 'queued'. + deployQueue := deployqueue.New() + // Initialize security handler securityHandler := NewSecurityHandler(db, cfg.JWTSecret) @@ -115,6 +120,7 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg if deploymentEngine != nil { c.Set("deployment_engine", deploymentEngine) } + c.Set("deploy_queue", deployQueue) c.Set("scheduler", scheduler) c.Set("metrics_collector", metricsCollector) c.Set("auto_scaler", autoScaler) @@ -289,6 +295,7 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg authed.POST("/services/:id/redeploy", handleServiceRedeploy) authed.POST("/services/:id/deployments", handleCreateDeployment) authed.POST("/deployments/:id/rollback", handleRollbackDeployment) + authed.POST("/deployments/:id/cancel", handleCancelDeployment) // Environment variables — secrets, owner/admin only. authed.GET("/services/:id/variables", handleGetVariables) diff --git a/app/backend/internal/api/runtime.go b/app/backend/internal/api/runtime.go index 51fd126..dee02cf 100644 --- a/app/backend/internal/api/runtime.go +++ b/app/backend/internal/api/runtime.go @@ -11,6 +11,7 @@ import ( "containr/internal/database" "containr/internal/deployment" + "containr/internal/deployqueue" "github.com/gin-gonic/gin" "github.com/google/uuid" @@ -381,6 +382,7 @@ func handleServiceRestart(c *gin.Context) { // handleServiceRedeploy re-applies the current spec: pulls the image for // image-sourced services and reconciles containers (env, ports, replicas). +// Runs through the per-service deploy queue so it never overlaps a build. func handleServiceRedeploy(c *gin.Context) { engine, db, ok := getRuntimeEngine(c) if !ok { @@ -406,7 +408,78 @@ func handleServiceRedeploy(c *gin.Context) { } } service.Image = image - reconcileNow(c, engine, db, service) + + finished := make(chan struct{}) + var recErr error + var state *deployment.RuntimeState + getDeployQueue(c).Enqueue(service.ID, deployqueue.Job{Run: func(jctx context.Context) { + defer close(finished) + state, recErr = reconcileServiceJob(jctx, db, engine, service) + }}) + + select { + case <-finished: + if recErr != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": recErr.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"runtime": state}) + case <-c.Request.Context().Done(): + c.JSON(http.StatusAccepted, gin.H{"status": "queued"}) + } +} + +// reconcileServiceJob is the shared queue-job body for redeploy paths: +// build the spec from current DB state, reconcile containers, persist the +// resulting status and published port. +func reconcileServiceJob(ctx context.Context, db *database.DB, engine *deployment.DeploymentEngine, service Service) (*deployment.RuntimeState, error) { + spec, err := serviceRuntimeSpec(db, service) + if err != nil { + return nil, err + } + state, err := engine.ReconcileService(ctx, spec) + if err != nil { + return nil, err + } + if state != nil { + persistPublishedPort(db, service.ID, state.Ports) + status := "stopped" + if state.Status != "" { + status = state.Status + } + _, _ = db.Exec(`UPDATE services SET status = $1, updated_at = $2 WHERE id = $3`, status, time.Now(), service.ID) + } + return state, nil +} + +// enqueueServiceRedeploy resolves a deployable image (service.Image, else +// the last successful deployment's tag) and queues a reconcile. Returns +// false when no image is available or the engine is missing. +func enqueueServiceRedeploy(c *gin.Context, db *database.DB, service Service) bool { + engineValue, exists := c.Get("deployment_engine") + if !exists || engineValue == nil { + return false + } + engine, ok := engineValue.(*deployment.DeploymentEngine) + if !ok { + return false + } + image := service.Image + if image == "" { + if err := db.QueryRow( + `SELECT image_name || ':' || image_tag FROM deployments + WHERE service_id = $1 AND status = 'deployed' AND image_name <> '' + ORDER BY created_at DESC LIMIT 1`, + service.ID, + ).Scan(&image); err != nil { + return false + } + } + service.Image = image + getDeployQueue(c).Enqueue(service.ID, deployqueue.Job{Run: func(jctx context.Context) { + _, _ = reconcileServiceJob(jctx, db, engine, service) + }}) + return true } // persistPublishedPort records the host port Docker actually bound so later diff --git a/app/backend/internal/api/system.go b/app/backend/internal/api/system.go index b075867..0dfbccc 100644 --- a/app/backend/internal/api/system.go +++ b/app/backend/internal/api/system.go @@ -42,8 +42,8 @@ type hostMonitoringResponse struct { Docker map[string]any `json:"docker,omitempty"` // Scope names what the telemetry actually measures: the physical host for // native dockerd, or the Docker Desktop VM when the daemon runs in one. - Scope string `json:"scope"` - CollectedAt string `json:"collected_at"` + Scope string `json:"scope"` + CollectedAt string `json:"collected_at"` } type hostCPU struct { diff --git a/app/backend/internal/api/variables.go b/app/backend/internal/api/variables.go index 709b22a..6982b0f 100644 --- a/app/backend/internal/api/variables.go +++ b/app/backend/internal/api/variables.go @@ -25,6 +25,7 @@ type EnvironmentVariable struct { type UpdateVariablesRequest struct { Variables []VariableInput `json:"variables" binding:"required"` + Redeploy bool `json:"redeploy"` } type VariableInput struct { @@ -204,5 +205,22 @@ func handleUpdateVariables(c *gin.Context) { variables = append(variables, v) } - c.JSON(http.StatusOK, gin.H{"variables": variables, "message": "Environment variables updated successfully"}) + resp := gin.H{"variables": variables, "message": "Environment variables updated successfully"} + if req.Redeploy { + // Load the service spec inputs and queue a reconcile behind any + // in-flight deployment — new env lands in the recreated containers. + var svc Service + svcErr := db.(*database.DB).QueryRow( + `SELECT id, project_id, name, COALESCE(image, ''), COALESCE(command, ''), + COALESCE(replicas, 1), COALESCE(port, 0), COALESCE(domain, ''), + COALESCE(healthcheck_path, ''), COALESCE(restart_policy, 'unless-stopped'), + COALESCE(cpu, ''), COALESCE(memory, '') + FROM services WHERE id = $1`, + serviceID, + ).Scan(&svc.ID, &svc.ProjectID, &svc.Name, &svc.Image, &svc.Command, + &svc.Replicas, &svc.Port, &svc.Domain, &svc.HealthCheckPath, + &svc.RestartPolicy, &svc.CPU, &svc.Memory) + resp["redeploy_queued"] = svcErr == nil && enqueueServiceRedeploy(c, db.(*database.DB), svc) + } + c.JSON(http.StatusOK, resp) } diff --git a/app/backend/internal/build/dockerfile.go b/app/backend/internal/build/dockerfile.go index 73434dd..bfc91d1 100644 --- a/app/backend/internal/build/dockerfile.go +++ b/app/backend/internal/build/dockerfile.go @@ -222,6 +222,7 @@ func (d *DockerfileBuilder) Build(ctx context.Context, req *types.BuildRequest) Dockerfile: "Dockerfile", Tags: []string{imageName}, BuildArgs: buildArgs, + NoCache: req.NoCache, Remove: true, } diff --git a/app/backend/internal/build/nixpacks.go b/app/backend/internal/build/nixpacks.go index 32aa000..e349d5a 100644 --- a/app/backend/internal/build/nixpacks.go +++ b/app/backend/internal/build/nixpacks.go @@ -498,7 +498,8 @@ func (n *NixpacksBuilder) Build(ctx context.Context, req *types.BuildRequest) (* BuildArgs: map[string]*string{ "BUILDKIT_INLINE_CACHE": strPtr("1"), }, - Remove: true, + NoCache: req.NoCache, + Remove: true, } _, err = n.dockerClient.BuildImage(ctx, buildCtx, buildOptions) diff --git a/app/backend/internal/build/railpack.go b/app/backend/internal/build/railpack.go index 7da929b..ce33cfd 100644 --- a/app/backend/internal/build/railpack.go +++ b/app/backend/internal/build/railpack.go @@ -156,6 +156,7 @@ func (rb *RailpackBuilder) Build(ctx context.Context, req *types.BuildRequest) ( Tags: []string{imageName}, BuildArgs: buildArgs, Labels: req.Labels, + NoCache: req.NoCache, Remove: true, } diff --git a/app/backend/internal/cli/commands/deploy.go b/app/backend/internal/cli/commands/deploy.go index 6b6a32f..b94e272 100644 --- a/app/backend/internal/cli/commands/deploy.go +++ b/app/backend/internal/cli/commands/deploy.go @@ -17,6 +17,9 @@ var DeployCmd = &cobra.Command{ return err } body := map[string]interface{}{"trigger": "cli"} + if nc, _ := cmd.Flags().GetBool("no-cache"); nc { + body["no_cache"] = true + } for _, f := range []string{"commit-hash", "branch"} { if v, _ := cmd.Flags().GetString(f); v != "" { key := "commit_hash" @@ -106,6 +109,29 @@ var deploymentsLogsCmd = &cobra.Command{ RunE: runDeploymentLogs, } +var deploymentsCancelCmd = &cobra.Command{ + Use: "cancel ", + Short: "Cancel a queued or running deployment", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + c, err := client() + if err != nil { + return err + } + data, err := c.Do("POST", "/deployments/"+args[0]+"/cancel", map[string]interface{}{}) + if err != nil { + return err + } + if JSONMode() { + PrintRaw(data) + return nil + } + obj, _ := unwrapObject(data) + fmt.Printf("Deployment %s: %s\n", args[0], str(obj, "status")) + return nil + }, +} + var deploymentsRollbackCmd = &cobra.Command{ Use: "rollback ", Short: "Roll back to a deployment", @@ -131,5 +157,6 @@ var deploymentsRollbackCmd = &cobra.Command{ func init() { DeployCmd.Flags().String("commit-hash", "", "commit to deploy") DeployCmd.Flags().String("branch", "", "branch to deploy") - DeploymentsCmd.AddCommand(deploymentsListCmd, deploymentsGetCmd, deploymentsLogsCmd, deploymentsRollbackCmd) + DeployCmd.Flags().Bool("no-cache", false, "build without Docker layer cache") + DeploymentsCmd.AddCommand(deploymentsListCmd, deploymentsGetCmd, deploymentsLogsCmd, deploymentsRollbackCmd, deploymentsCancelCmd) } diff --git a/app/backend/internal/cli/commands/variables.go b/app/backend/internal/cli/commands/variables.go index 77afb7b..2e64895 100644 --- a/app/backend/internal/cli/commands/variables.go +++ b/app/backend/internal/cli/commands/variables.go @@ -49,7 +49,8 @@ var varsSetCmd = &cobra.Command{ Args: cobra.MinimumNArgs(2), RunE: func(cmd *cobra.Command, args []string) error { secret, _ := cmd.Flags().GetBool("secret") - return mutateVars(args[0], func(vars []map[string]interface{}) []map[string]interface{} { + redeploy, _ := cmd.Flags().GetBool("redeploy") + return mutateVars(args[0], redeploy, func(vars []map[string]interface{}) []map[string]interface{} { for _, pair := range args[1:] { k, v, found := strings.Cut(pair, "=") if !found || k == "" { @@ -71,7 +72,8 @@ var varsUnsetCmd = &cobra.Command{ for _, k := range args[1:] { drop[k] = true } - return mutateVars(args[0], func(vars []map[string]interface{}) []map[string]interface{} { + redeploy, _ := cmd.Flags().GetBool("redeploy") + return mutateVars(args[0], redeploy, func(vars []map[string]interface{}) []map[string]interface{} { out := vars[:0] for _, v := range vars { if !drop[str(v, "key")] { @@ -91,7 +93,7 @@ func fetchVars(c *Client, serviceID string) ([]map[string]interface{}, error) { return unwrapList(data, "variables") } -func mutateVars(serviceID string, mutate func([]map[string]interface{}) []map[string]interface{}) error { +func mutateVars(serviceID string, redeploy bool, mutate func([]map[string]interface{}) []map[string]interface{}) error { c, err := client() if err != nil { return err @@ -109,7 +111,11 @@ func mutateVars(serviceID string, mutate func([]map[string]interface{}) []map[st "is_secret": v["is_secret"] == true, }) } - data, err := c.Do("PUT", "/services/"+serviceID+"/variables", map[string]interface{}{"variables": payload}) + body := map[string]interface{}{"variables": payload} + if redeploy { + body["redeploy"] = true + } + data, err := c.Do("PUT", "/services/"+serviceID+"/variables", body) if err != nil { return err } @@ -117,7 +123,12 @@ func mutateVars(serviceID string, mutate func([]map[string]interface{}) []map[st PrintRaw(data) return nil } - fmt.Printf("Variables updated on %s — redeploy to apply to running containers\n", serviceID) + obj, _ := unwrapObject(data) + if obj["redeploy_queued"] == true { + fmt.Printf("Variables updated on %s — redeploy queued\n", serviceID) + } else { + fmt.Printf("Variables updated on %s — redeploy to apply to running containers\n", serviceID) + } return nil } @@ -134,5 +145,7 @@ func upsertVar(vars []map[string]interface{}, key, value string, secret bool) [] func init() { varsSetCmd.Flags().Bool("secret", false, "mark the variable(s) as secret") + varsSetCmd.Flags().Bool("redeploy", false, "redeploy the service after saving") + varsUnsetCmd.Flags().Bool("redeploy", false, "redeploy the service after saving") VariablesCmd.AddCommand(varsListCmd, varsSetCmd, varsUnsetCmd) } diff --git a/app/backend/internal/deployment/engine.go b/app/backend/internal/deployment/engine.go index 4cd265a..bc406c6 100644 --- a/app/backend/internal/deployment/engine.go +++ b/app/backend/internal/deployment/engine.go @@ -150,6 +150,7 @@ type BuildConfig struct { Environment map[string]string `json:"environment"` Branch string `json:"branch"` Commit string `json:"commit"` + NoCache bool `json:"no_cache"` } type TriggerConfig struct { @@ -291,6 +292,7 @@ func (de *DeploymentEngine) buildImage(ctx context.Context, deployment *Deployme TriggeredBy: "deployment_engine", Branch: buildConfig.Branch, Commit: buildConfig.Commit, + NoCache: buildConfig.NoCache, } response, err := de.buildManager.Build(ctx, buildReq) diff --git a/app/backend/internal/deployqueue/queue.go b/app/backend/internal/deployqueue/queue.go new file mode 100644 index 0000000..d6c583f --- /dev/null +++ b/app/backend/internal/deployqueue/queue.go @@ -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(): + } +} diff --git a/app/backend/internal/deployqueue/queue_test.go b/app/backend/internal/deployqueue/queue_test.go new file mode 100644 index 0000000..463934c --- /dev/null +++ b/app/backend/internal/deployqueue/queue_test.go @@ -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) +} diff --git a/app/backend/internal/docker/client.go b/app/backend/internal/docker/client.go index 3375108..deccf98 100644 --- a/app/backend/internal/docker/client.go +++ b/app/backend/internal/docker/client.go @@ -297,6 +297,7 @@ func (c *Client) BuildImage(ctx context.Context, buildContext io.Reader, options BuildArgs: options.BuildArgs, Labels: options.Labels, Remove: options.Remove, + NoCache: options.NoCache, }) } diff --git a/app/backend/internal/docker/types.go b/app/backend/internal/docker/types.go index c427b98..7e80196 100644 --- a/app/backend/internal/docker/types.go +++ b/app/backend/internal/docker/types.go @@ -44,6 +44,7 @@ type BuildOptions struct { BuildArgs map[string]*string Labels map[string]string Remove bool + NoCache bool } // NetworkConfig represents the configuration for creating a network diff --git a/app/backend/internal/mcpserver/server.go b/app/backend/internal/mcpserver/server.go index f809f90..48ad55e 100644 --- a/app/backend/internal/mcpserver/server.go +++ b/app/backend/internal/mcpserver/server.go @@ -64,15 +64,16 @@ func init() { // Variables register(spec{"containr_variables_list", "List service environment variables (secrets masked)", "GET", "/services/{id}/variables", []string{"id"}, nil, false}) - register(spec{"containr_variables_set", "Replace all variables on a service. Args: id, variables = [{key, value, is_secret}]. Fetch the current list first and merge — the server replaces wholesale.", "PUT", "/services/{id}/variables", []string{"id"}, []string{"variables"}, false}) + register(spec{"containr_variables_set", "Replace all variables on a service. Args: id, variables = [{key, value, is_secret}], redeploy (queue a redeploy to apply). Fetch the current list first and merge — the server replaces wholesale.", "PUT", "/services/{id}/variables", []string{"id"}, []string{"variables", "redeploy"}, false}) // Deployments - register(spec{"containr_deploy", "Trigger a deployment for a service. Args: id (service), commit_hash, branch.", "POST", "/services/{id}/deployments", []string{"id"}, []string{"commit_hash", "branch", "trigger"}, false}) + register(spec{"containr_deploy", "Trigger a deployment for a service. Args: id (service), commit_hash, branch, no_cache.", "POST", "/services/{id}/deployments", []string{"id"}, []string{"commit_hash", "branch", "trigger", "no_cache"}, false}) register(spec{"containr_deployments_list", "List recent deployments across services", "GET", "/deployments", nil, nil, false}) register(spec{"containr_services_deployments", "List deployments for one service", "GET", "/services/{id}/deployments", []string{"id"}, nil, false}) register(spec{"containr_deployments_get", "Get a deployment", "GET", "/deployments/{id}", []string{"id"}, nil, false}) register(spec{"containr_deployments_logs", "Get persisted build/runtime logs for a deployment. Args: id, type (all|build|runtime).", "GET", "/deployments/{id}/logs", []string{"id"}, nil, false}) register(spec{"containr_deployments_rollback", "Roll back to a deployment. Requires confirm:true.", "POST", "/deployments/{id}/rollback", []string{"id"}, nil, true}) + register(spec{"containr_deployments_cancel", "Cancel a queued or running deployment", "POST", "/deployments/{id}/cancel", []string{"id"}, nil, false}) // Databases register(spec{"containr_databases_list", "List managed databases", "GET", "/databases", nil, nil, false}) diff --git a/app/backend/internal/types/build.go b/app/backend/internal/types/build.go index 586eeb6..78f86e9 100644 --- a/app/backend/internal/types/build.go +++ b/app/backend/internal/types/build.go @@ -28,6 +28,7 @@ type BuildRequest struct { TriggeredBy string `json:"triggered_by"` // Who triggered the build Branch string `json:"branch"` // Git branch Commit string `json:"commit"` // Git commit SHA + NoCache bool `json:"no_cache"` // Disable Docker layer cache } // BuildResponse represents the response from a build operation diff --git a/app/frontend/src/features/service/pages/ServiceDetailPage.tsx b/app/frontend/src/features/service/pages/ServiceDetailPage.tsx index cfb6760..627233d 100644 --- a/app/frontend/src/features/service/pages/ServiceDetailPage.tsx +++ b/app/frontend/src/features/service/pages/ServiceDetailPage.tsx @@ -28,6 +28,7 @@ import { manualScaleService, execInService, rollbackDeployment, + cancelDeployment, updateServiceVariables, updateService, restartService, @@ -66,6 +67,7 @@ import { Sparkles, RefreshCw, Box, + Ban, Play, ChevronDown, GitPullRequest, @@ -104,7 +106,9 @@ function StatusBadge({ status }: { status: string }) { failed: { color: 'var(--error)', bg: 'var(--error-soft)', Icon: X, animate: false }, building: { color: 'var(--warning)', bg: 'var(--warning-soft)', Icon: Loader2, animate: true }, pending: { color: 'var(--warning)', bg: 'var(--warning-soft)', Icon: Loader2, animate: true }, + queued: { color: 'var(--warning)', bg: 'var(--warning-soft)', Icon: Clock, animate: false }, rolling_back: { color: 'var(--warning)', bg: 'var(--warning-soft)', Icon: RefreshCw, animate: true }, + cancelled: { color: 'var(--text-tertiary)', bg: 'var(--surface-muted)', Icon: Ban, animate: false }, stopped: { color: 'var(--text-tertiary)', bg: 'var(--surface-muted)', Icon: Box, animate: false }, }[status] || { color: 'var(--text-tertiary)', bg: 'var(--surface-muted)', Icon: Box, animate: false }; @@ -374,6 +378,14 @@ export function ServiceDetailPage() { }, }); + const cancelDeploymentMutation = useMutation({ + mutationFn: (deploymentId: string) => cancelDeployment(deploymentId), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: ['service-deployments', serviceId] }); + queryClient.invalidateQueries({ queryKey: ['service', serviceId] }); + }, + }); + const rollbackMutation = useMutation({ mutationFn: (deploymentId: string) => rollbackDeployment(deploymentId), onSuccess: () => { @@ -514,7 +526,7 @@ export function ServiceDetailPage() { const deployments = isDemoMode ? [] : deploymentsQuery.data ?? []; const total = deployments.length; const active = deployments.filter((deployment) => - ['pending', 'building', 'deploying', 'rolling_back'].includes(deployment.status), + ['queued', 'pending', 'building', 'deploying', 'rolling_back'].includes(deployment.status), ).length; const failed = deployments.filter((deployment) => deployment.status === 'failed').length; @@ -2121,6 +2133,19 @@ export function ServiceDetailPage() { > Logs + {['queued', 'pending', 'building', 'deploying', 'rolling_back'].includes(deployment.status) && ( + + )}