From 813c3b7655d2747b361554d5f3a8af47c0e96d0a Mon Sep 17 00:00:00 2001 From: vthwang Date: Thu, 15 Oct 2026 13:26:01 -0700 Subject: [PATCH 1/2] feat: recover failed full-stack deployments Allow sessions blocked by a bad deploy image to resume without rebuilding completed setup state. - add owner and admin recovery endpoints with image validation - resume each deploy stage from its exact continuation point - reconcile Kubernetes images and preserve crash-safe recovery Signed-off-by: vthwang --- internal/apidocs/openapi.yaml | 90 ++++++++++++ internal/handler/setup_admin.go | 10 +- internal/handler/setup_recovery.go | 169 +++++++++++++++++++++++ internal/handler/setup_recovery_test.go | 34 +++++ internal/k8s/upgrade.go | 30 ++++ internal/k8s/upgrade_test.go | 45 ++++++ internal/router/router.go | 2 + internal/setup/initial_provision.go | 4 +- internal/setup/orchestrator_fullstack.go | 18 ++- internal/setup/orchestrator_recovery.go | 145 +++++++++++++++++++ internal/setup/resume_coverage_test.go | 19 ++- 11 files changed, 556 insertions(+), 10 deletions(-) create mode 100644 internal/handler/setup_recovery.go create mode 100644 internal/handler/setup_recovery_test.go create mode 100644 internal/k8s/upgrade_test.go create mode 100644 internal/setup/orchestrator_recovery.go diff --git a/internal/apidocs/openapi.yaml b/internal/apidocs/openapi.yaml index 9f0e87c..2f4c3d5 100644 --- a/internal/apidocs/openapi.yaml +++ b/internal/apidocs/openapi.yaml @@ -903,6 +903,32 @@ components: dids: { type: string, example: "https://dids-myvta.example.com" } vtc: { type: string, example: "https://vtc-myvtc.example.com" } + RecoverSetupRequest: + type: object + required: [component, image] + properties: + component: + type: string + enum: [vta, mediator, dids, vtc] + description: Must match the component implied by the failed deploy stage. + image: + type: string + description: Full image URL returned by the component's setup images endpoint. + example: "ghcr.io/ic3software/vtc:0.3.1-917881c" + + RecoverSetupResponse: + type: object + required: [status, component, image] + properties: + status: + type: string + enum: [deploy_dids, deploy_mediator, deploy_vta, deploy_vtc] + component: + type: string + enum: [vta, mediator, dids, vtc] + image: + type: string + ModeAvailability: type: object description: Whether one setup mode can be created right now, and if not, why. @@ -2188,6 +2214,9 @@ paths: status: type: string example: "running" + failed_stage: + type: string + description: State-machine value that was active immediately before status became `failed`. error_msg: type: string description: Present when the session failed. @@ -3066,6 +3095,36 @@ paths: schema: $ref: "#/components/schemas/Error" + /api/v1/admin/setup-sessions/{id}/recover: + post: + summary: Recover a failed deployment (admin) + description: | + Admin-cookie twin of `POST /setup/{id}/recover`. The session must be a + full-stack session whose `failed_stage` is `deploy_dids`, + `deploy_mediator`, `deploy_vta`, or `deploy_vtc`. The replacement image + is persisted, the existing Deployment is reconciled to it, and setup + continues from the next step without repeating completed setup jobs. + tags: [Admin] + security: [{ CookieAuthAdmin: [] }] + parameters: + - { name: id, in: path, required: true, description: The session's `vta_name`., schema: { type: string } } + requestBody: + required: true + content: + application/json: + schema: { $ref: "#/components/schemas/RecoverSetupRequest" } + responses: + "202": + description: Recovery accepted and started + content: { application/json: { schema: { $ref: "#/components/schemas/RecoverSetupResponse" } } } + "400": { description: Component does not match the failed stage or image is unavailable, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "401": { description: Missing or invalid token, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "403": { description: Insufficient role, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "404": { description: Session not found, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "409": { description: Session is not at a recoverable deploy failure or has an active upgrade, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "502": { description: Failed to fetch image tags from GHCR, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "503": { description: K8s/orchestrator or image source not configured, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + /api/v1/admin/setup-sessions/{id}/dids/reissue-enroll: post: summary: Reissue the DID-hosting enrollment link (admin) @@ -5212,6 +5271,37 @@ paths: schema: $ref: "#/components/schemas/Error" + /api/v1/setup/{id}/recover: + post: + summary: Recover the caller's failed deployment + description: | + Changes the image for the component whose full-stack deployment failed, + retries only that deployment, and continues from the next setup step. + Supported mappings are `deploy_dids` → dids, `deploy_mediator` → + mediator, `deploy_vta` → vta, and `deploy_vtc` → vtc. Completed setup + jobs and PVC data are retained. The response returns immediately; poll + `GET /setup/{id}` for progress. A service restart also resumes from the + persisted deploy stage. + tags: [User] + security: [{ CookieAuthUser: [] }] + parameters: + - { name: id, in: path, required: true, schema: { type: string } } + requestBody: + required: true + content: + application/json: + schema: { $ref: "#/components/schemas/RecoverSetupRequest" } + responses: + "202": + description: Recovery accepted and started + content: { application/json: { schema: { $ref: "#/components/schemas/RecoverSetupResponse" } } } + "400": { description: Component does not match the failed stage or image is unavailable, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "401": { description: Missing or invalid token, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "404": { description: Session not found or owned by another user, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "409": { description: Session is not at a recoverable deploy failure or has an active upgrade, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "502": { description: Failed to fetch image tags from GHCR, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "503": { description: K8s/orchestrator or image source not configured, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + /api/v1/setup/{id}/upgrade: post: summary: Change the session's component images (self-service) diff --git a/internal/handler/setup_admin.go b/internal/handler/setup_admin.go index f15d456..4befd81 100644 --- a/internal/handler/setup_admin.go +++ b/internal/handler/setup_admin.go @@ -156,10 +156,11 @@ func (h *SetupHandler) AdminListSessions(c *gin.Context) { Mode string `json:"mode"` // managed | custom | platform — the platform row is the farm's own // stack and the one deletion that needs an explicit confirm. - DomainType string `json:"domain_type"` - Status string `json:"status"` - ErrorMsg string `json:"error_msg,omitempty"` - Fqdn string `json:"fqdn"` + DomainType string `json:"domain_type"` + Status string `json:"status"` + FailedStage string `json:"failed_stage,omitempty"` + ErrorMsg string `json:"error_msg,omitempty"` + Fqdn string `json:"fqdn"` // Per-component images — empty for components the mode doesn't run. VtaImage string `json:"vta_image,omitempty"` MediatorImage string `json:"mediator_image,omitempty"` @@ -190,6 +191,7 @@ func (h *SetupHandler) AdminListSessions(c *gin.Context) { Mode: s.Mode, DomainType: s.DomainType, Status: s.Status, + FailedStage: s.FailureStage(), ErrorMsg: s.ErrorMsg, Fqdn: s.FQDN(), VtaImage: s.VtaImage, diff --git a/internal/handler/setup_recovery.go b/internal/handler/setup_recovery.go new file mode 100644 index 0000000..22b483a --- /dev/null +++ b/internal/handler/setup_recovery.go @@ -0,0 +1,169 @@ +package handler + +import ( + "errors" + "net/http" + "slices" + "time" + + "github.com/gin-gonic/gin" + "gorm.io/gorm" + "gorm.io/gorm/clause" + + "github.com/ic3software/vtafarm-api/internal/ghcr" + "github.com/ic3software/vtafarm-api/internal/model" +) + +type recoverSetupRequest struct { + Component string `json:"component" binding:"required"` + Image string `json:"image" binding:"required"` +} + +type deploymentRecovery struct { + Stage string + Component string + ImageColumn string +} + +func recoveryForStage(stage string) (deploymentRecovery, bool) { + component := "" + switch stage { + case "deploy_dids": + component = "dids" + case "deploy_mediator": + component = "mediator" + case "deploy_vta": + component = "vta" + case "deploy_vtc": + component = "vtc" + default: + return deploymentRecovery{}, false + } + return deploymentRecovery{Stage: stage, Component: component, ImageColumn: model.UpgradeImageColumn(component)}, true +} + +func (h *SetupHandler) recoveryImageClient(component string) *ghcr.Client { + switch component { + case "vta": + return h.ghcr + case "mediator": + return h.mediatorGhcr + case "dids": + return h.didsGhcr + case "vtc": + return h.vtcGhcr + default: + return nil + } +} + +// RecoverSession retries a failed full-stack deployment owned by the caller. +func (h *SetupHandler) RecoverSession(c *gin.Context) { + if session := h.userSession(c); session != nil { + h.recoverSession(c, session) + } +} + +// AdminRecoverSession is the admin-cookie twin of RecoverSession. +func (h *SetupHandler) AdminRecoverSession(c *gin.Context) { + if session := h.adminSession(c); session != nil { + h.recoverSession(c, session) + } +} + +var errRecoveryConflict = errors.New("session is not recoverable") + +func (h *SetupHandler) recoverSession(c *gin.Context, session *model.SetupSession) { + if h.k8s == nil || h.orch == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "K8s not configured — recovery unavailable"}) + return + } + + var req recoverSetupRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + recovery, ok := recoveryForStage(session.FailureStage()) + if session.Mode != model.ModeFullStack || !ok { + c.JSON(http.StatusConflict, gin.H{"error": "only full-stack sessions failed at a deploy stage can be recovered"}) + return + } + if req.Component != recovery.Component { + c.JSON(http.StatusBadRequest, gin.H{"error": "failed stage " + recovery.Stage + " requires component " + recovery.Component}) + return + } + + client := h.recoveryImageClient(recovery.Component) + if client == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "image source not configured for " + recovery.Component}) + return + } + tags, err := client.ListTags(c.Request.Context()) + if err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": "failed to fetch images: " + err.Error()}) + return + } + if !slices.ContainsFunc(tags, func(tag ghcr.ImageTag) bool { return tag.Image == req.Image }) { + c.JSON(http.StatusBadRequest, gin.H{"error": "image is not available for " + recovery.Component}) + return + } + + conflictMessage := "session is no longer recoverable" + err = h.db.WithContext(c.Request.Context()).Transaction(func(tx *gorm.DB) error { + var locked model.SetupSession + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&locked, session.ID).Error; err != nil { + return err + } + lockedRecovery, ok := recoveryForStage(locked.FailureStage()) + if locked.Mode != model.ModeFullStack || !ok { + return errRecoveryConflict + } + if lockedRecovery.Component != req.Component { + conflictMessage = "session failure stage changed; reload before recovering" + return errRecoveryConflict + } + + var activeUpgrades int64 + if err := tx.Model(&model.UpgradeTask{}). + Where("session_id = ? AND status IN ?", locked.ID, []string{ + model.UpgradeTaskPending, model.UpgradeTaskRunning, model.UpgradeTaskRollingBack, + }).Count(&activeUpgrades).Error; err != nil { + return err + } + if activeUpgrades > 0 { + conflictMessage = "an upgrade or rollback for this session is already in progress" + return errRecoveryConflict + } + + result := tx.Model(&locked).Where("status = ?", "failed").Updates(map[string]any{ + lockedRecovery.ImageColumn: req.Image, + "status": lockedRecovery.Stage, + "error_msg": "", + "failed_stage": "", + "updated_at": time.Now(), + }) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return errRecoveryConflict + } + recovery = lockedRecovery + return nil + }) + if errors.Is(err, errRecoveryConflict) { + c.JSON(http.StatusConflict, gin.H{"error": conflictMessage}) + return + } + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "could not start recovery"}) + return + } + + h.orch.RecoverDeployment(session.ID) + c.JSON(http.StatusAccepted, gin.H{ + "status": recovery.Stage, "component": recovery.Component, "image": req.Image, + }) +} diff --git a/internal/handler/setup_recovery_test.go b/internal/handler/setup_recovery_test.go new file mode 100644 index 0000000..d7461ec --- /dev/null +++ b/internal/handler/setup_recovery_test.go @@ -0,0 +1,34 @@ +package handler + +import "testing" + +func TestRecoveryForStage(t *testing.T) { + tests := []struct { + stage, component, column string + }{ + {"deploy_dids", "dids", "dids_image"}, + {"deploy_mediator", "mediator", "mediator_image"}, + {"deploy_vta", "vta", "vta_image"}, + {"deploy_vtc", "vtc", "vtc_image"}, + } + + for _, test := range tests { + t.Run(test.stage, func(t *testing.T) { + got, ok := recoveryForStage(test.stage) + if !ok { + t.Fatalf("recoveryForStage(%q) was not recoverable", test.stage) + } + if got.Stage != test.stage || got.Component != test.component || got.ImageColumn != test.column { + t.Fatalf("recoveryForStage(%q) = %#v", test.stage, got) + } + }) + } +} + +func TestRecoveryForStageRejectsNonDeploymentStages(t *testing.T) { + for _, stage := range []string{"", "failed", "step_vtc_setup", "running"} { + if _, ok := recoveryForStage(stage); ok { + t.Errorf("recoveryForStage(%q) unexpectedly recoverable", stage) + } + } +} diff --git a/internal/k8s/upgrade.go b/internal/k8s/upgrade.go index 5653cfa..08b5640 100644 --- a/internal/k8s/upgrade.go +++ b/internal/k8s/upgrade.go @@ -3,12 +3,16 @@ package k8s import ( "context" "fmt" + "time" appsv1 "k8s.io/api/apps/v1" + k8serrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/util/retry" ) +const recoveryRestartAnnotation = "vtafarm.ic3software.com/recovery-restarted-at" + // VtaDeploymentName exposes vta_only's Deployment name (vta-) to // the upgrade runner, which patches Deployments created by CreateVtaDeployment. // full_stack components use the exported FS*Name funcs directly. @@ -37,6 +41,32 @@ func (c *Client) SetDeploymentImage(ctx context.Context, ns, name, image string) }) } +// RecoverDeploymentImage reconciles an existing failed Deployment to image +// and changes its pod template even when image is unchanged. A deployment can +// be absent when the original create call failed before persistence; that case +// is left to the normal idempotent create helper used immediately afterwards. +func (c *Client) RecoverDeploymentImage(ctx context.Context, ns, name, image string) error { + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + deploy, err := c.kube.AppsV1().Deployments(ns).Get(ctx, name, metav1.GetOptions{}) + if k8serrors.IsNotFound(err) { + return nil + } + if err != nil { + return err + } + deploy.Spec.Strategy = appsv1.DeploymentStrategy{Type: appsv1.RecreateDeploymentStrategyType} + for i := range deploy.Spec.Template.Spec.Containers { + deploy.Spec.Template.Spec.Containers[i].Image = image + } + if deploy.Spec.Template.Annotations == nil { + deploy.Spec.Template.Annotations = map[string]string{} + } + deploy.Spec.Template.Annotations[recoveryRestartAnnotation] = time.Now().UTC().Format(time.RFC3339Nano) + _, err = c.kube.AppsV1().Deployments(ns).Update(ctx, deploy, metav1.UpdateOptions{}) + return err + }) +} + // RolloutStatus is a point-in-time view of a Deployment rollout. Reason is a // pod-level explanation (ImagePullBackOff, CrashLoopBackOff, …) when the // rollout is stuck — empty while pods are progressing normally. diff --git a/internal/k8s/upgrade_test.go b/internal/k8s/upgrade_test.go new file mode 100644 index 0000000..4530f9e --- /dev/null +++ b/internal/k8s/upgrade_test.go @@ -0,0 +1,45 @@ +package k8s + +import ( + "context" + "testing" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +func TestRecoverDeploymentImageReplacesImageAndForcesRestart(t *testing.T) { + deployment := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: "vtc", Namespace: "test"}, + Spec: appsv1.DeploymentSpec{Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "vtc", Image: "example/vtc:bad"}}}, + }}, + } + client := &Client{kube: fake.NewSimpleClientset(deployment)} + + if err := client.RecoverDeploymentImage(context.Background(), "test", "vtc", "example/vtc:good"); err != nil { + t.Fatalf("RecoverDeploymentImage() error = %v", err) + } + got, err := client.kube.AppsV1().Deployments("test").Get(context.Background(), "vtc", metav1.GetOptions{}) + if err != nil { + t.Fatalf("get Deployment: %v", err) + } + if got.Spec.Template.Spec.Containers[0].Image != "example/vtc:good" { + t.Fatalf("image = %q", got.Spec.Template.Spec.Containers[0].Image) + } + if got.Spec.Strategy.Type != appsv1.RecreateDeploymentStrategyType { + t.Fatalf("strategy = %q", got.Spec.Strategy.Type) + } + if got.Spec.Template.Annotations[recoveryRestartAnnotation] == "" { + t.Fatal("recovery restart annotation is empty") + } +} + +func TestRecoverDeploymentImageAllowsMissingDeployment(t *testing.T) { + client := &Client{kube: fake.NewSimpleClientset()} + if err := client.RecoverDeploymentImage(context.Background(), "test", "missing", "example/vtc:good"); err != nil { + t.Fatalf("RecoverDeploymentImage() error = %v", err) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 99134d0..bde4e00 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -164,6 +164,7 @@ func Setup( adminAuth.GET("/admin/resource-defaults", sh.AdminResourceDefaults) adminAuth.PUT("/admin/resource-defaults", sh.AdminSaveResourceDefaults) adminAuth.GET("/admin/setup-sessions", sh.AdminListSessions) + adminAuth.POST("/admin/setup-sessions/:id/recover", sh.AdminRecoverSession) adminAuth.GET("/admin/setup-sessions/:id/resources", sh.AdminSessionResources) adminAuth.PUT("/admin/setup-sessions/resources", sh.AdminApplySessionResources) adminAuth.POST("/admin/load-tests", sh.AdminCreateLoadTest) @@ -288,6 +289,7 @@ func Setup( userAuth.POST("/setup", sh.Create) userAuth.POST("/setup/connection/inspect", middleware.RateLimit(20, time.Minute), sh.InspectConnection) userAuth.GET("/setup/:id", sh.Get) + userAuth.POST("/setup/:id/recover", sh.RecoverSession) userAuth.DELETE("/setup/:id", sh.Delete) userAuth.GET("/setup/:id/logs", sh.Logs) userAuth.GET("/setup/:id/did-log", sh.DownloadDIDLog) diff --git a/internal/setup/initial_provision.go b/internal/setup/initial_provision.go index 55e99d9..02efb00 100644 --- a/internal/setup/initial_provision.go +++ b/internal/setup/initial_provision.go @@ -11,8 +11,8 @@ import ( ) var provisionStatuses = []string{ - "provisioning", "step_import_admin_did", "deploy_vta", "step_vtc_setup_key", - "step_vtc_acl_grant", "step_vtc_setup", "deploy_vtc", + "provisioning", "step_import_admin_did", "step_vtc_setup_key", + "step_vtc_acl_grant", "step_vtc_setup", } func (o *Orchestrator) provisionAdminLabel(sessionID uint, adminDid, fallback string) string { diff --git a/internal/setup/orchestrator_fullstack.go b/internal/setup/orchestrator_fullstack.go index 9e5f2c3..4ca3c03 100644 --- a/internal/setup/orchestrator_fullstack.go +++ b/internal/setup/orchestrator_fullstack.go @@ -1200,17 +1200,20 @@ var ( "step_vta_setup", "step_mediator_p1", "step_mediator_reprov", "step_mediator_p2", "step_dids_p1", "step_dids_provision", "step_dids_p2", "step_dids_invite", - "step_dids_load_did", "step_dids_grant_farm", "deploy_dids", "deploy_mediator", + "step_dids_load_did", "step_dids_grant_farm", "step_vta_register_dids", } // fsResumePostGate: past the admin DID gate, so Provision continues from // there rather than rebuilding what already exists. fsResumePostGate = []string{ - "step_import_admin_did", "deploy_vta", - "step_vtc_setup_key", "step_vtc_acl_grant", "step_vtc_setup", "deploy_vtc", + "step_import_admin_did", "step_vtc_setup_key", "step_vtc_acl_grant", "step_vtc_setup", } + // Deployment stages resume exactly at the failed/current deployment. Going + // through Start or Provision would repeat setup jobs that already succeeded. + fsResumeDeployment = []string{"deploy_dids", "deploy_mediator", "deploy_vta", "deploy_vtc"} + // fsResumeTerminal: nothing is owed. awaiting_admin_did is only terminal // while admin_did is empty — resumeFullStack queries that case separately, // since supplying the DID is what makes it resumable. @@ -1238,6 +1241,15 @@ func (o *Orchestrator) resumeFullStack() { o.Provision(s.ID, s.AdminDid) } + var deployments []model.SetupSession + if err := o.db.Where("mode = ? AND status IN ?", model.ModeFullStack, fsResumeDeployment).Find(&deployments).Error; err != nil { + log.Printf("[orchestrator] resume full_stack deployments: query failed: %v", err) + } + for _, s := range deployments { + log.Printf("[orchestrator] resuming %s deployment %d (status=%s)", s.Mode, s.ID, s.Status) + o.RecoverDeployment(s.ID) + } + var gatedWithAdmin []model.SetupSession if err := o.db.Where("mode = ? AND status = ? AND admin_did != ''", model.ModeFullStack, "awaiting_admin_did").Find(&gatedWithAdmin).Error; err != nil { log.Printf("[orchestrator] resume full_stack: query failed: %v", err) diff --git a/internal/setup/orchestrator_recovery.go b/internal/setup/orchestrator_recovery.go new file mode 100644 index 0000000..834d670 --- /dev/null +++ b/internal/setup/orchestrator_recovery.go @@ -0,0 +1,145 @@ +package setup + +import ( + "context" + "fmt" + "log" + "time" + + "github.com/ic3software/vtafarm-api/internal/k8s" + "github.com/ic3software/vtafarm-api/internal/model" +) + +// RecoverDeployment retries only the failed deployment, then continues from +// the next state-machine step. It is intentionally separate from image +// upgrades: these sessions never reached running, so there is no healthy +// deployment to roll back to. +func (o *Orchestrator) RecoverDeployment(sessionID uint) { + o.launch(sessionID, func(ctx context.Context) { + o.recoverDeployment(ctx, sessionID) + }) +} + +func (o *Orchestrator) recoverDeployment(ctx context.Context, sessionID uint) { + var session model.SetupSession + if err := o.db.First(&session, sessionID).Error; err != nil { + log.Printf("[orchestrator] deployment recovery %d: load failed: %v", sessionID, err) + return + } + if session.Mode != model.ModeFullStack { + return + } + + s := &session + ns := o.k8s.UserNamespace(fmt.Sprintf("%d", s.UserID)) + fail := func(stage, prefix string, err error) bool { + if err == nil { + return false + } + if ctx.Err() != nil { + return true + } + o.markFailedAt(sessionID, stage, prefix+": "+err.Error()) + return true + } + + switch s.Status { + case "deploy_dids": + if fail("deploy_dids", "failed to reconcile dids deployment image", o.k8s.RecoverDeploymentImage(ctx, ns, k8s.FSDidsName(s.ID), s.DidsImage)) { + return + } + if fail("deploy_dids", "failed to deploy dids daemon", o.fsDeployDids(ctx, ns, s)) { + return + } + o.fsSetStatus(sessionID, "deploy_mediator") + if fail("deploy_mediator", "failed to deploy mediator", o.fsDeployMediator(ctx, ns, s)) { + return + } + o.continueRecoveredPreGate(ctx, ns, s) + + case "deploy_mediator": + if fail("deploy_mediator", "failed to reconcile mediator deployment image", o.k8s.RecoverDeploymentImage(ctx, ns, k8s.FSMediatorName(s.ID), s.MediatorImage)) { + return + } + if fail("deploy_mediator", "failed to deploy mediator", o.fsDeployMediator(ctx, ns, s)) { + return + } + o.continueRecoveredPreGate(ctx, ns, s) + + case "deploy_vta": + if fail("deploy_vta", "failed to reconcile vta deployment image", o.k8s.RecoverDeploymentImage(ctx, ns, k8s.FSVtaName(s.ID), s.VtaImage)) { + return + } + if fail("deploy_vta", "failed to deploy vta", o.fsDeployVta(ctx, ns, s)) { + return + } + o.continueRecoveredPostGate(ctx, ns, s, fail) + + case "deploy_vtc": + if fail("deploy_vtc", "failed to reconcile vtc deployment image", o.k8s.RecoverDeploymentImage(ctx, ns, k8s.FSVtcName(s.ID), s.VtcImage)) { + return + } + if fail("deploy_vtc", "failed to deploy vtc", o.fsDeployVtc(ctx, ns, s)) { + return + } + o.finishRecoveredSession(sessionID, s) + } +} + +func (o *Orchestrator) continueRecoveredPreGate(ctx context.Context, ns string, s *model.SetupSession) { + o.fsSetStatus(s.ID, "step_vta_register_dids") + if err := o.fsStepVtaRegisterDids(ctx, ns, s); err != nil { + if ctx.Err() == nil { + o.markFailedAt(s.ID, "step_vta_register_dids", "vta register-dids failed: "+err.Error()) + } + return + } + if s.AdminDid != "" { + // The DID was accepted before this deployment failed. Do not send it + // through AcceptManual again; restore a post-gate status and let the + // crash-safe provision worker continue with the persisted DID. + o.fsSetStatus(s.ID, "step_import_admin_did") + o.KickProvision(s.ID) + return + } + o.fsSetStatus(s.ID, "awaiting_admin_did") + log.Printf("[orchestrator] recovered full-stack session %d: awaiting admin DID", s.ID) +} + +func (o *Orchestrator) continueRecoveredPostGate( + ctx context.Context, + ns string, + s *model.SetupSession, + fail func(stage, prefix string, err error) bool, +) { + o.fsSetStatus(s.ID, "step_vtc_setup") + outcome, err := o.fsStepVtcSetup(ctx, ns, s) + if fail("step_vtc_setup", "vtc setup failed", err) { + return + } + s.VtcDid, s.VtcAdminDid = outcome.VtcDid, outcome.AdminDid + s.VtcInstallURL, s.VtcClaimCode = outcome.InstallURL, outcome.ClaimCode + if err := o.db.WithContext(ctx).Model(&model.SetupSession{}).Where("id = ?", s.ID).Updates(map[string]any{ + "vtc_did": outcome.VtcDid, "vtc_admin_did": outcome.AdminDid, + "vtc_install_url": outcome.InstallURL, "vtc_claim_code": outcome.ClaimCode, + "vtc_install_used": false, "updated_at": time.Now(), + }).Error; err != nil { + if ctx.Err() == nil { + o.markFailedAt(s.ID, "step_vtc_setup", "persist vtc setup output failed: "+err.Error()) + } + return + } + + o.fsSetStatus(s.ID, "deploy_vtc") + if fail("deploy_vtc", "failed to deploy vtc", o.fsDeployVtc(ctx, ns, s)) { + return + } + o.finishRecoveredSession(s.ID, s) +} + +func (o *Orchestrator) finishRecoveredSession(sessionID uint, s *model.SetupSession) { + o.db.Model(&model.SetupSession{}).Where("id = ?", sessionID).Updates(map[string]any{ + "status": "running", "error_msg": "", "failed_stage": "", "updated_at": time.Now(), + }) + log.Printf("[orchestrator] recovered full-stack session %d: running at https://%s", sessionID, s.VtcFQDN()) +} diff --git a/internal/setup/resume_coverage_test.go b/internal/setup/resume_coverage_test.go index 2cd7ade..45b32bd 100644 --- a/internal/setup/resume_coverage_test.go +++ b/internal/setup/resume_coverage_test.go @@ -75,6 +75,9 @@ func TestResumeCoversEveryStatus(t *testing.T) { for _, s := range fsResumePostGate { classified[s] = "fsResumePostGate" } + for _, s := range fsResumeDeployment { + classified[s] = "fsResumeDeployment" + } for _, s := range fsResumeTerminal { classified[s] = "fsResumeTerminal" } @@ -84,7 +87,7 @@ func TestResumeCoversEveryStatus(t *testing.T) { t.Errorf("status %q is written by the pipeline but is in no resume list — "+ "a restart while a session holds it would strand the session with nothing "+ "driving it and no budget to fail it. Add it to fsResumePreGate (before the "+ - "admin gate), fsResumePostGate (after it), or fsResumeTerminal (nothing owed).", + "admin gate), fsResumePostGate (after it), fsResumeDeployment, or fsResumeTerminal (nothing owed).", status) } } @@ -107,6 +110,7 @@ func TestResumeListsHaveNoDeadStatuses(t *testing.T) { }{ {"fsResumePreGate", fsResumePreGate}, {"fsResumePostGate", fsResumePostGate}, + {"fsResumeDeployment", fsResumeDeployment}, {"fsResumeTerminal", fsResumeTerminal}, } { for _, status := range list.entries { @@ -129,6 +133,7 @@ func TestResumeListsAreDisjoint(t *testing.T) { }{ {"fsResumePreGate", fsResumePreGate}, {"fsResumePostGate", fsResumePostGate}, + {"fsResumeDeployment", fsResumeDeployment}, {"fsResumeTerminal", fsResumeTerminal}, } { for _, status := range list.entries { @@ -156,3 +161,15 @@ func TestCustomDomainStatusesAreResumable(t *testing.T) { } } } + +func TestDeploymentStatusesUseExactRecovery(t *testing.T) { + want := []string{"deploy_dids", "deploy_mediator", "deploy_vta", "deploy_vtc"} + if len(fsResumeDeployment) != len(want) { + t.Fatalf("fsResumeDeployment = %v, want %v", fsResumeDeployment, want) + } + for i := range want { + if fsResumeDeployment[i] != want[i] { + t.Fatalf("fsResumeDeployment = %v, want %v", fsResumeDeployment, want) + } + } +} From ea0e850ee84561e5edd91691c43afe2a25b6a265 Mon Sep 17 00:00:00 2001 From: vthwang Date: Sat, 17 Oct 2026 12:38:33 -0700 Subject: [PATCH 2/2] feat: recover sessions with selected component images Let failed deployments resume with a mode-appropriate image set. - reconcile existing full-stack components before continuing setup - support VTA-only deployment recovery and document the API contract Signed-off-by: vthwang --- internal/apidocs/openapi.yaml | 72 +++++++----- internal/handler/setup_recovery.go | 150 ++++++++++++++++-------- internal/handler/setup_recovery_test.go | 67 +++++++++-- internal/model/setup_session.go | 18 ++- internal/model/setup_session_test.go | 7 +- internal/setup/orchestrator.go | 73 ++++++------ internal/setup/orchestrator_recovery.go | 75 +++++++++--- 7 files changed, 314 insertions(+), 148 deletions(-) diff --git a/internal/apidocs/openapi.yaml b/internal/apidocs/openapi.yaml index 2f4c3d5..c550bb0 100644 --- a/internal/apidocs/openapi.yaml +++ b/internal/apidocs/openapi.yaml @@ -905,29 +905,42 @@ components: RecoverSetupRequest: type: object - required: [component, image] + required: [components] properties: - component: - type: string - enum: [vta, mediator, dids, vtc] - description: Must match the component implied by the failed deploy stage. - image: - type: string - description: Full image URL returned by the component's setup images endpoint. - example: "ghcr.io/ic3software/vtc:0.3.1-917881c" + components: + type: array + minItems: 1 + description: | + The complete image set for the session mode. `full_stack` requires + exactly vta, mediator, dids, and vtc; `vta_only` requires exactly + vta. Each image must be returned by that component's setup images + endpoint. + items: + type: object + required: [component, image] + properties: + component: + type: string + enum: [vta, mediator, dids, vtc] + image: + type: string + example: "ghcr.io/ic3software/vtc:0.3.1-917881c" RecoverSetupResponse: type: object - required: [status, component, image] + required: [status, components] properties: status: type: string enum: [deploy_dids, deploy_mediator, deploy_vta, deploy_vtc] - component: - type: string - enum: [vta, mediator, dids, vtc] - image: - type: string + components: + type: array + items: + type: object + required: [component, image] + properties: + component: { type: string, enum: [vta, mediator, dids, vtc] } + image: { type: string } ModeAvailability: type: object @@ -3099,11 +3112,12 @@ paths: post: summary: Recover a failed deployment (admin) description: | - Admin-cookie twin of `POST /setup/{id}/recover`. The session must be a - full-stack session whose `failed_stage` is `deploy_dids`, - `deploy_mediator`, `deploy_vta`, or `deploy_vtc`. The replacement image - is persisted, the existing Deployment is reconciled to it, and setup - continues from the next step without repeating completed setup jobs. + Admin-cookie twin of `POST /setup/{id}/recover`. A full-stack request + selects all four component images; a VTA-only request selects only its + VTA image. Existing Deployments are reconciled to the selected versions + through the failed stage, and later components use the persisted + selections when setup continues. Completed setup jobs and PVC data are + retained. tags: [Admin] security: [{ CookieAuthAdmin: [] }] parameters: @@ -3117,7 +3131,7 @@ paths: "202": description: Recovery accepted and started content: { application/json: { schema: { $ref: "#/components/schemas/RecoverSetupResponse" } } } - "400": { description: Component does not match the failed stage or image is unavailable, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "400": { description: Component set does not match the session mode or an image is unavailable, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } "401": { description: Missing or invalid token, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } "403": { description: Insufficient role, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } "404": { description: Session not found, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } @@ -5275,13 +5289,13 @@ paths: post: summary: Recover the caller's failed deployment description: | - Changes the image for the component whose full-stack deployment failed, - retries only that deployment, and continues from the next setup step. - Supported mappings are `deploy_dids` → dids, `deploy_mediator` → - mediator, `deploy_vta` → vta, and `deploy_vtc` → vtc. Completed setup - jobs and PVC data are retained. The response returns immediately; poll - `GET /setup/{id}` for progress. A service restart also resumes from the - persisted deploy stage. + Selects the complete image set for a failed session, reconciles existing + Deployments through the failed stage, and continues from the next setup + step. `full_stack` accepts vta, mediator, dids, and vtc; `vta_only` + accepts only vta and is recoverable only from `deploy_vta`. Completed + setup jobs and PVC data are retained. The response returns immediately; + poll `GET /setup/{id}` for progress. A service restart also resumes from + the persisted deploy stage. tags: [User] security: [{ CookieAuthUser: [] }] parameters: @@ -5295,7 +5309,7 @@ paths: "202": description: Recovery accepted and started content: { application/json: { schema: { $ref: "#/components/schemas/RecoverSetupResponse" } } } - "400": { description: Component does not match the failed stage or image is unavailable, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } + "400": { description: Component set does not match the session mode or an image is unavailable, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } "401": { description: Missing or invalid token, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } "404": { description: Session not found or owned by another user, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } "409": { description: Session is not at a recoverable deploy failure or has an active upgrade, content: { application/json: { schema: { $ref: "#/components/schemas/Error" } } } } diff --git a/internal/handler/setup_recovery.go b/internal/handler/setup_recovery.go index 22b483a..e42a064 100644 --- a/internal/handler/setup_recovery.go +++ b/internal/handler/setup_recovery.go @@ -2,8 +2,10 @@ package handler import ( "errors" + "fmt" "net/http" "slices" + "strings" "time" "github.com/gin-gonic/gin" @@ -15,31 +17,72 @@ import ( ) type recoverSetupRequest struct { - Component string `json:"component" binding:"required"` - Image string `json:"image" binding:"required"` + Components []componentImage `json:"components" binding:"required,min=1,dive"` } type deploymentRecovery struct { - Stage string - Component string - ImageColumn string + Stage string } -func recoveryForStage(stage string) (deploymentRecovery, bool) { - component := "" - switch stage { - case "deploy_dids": - component = "dids" - case "deploy_mediator": - component = "mediator" - case "deploy_vta": - component = "vta" - case "deploy_vtc": - component = "vtc" - default: +func recoveryForStage(mode, stage string) (deploymentRecovery, bool) { + if mode == model.ModeVtaOnly { + if stage != "deploy_vta" { + return deploymentRecovery{}, false + } + return deploymentRecovery{Stage: stage}, true + } + if mode != model.ModeFullStack || !slices.Contains([]string{ + "deploy_dids", "deploy_mediator", "deploy_vta", "deploy_vtc", + }, stage) { return deploymentRecovery{}, false } - return deploymentRecovery{Stage: stage, Component: component, ImageColumn: model.UpgradeImageColumn(component)}, true + return deploymentRecovery{Stage: stage}, true +} + +func recoveryComponentsForMode(mode string) []string { + if mode == model.ModeVtaOnly { + return []string{"vta"} + } + if mode == model.ModeFullStack { + return []string{"vta", "mediator", "dids", "vtc"} + } + return nil +} + +func validateRecoveryComponents(mode string, components []componentImage) error { + required := recoveryComponentsForMode(mode) + if len(required) == 0 { + return fmt.Errorf("mode %s cannot be recovered", mode) + } + seen := make(map[string]bool, len(components)) + for _, ci := range components { + if seen[ci.Component] { + return fmt.Errorf("duplicate component %s", ci.Component) + } + seen[ci.Component] = true + } + var missing, extra []string + for _, component := range required { + if !seen[component] { + missing = append(missing, component) + } + } + for component := range seen { + if !slices.Contains(required, component) { + extra = append(extra, component) + } + } + if len(missing) > 0 || len(extra) > 0 { + parts := make([]string, 0, 2) + if len(missing) > 0 { + parts = append(parts, "missing "+strings.Join(missing, ", ")) + } + if len(extra) > 0 { + parts = append(parts, "unsupported "+strings.Join(extra, ", ")) + } + return fmt.Errorf("mode %s requires exactly its component images: %s", mode, strings.Join(parts, "; ")) + } + return nil } func (h *SetupHandler) recoveryImageClient(component string) *ghcr.Client { @@ -85,42 +128,50 @@ func (h *SetupHandler) recoverSession(c *gin.Context, session *model.SetupSessio return } - recovery, ok := recoveryForStage(session.FailureStage()) - if session.Mode != model.ModeFullStack || !ok { - c.JSON(http.StatusConflict, gin.H{"error": "only full-stack sessions failed at a deploy stage can be recovered"}) + recovery, ok := recoveryForStage(session.Mode, session.FailureStage()) + if !ok { + c.JSON(http.StatusConflict, gin.H{"error": "only sessions failed at a recoverable deploy stage can be recovered"}) return } - if req.Component != recovery.Component { - c.JSON(http.StatusBadRequest, gin.H{"error": "failed stage " + recovery.Stage + " requires component " + recovery.Component}) + if err := validateRecoveryComponents(session.Mode, req.Components); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } - client := h.recoveryImageClient(recovery.Component) - if client == nil { - c.JSON(http.StatusServiceUnavailable, gin.H{"error": "image source not configured for " + recovery.Component}) - return - } - tags, err := client.ListTags(c.Request.Context()) - if err != nil { - c.JSON(http.StatusBadGateway, gin.H{"error": "failed to fetch images: " + err.Error()}) - return - } - if !slices.ContainsFunc(tags, func(tag ghcr.ImageTag) bool { return tag.Image == req.Image }) { - c.JSON(http.StatusBadRequest, gin.H{"error": "image is not available for " + recovery.Component}) - return + for _, ci := range req.Components { + // Keeping a component on its persisted image is valid even when that old + // tag has since fallen out of the registry listing. Only replacements + // need to prove that the registry currently serves them. + if session.ComponentImage(ci.Component) == ci.Image { + continue + } + client := h.recoveryImageClient(ci.Component) + if client == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "image source not configured for " + ci.Component}) + return + } + tags, err := client.ListTags(c.Request.Context()) + if err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": "failed to fetch " + ci.Component + " images: " + err.Error()}) + return + } + if !slices.ContainsFunc(tags, func(tag ghcr.ImageTag) bool { return tag.Image == ci.Image }) { + c.JSON(http.StatusBadRequest, gin.H{"error": "image is not available for " + ci.Component}) + return + } } conflictMessage := "session is no longer recoverable" - err = h.db.WithContext(c.Request.Context()).Transaction(func(tx *gorm.DB) error { + err := h.db.WithContext(c.Request.Context()).Transaction(func(tx *gorm.DB) error { var locked model.SetupSession if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&locked, session.ID).Error; err != nil { return err } - lockedRecovery, ok := recoveryForStage(locked.FailureStage()) - if locked.Mode != model.ModeFullStack || !ok { + lockedRecovery, ok := recoveryForStage(locked.Mode, locked.FailureStage()) + if !ok { return errRecoveryConflict } - if lockedRecovery.Component != req.Component { + if locked.Mode != session.Mode || lockedRecovery.Stage != recovery.Stage { conflictMessage = "session failure stage changed; reload before recovering" return errRecoveryConflict } @@ -137,13 +188,16 @@ func (h *SetupHandler) recoverSession(c *gin.Context, session *model.SetupSessio return errRecoveryConflict } - result := tx.Model(&locked).Where("status = ?", "failed").Updates(map[string]any{ - lockedRecovery.ImageColumn: req.Image, - "status": lockedRecovery.Stage, - "error_msg": "", - "failed_stage": "", - "updated_at": time.Now(), - }) + updates := map[string]any{ + "status": lockedRecovery.Stage, + "error_msg": "", + "failed_stage": "", + "updated_at": time.Now(), + } + for _, ci := range req.Components { + updates[model.UpgradeImageColumn(ci.Component)] = ci.Image + } + result := tx.Model(&locked).Where("status = ?", "failed").Updates(updates) if result.Error != nil { return result.Error } @@ -164,6 +218,6 @@ func (h *SetupHandler) recoverSession(c *gin.Context, session *model.SetupSessio h.orch.RecoverDeployment(session.ID) c.JSON(http.StatusAccepted, gin.H{ - "status": recovery.Stage, "component": recovery.Component, "image": req.Image, + "status": recovery.Stage, "components": req.Components, }) } diff --git a/internal/handler/setup_recovery_test.go b/internal/handler/setup_recovery_test.go index d7461ec..4d7daf1 100644 --- a/internal/handler/setup_recovery_test.go +++ b/internal/handler/setup_recovery_test.go @@ -1,25 +1,31 @@ package handler -import "testing" +import ( + "strings" + "testing" + + "github.com/ic3software/vtafarm-api/internal/model" +) func TestRecoveryForStage(t *testing.T) { tests := []struct { - stage, component, column string + mode, stage string }{ - {"deploy_dids", "dids", "dids_image"}, - {"deploy_mediator", "mediator", "mediator_image"}, - {"deploy_vta", "vta", "vta_image"}, - {"deploy_vtc", "vtc", "vtc_image"}, + {model.ModeFullStack, "deploy_dids"}, + {model.ModeFullStack, "deploy_mediator"}, + {model.ModeFullStack, "deploy_vta"}, + {model.ModeFullStack, "deploy_vtc"}, + {model.ModeVtaOnly, "deploy_vta"}, } for _, test := range tests { - t.Run(test.stage, func(t *testing.T) { - got, ok := recoveryForStage(test.stage) + t.Run(test.mode+"/"+test.stage, func(t *testing.T) { + got, ok := recoveryForStage(test.mode, test.stage) if !ok { - t.Fatalf("recoveryForStage(%q) was not recoverable", test.stage) + t.Fatalf("recoveryForStage(%q, %q) was not recoverable", test.mode, test.stage) } - if got.Stage != test.stage || got.Component != test.component || got.ImageColumn != test.column { - t.Fatalf("recoveryForStage(%q) = %#v", test.stage, got) + if got.Stage != test.stage { + t.Fatalf("recoveryForStage(%q, %q) = %#v", test.mode, test.stage, got) } }) } @@ -27,8 +33,45 @@ func TestRecoveryForStage(t *testing.T) { func TestRecoveryForStageRejectsNonDeploymentStages(t *testing.T) { for _, stage := range []string{"", "failed", "step_vtc_setup", "running"} { - if _, ok := recoveryForStage(stage); ok { + if _, ok := recoveryForStage(model.ModeFullStack, stage); ok { t.Errorf("recoveryForStage(%q) unexpectedly recoverable", stage) } } + if _, ok := recoveryForStage(model.ModeVtaOnly, "deploy_vtc"); ok { + t.Error("vta_only deploy_vtc unexpectedly recoverable") + } +} + +func TestValidateRecoveryComponents(t *testing.T) { + fullStack := []componentImage{ + {Component: "vta", Image: "vta-image"}, + {Component: "mediator", Image: "mediator-image"}, + {Component: "dids", Image: "dids-image"}, + {Component: "vtc", Image: "vtc-image"}, + } + if err := validateRecoveryComponents(model.ModeFullStack, fullStack); err != nil { + t.Fatalf("full_stack components rejected: %v", err) + } + if err := validateRecoveryComponents(model.ModeVtaOnly, []componentImage{{Component: "vta", Image: "vta-image"}}); err != nil { + t.Fatalf("vta_only VTA component rejected: %v", err) + } + + tests := []struct { + name string + mode string + components []componentImage + want string + }{ + {"full stack missing VTC", model.ModeFullStack, fullStack[:3], "missing vtc"}, + {"vta only extra VTC", model.ModeVtaOnly, []componentImage{{Component: "vta"}, {Component: "vtc"}}, "unsupported vtc"}, + {"duplicate", model.ModeFullStack, append(fullStack, componentImage{Component: "vta"}), "duplicate component vta"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + err := validateRecoveryComponents(test.mode, test.components) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("validateRecoveryComponents() error = %v, want %q", err, test.want) + } + }) + } } diff --git a/internal/model/setup_session.go b/internal/model/setup_session.go index f99ebf1..4a15fb7 100644 --- a/internal/model/setup_session.go +++ b/internal/model/setup_session.go @@ -197,16 +197,26 @@ func (s *SetupSession) FailureStage() string { return s.FailedStage } if s.Mode == ModeVtaOnly { + for _, prefix := range []string{ + "failed to load resource defaults", + "failed to identify pool session", + "pool resource profile is missing", + "failed to create VTA deployment", + "failed to start prebuilt VTA deployment", + "failed to create VTA service", + "failed to create VTA ingress", + "VTA deployment did not become ready", + } { + if strings.HasPrefix(s.ErrorMsg, prefix) { + return "deploy_vta" + } + } for _, prefix := range []string{ "failed to reach DID hosting control API", "failed to add VTA DID to hosting ACL", "failed to create provision job", "provision job watch error", "provision job failed", - "failed to create VTA deployment", - "failed to create VTA service", - "failed to create VTA ingress", - "VTA deployment did not become ready", } { if strings.HasPrefix(s.ErrorMsg, prefix) { return "provisioning" diff --git a/internal/model/setup_session_test.go b/internal/model/setup_session_test.go index ce5ee34..668e3f3 100644 --- a/internal/model/setup_session_test.go +++ b/internal/model/setup_session_test.go @@ -19,8 +19,13 @@ func TestFailureStage(t *testing.T) { want: "step_dids_provision", }, { - name: "legacy vta provisioning error", + name: "legacy vta deployment error", session: SetupSession{Status: "failed", Mode: ModeVtaOnly, ErrorMsg: "VTA deployment did not become ready: timeout"}, + want: "deploy_vta", + }, + { + name: "legacy vta provision job error", + session: SetupSession{Status: "failed", Mode: ModeVtaOnly, ErrorMsg: "provision job failed: invalid config"}, want: "provisioning", }, { diff --git a/internal/setup/orchestrator.go b/internal/setup/orchestrator.go index 98c4914..85430d4 100644 --- a/internal/setup/orchestrator.go +++ b/internal/setup/orchestrator.go @@ -470,75 +470,70 @@ func (o *Orchestrator) runProvision(ctx context.Context, sessionID uint, adminDi log.Printf("[orchestrator] session %d: provision job completed", sessionID) log.Printf("[orchestrator] session %d: starting VTA deployment", sessionID) + pooled, err := o.deployVtaOnly(ctx, ns, &session) + if err != nil { + if ctx.Err() != nil { + return + } + o.markFailedAt(sessionID, "deploy_vta", err.Error()) + return + } + + o.finishVtaOnlyDeployment(ctx, &session, pooled) +} +func (o *Orchestrator) deployVtaOnly(ctx context.Context, ns string, session *model.SetupSession) (bool, error) { profiles, err := resourceprofile.Load(ctx, o.db) if err != nil { - o.markFailed(sessionID, "failed to load resource defaults: "+err.Error()) - return + return false, fmt.Errorf("failed to load resource defaults: %w", err) } pooled, err := o.isPoolSession(ctx, session.ID) if err != nil { - o.markFailed(sessionID, "failed to identify pool session") - return + return false, fmt.Errorf("failed to identify pool session: %w", err) } if pooled { var saved model.WorkloadResource if err := o.db.Where("setup_session_id = ? AND component = 'vta'", session.ID).First(&saved).Error; err != nil { - o.markFailed(sessionID, "pool resource profile is missing") - return + return false, fmt.Errorf("pool resource profile is missing: %w", err) } p := profiles[k8s.ComponentVTA] p.MemoryRequest, p.MemoryLimit = saved.MemoryRequest, saved.MemoryLimit profiles[k8s.ComponentVTA] = p } - if err := o.k8s.CreateVtaDeployment(ctx, ns, sessionID, session.VtaImage, profiles.Requirements(k8s.ComponentVTA)); err != nil { - if ctx.Err() != nil { - return - } - o.markFailed(sessionID, "failed to create VTA deployment: "+err.Error()) - return + if err := o.k8s.CreateVtaDeployment(ctx, ns, session.ID, session.VtaImage, profiles.Requirements(k8s.ComponentVTA)); err != nil { + return pooled, fmt.Errorf("failed to create VTA deployment: %w", err) } if pooled { - if err := o.k8s.ScaleComponentDeployment(ctx, ns, k8s.VtaDeploymentName(sessionID), 1); err != nil { - o.markFailed(sessionID, "failed to start prebuilt VTA deployment: "+err.Error()) - return + if err := o.k8s.ScaleComponentDeployment(ctx, ns, k8s.VtaDeploymentName(session.ID), 1); err != nil { + return pooled, fmt.Errorf("failed to start prebuilt VTA deployment: %w", err) } } - - if err := o.k8s.CreateVtaService(ctx, ns, sessionID); err != nil { - if ctx.Err() != nil { - return - } - o.markFailed(sessionID, "failed to create VTA service: "+err.Error()) - return + if err := o.k8s.CreateVtaService(ctx, ns, session.ID); err != nil { + return pooled, fmt.Errorf("failed to create VTA service: %w", err) } - - if err := o.k8s.CreateVtaIngress(ctx, ns, sessionID, session.FQDN()); err != nil { - if ctx.Err() != nil { - return - } - o.markFailed(sessionID, "failed to create VTA ingress: "+err.Error()) - return + if err := o.k8s.CreateVtaIngress(ctx, ns, session.ID, session.FQDN()); err != nil { + return pooled, fmt.Errorf("failed to create VTA ingress: %w", err) } // A Deployment exists before the VTA has finished initializing. Its // /health readiness probe is the serving boundary: do not advertise this // session as running until Kubernetes confirms at least one ready replica. if err := o.k8s.WaitForComponentDeploymentReady( - ctx, ns, k8s.VtaDeploymentName(sessionID), 2*time.Minute, + ctx, ns, k8s.VtaDeploymentName(session.ID), 2*time.Minute, ); err != nil { - if ctx.Err() != nil { - return - } - o.markFailed(sessionID, "VTA deployment did not become ready: "+err.Error()) - return + return pooled, fmt.Errorf("VTA deployment did not become ready: %w", err) } + return pooled, nil +} - o.db.Model(&model.SetupSession{}).Where("id = ?", sessionID).Updates(map[string]any{ - "status": "running", - "updated_at": time.Now(), +func (o *Orchestrator) finishVtaOnlyDeployment(ctx context.Context, session *model.SetupSession, pooled bool) { + o.db.Model(&model.SetupSession{}).Where("id = ?", session.ID).Updates(map[string]any{ + "status": "running", + "error_msg": "", + "failed_stage": "", + "updated_at": time.Now(), }) - log.Printf("[orchestrator] session %d: VTA running at %s", sessionID, session.FQDN()) + log.Printf("[orchestrator] session %d: VTA running at %s", session.ID, session.FQDN()) if pooled { if current, err := connection.Current(o.db, session.ID); err == nil && current != nil { _ = connection.MarkReleased(ctx, o.db, current.ID) diff --git a/internal/setup/orchestrator_recovery.go b/internal/setup/orchestrator_recovery.go index 834d670..c720a68 100644 --- a/internal/setup/orchestrator_recovery.go +++ b/internal/setup/orchestrator_recovery.go @@ -26,9 +26,6 @@ func (o *Orchestrator) recoverDeployment(ctx context.Context, sessionID uint) { log.Printf("[orchestrator] deployment recovery %d: load failed: %v", sessionID, err) return } - if session.Mode != model.ModeFullStack { - return - } s := &session ns := o.k8s.UserNamespace(fmt.Sprintf("%d", s.UserID)) @@ -42,12 +39,31 @@ func (o *Orchestrator) recoverDeployment(ctx context.Context, sessionID uint) { o.markFailedAt(sessionID, stage, prefix+": "+err.Error()) return true } + if s.Mode == model.ModeVtaOnly { + if s.Status != "deploy_vta" { + return + } + if fail("deploy_vta", "failed to reconcile vta deployment image", o.k8s.RecoverDeploymentImage( + ctx, ns, k8s.VtaDeploymentName(s.ID), s.VtaImage, + )) { + return + } + pooled, err := o.deployVtaOnly(ctx, ns, s) + if fail("deploy_vta", "failed to deploy vta", err) { + return + } + o.finishVtaOnlyDeployment(ctx, s, pooled) + return + } + if s.Mode != model.ModeFullStack { + return + } + if fail(s.Status, "failed to reconcile recovery images", o.reconcileFullStackRecoveryImages(ctx, ns, s)) { + return + } switch s.Status { case "deploy_dids": - if fail("deploy_dids", "failed to reconcile dids deployment image", o.k8s.RecoverDeploymentImage(ctx, ns, k8s.FSDidsName(s.ID), s.DidsImage)) { - return - } if fail("deploy_dids", "failed to deploy dids daemon", o.fsDeployDids(ctx, ns, s)) { return } @@ -58,27 +74,18 @@ func (o *Orchestrator) recoverDeployment(ctx context.Context, sessionID uint) { o.continueRecoveredPreGate(ctx, ns, s) case "deploy_mediator": - if fail("deploy_mediator", "failed to reconcile mediator deployment image", o.k8s.RecoverDeploymentImage(ctx, ns, k8s.FSMediatorName(s.ID), s.MediatorImage)) { - return - } if fail("deploy_mediator", "failed to deploy mediator", o.fsDeployMediator(ctx, ns, s)) { return } o.continueRecoveredPreGate(ctx, ns, s) case "deploy_vta": - if fail("deploy_vta", "failed to reconcile vta deployment image", o.k8s.RecoverDeploymentImage(ctx, ns, k8s.FSVtaName(s.ID), s.VtaImage)) { - return - } if fail("deploy_vta", "failed to deploy vta", o.fsDeployVta(ctx, ns, s)) { return } o.continueRecoveredPostGate(ctx, ns, s, fail) case "deploy_vtc": - if fail("deploy_vtc", "failed to reconcile vtc deployment image", o.k8s.RecoverDeploymentImage(ctx, ns, k8s.FSVtcName(s.ID), s.VtcImage)) { - return - } if fail("deploy_vtc", "failed to deploy vtc", o.fsDeployVtc(ctx, ns, s)) { return } @@ -86,6 +93,44 @@ func (o *Orchestrator) recoverDeployment(ctx context.Context, sessionID uint) { } } +func (o *Orchestrator) reconcileFullStackRecoveryImages(ctx context.Context, ns string, s *model.SetupSession) error { + components := []struct { + stage string + name string + image string + }{ + {stage: "deploy_dids", name: k8s.FSDidsName(s.ID), image: s.DidsImage}, + {stage: "deploy_mediator", name: k8s.FSMediatorName(s.ID), image: s.MediatorImage}, + {stage: "deploy_vta", name: k8s.FSVtaName(s.ID), image: s.VtaImage}, + {stage: "deploy_vtc", name: k8s.FSVtcName(s.ID), image: s.VtcImage}, + } + failedIndex := -1 + for i, component := range components { + if component.stage == s.Status { + failedIndex = i + break + } + } + if failedIndex < 0 { + return fmt.Errorf("unsupported recovery stage %s", s.Status) + } + for i := 0; i <= failedIndex; i++ { + component := components[i] + if err := o.k8s.RecoverDeploymentImage(ctx, ns, component.name, component.image); err != nil { + return fmt.Errorf("reconcile %s: %w", component.name, err) + } + // The failed component's normal deploy helper below creates any missing + // resources and performs its readiness wait. Components that were already + // running must finish their replacement rollout before setup continues. + if i < failedIndex { + if err := o.k8s.WaitForComponentDeploymentReady(ctx, ns, component.name, 2*time.Minute); err != nil { + return fmt.Errorf("wait for %s: %w", component.name, err) + } + } + } + return nil +} + func (o *Orchestrator) continueRecoveredPreGate(ctx context.Context, ns string, s *model.SetupSession) { o.fsSetStatus(s.ID, "step_vta_register_dids") if err := o.fsStepVtaRegisterDids(ctx, ns, s); err != nil {