Merge pull request #509 from chaitin/fix/backend-lifecycle-vm-failed-20260416

fix(backend): 统一 VM 回调到生命周期

Author: yokowu

backend/biz/host/handler/v1/internal.go

@@ -34,8 +34,7 @@ type InternalHostHandler struct {
 	teamRepo       domain.TeamHostRepo
 	redis          *redis.Client
 	cache          *cache.Cache
-	hook           domain.InternalHook // 可选，由内部项目通过 WithInternalHook 注入
-	taskLifecycle  *lifecycle.Manager[uuid.UUID, consts.TaskStatus, lifecycle.TaskMetadata]
+	vmLifecycle    *lifecycle.Manager[string, lifecycle.VMState, lifecycle.VMMetadata]
 	hostUsecase    domain.HostUsecase
 	taskConns      *ws.TaskConn
 	projectUsecase domain.ProjectUsecase
@@ -51,18 +50,13 @@ func NewInternalHostHandler(i *do.Injector) (*InternalHostHandler, error) {
 		teamRepo:       do.MustInvoke[domain.TeamHostRepo](i),
 		redis:          do.MustInvoke[*redis.Client](i),
 		cache:          cache.New(15*time.Minute, 10*time.Minute),
-		taskLifecycle:  do.MustInvoke[*lifecycle.Manager[uuid.UUID, consts.TaskStatus, lifecycle.TaskMetadata]](i),
+		vmLifecycle:    do.MustInvoke[*lifecycle.Manager[string, lifecycle.VMState, lifecycle.VMMetadata]](i),
 		hostUsecase:    do.MustInvoke[domain.HostUsecase](i),
 		taskConns:      do.MustInvoke[*ws.TaskConn](i),
 		projectUsecase: do.MustInvoke[domain.ProjectUsecase](i),
 		tokenProvider:  do.MustInvoke[*gituc.TokenProvider](i),
 	}
 
-	// 可选注入 InternalHook
-	if hook, err := do.Invoke[domain.InternalHook](i); err == nil {
-		h.hook = hook
-	}
-
 	g := w.Group("/internal")
 	g.POST("/check-token", web.BindHandler(h.CheckToken))
 	g.POST("/host-info", web.BindHandler(h.ReportHostInfo))
@@ -350,16 +344,13 @@ func (h *InternalHostHandler) VmReady(c *web.Context, req taskflow.VirtualMachin
 			continue
 		}
 
-		if t.Kind == consts.TaskTypeReview && t.SubType == consts.TaskSubTypePrReview {
-		} else {
-			if err := h.taskLifecycle.Transition(c.Request().Context(), t.ID, consts.TaskStatusProcessing, lifecycle.TaskMetadata{
-				TaskID: t.ID,
-				UserID: t.UserID,
-			}); err != nil {
-				h.logger.With("task", t, "error", err).ErrorContext(c.Request().Context(), "failed to transition task to processing")
-			}
+		if err := h.vmLifecycle.Transition(c.Request().Context(), vm.ID, lifecycle.VMStateRunning, lifecycle.VMMetadata{
+			VMID:   vm.ID,
+			TaskID: &t.ID,
+			UserID: t.UserID,
+		}); err != nil {
+			h.logger.With("task", t, "error", err).ErrorContext(c.Request().Context(), "failed to transition vm to running")
 		}
-
 	}
 
 	return c.Success(nil)
@@ -384,15 +375,19 @@ func (h *InternalHostHandler) VmConditions(c *web.Context, req taskflow.VirtualM
 		return err
 	}
 
-	// 条件失败时通过 hook 通知内部项目（任务状态转换等）
-	if h.hook != nil {
+	for _, task := range vm.Edges.Tasks {
 		for _, cond := range req.Conditions {
-			if cond.Type == string(etypes.ConditionTypeFailed) {
-				if err := h.hook.OnVmConditionFailed(c.Request().Context(), vm.ID); err != nil {
-					h.logger.With("error", err).ErrorContext(c.Request().Context(), "hook OnVmConditionFailed failed")
-				}
-				break
+			if cond.Type != string(etypes.ConditionTypeFailed) {
+				continue
+			}
+			if err := h.vmLifecycle.Transition(c.Request().Context(), vm.ID, lifecycle.VMStateFailed, lifecycle.VMMetadata{
+				VMID:   vm.ID,
+				TaskID: &task.ID,
+				UserID: task.UserID,
+			}); err != nil {
+				h.logger.With("task", task, "error", err).ErrorContext(c.Request().Context(), "failed to transition vm to failed")
 			}
+			break
 		}
 	}
 

backend/go.mod

@@ -6,6 +6,7 @@ require (
 	entgo.io/ent v0.14.5
 	github.com/GoYoko/web v1.6.0
 	github.com/ackcoder/go-cap v1.1.3
+	github.com/alicebob/miniredis/v2 v2.35.0
 	github.com/aliyun/alibabacloud-nls-go-sdk v1.1.1
 	github.com/coder/websocket v1.8.14
 	github.com/gogo/protobuf v1.3.2
@@ -83,6 +84,7 @@ require (
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	github.com/valyala/fasttemplate v1.2.2 // indirect
+	github.com/yuin/gopher-lua v1.1.1 // indirect
 	github.com/zclconf/go-cty v1.14.4 // indirect
 	github.com/zclconf/go-cty-yaml v1.1.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect

backend/go.sum

@@ -16,6 +16,8 @@ github.com/ackcoder/go-cap v1.1.3 h1:rHIZEmyOM/KlXJQxGoy3UHpzpeUhw+V8qa/OoEaJR7A
 github.com/ackcoder/go-cap v1.1.3/go.mod h1:NRffl9i4+VPdgAgMT4G62cXakEyCyZtXg9ZMX3/RsDA=
 github.com/agext/levenshtein v1.2.3 h1:YB2fHEn0UJagG8T1rrWknE3ZQzWM06O8AMAatNn7lmo=
 github.com/agext/levenshtein v1.2.3/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki2W0IB5558=
+github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI=
+github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
 github.com/aliyun/alibaba-cloud-sdk-go v1.61.1376 h1:lExo7heZgdFn5AbaNJEllbA0KSJ/Z8T7MphvMREJOOo=
 github.com/aliyun/alibaba-cloud-sdk-go v1.61.1376/go.mod h1:9CMdKNL3ynIGPpfTcdwTvIm8SGuAZYYC4jFVSSvE1YQ=
 github.com/aliyun/alibabacloud-nls-go-sdk v1.1.1 h1:LjItoNZuu5xHlsByFo+kr3nGa4LRIESCGWhfurayxBg=
@@ -243,6 +245,8 @@ github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQ
 github.com/valyala/fasttemplate v1.2.2/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
+github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
 github.com/zclconf/go-cty v1.14.4 h1:uXXczd9QDGsgu0i/QFR/hzI5NYCHLf6NQw/atrbnhq8=
 github.com/zclconf/go-cty v1.14.4/go.mod h1:VvMs5i0vgZdhYawQNq5kePSpLAoz8u1xvZgrPIxfnZE=
 github.com/zclconf/go-cty-yaml v1.1.0 h1:nP+jp0qPHv2IhUVqmQSzjvqAWcObN0KBkUl2rWBdig0=

backend/pkg/lifecycle/vmtaskhook.go

@@ -0,0 +1,50 @@
+package lifecycle
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/google/uuid"
+	"github.com/samber/do"
+
+	"github.com/chaitin/MonkeyCode/backend/consts"
+)
+
+// VMTaskHook 将 VM 生命周期状态同步到任务生命周期。
+type VMTaskHook struct {
+	taskLifecycle *Manager[uuid.UUID, consts.TaskStatus, TaskMetadata]
+	logger        *slog.Logger
+}
+
+func NewVMTaskHook(i *do.Injector) *VMTaskHook {
+	return &VMTaskHook{
+		taskLifecycle: do.MustInvoke[*Manager[uuid.UUID, consts.TaskStatus, TaskMetadata]](i),
+		logger:        do.MustInvoke[*slog.Logger](i).With("hook", "vm-task-hook"),
+	}
+}
+
+func (h *VMTaskHook) Name() string  { return "vm-task-hook" }
+func (h *VMTaskHook) Priority() int { return 100 }
+func (h *VMTaskHook) Async() bool   { return false }
+
+func (h *VMTaskHook) OnStateChange(ctx context.Context, _ string, _ VMState, to VMState, metadata VMMetadata) error {
+	if metadata.TaskID == nil {
+		return nil
+	}
+
+	var target consts.TaskStatus
+	switch to {
+	case VMStateRunning:
+		target = consts.TaskStatusProcessing
+	case VMStateFailed:
+		target = consts.TaskStatusError
+	default:
+		return nil
+	}
+
+	h.logger.InfoContext(ctx, "sync task lifecycle from vm lifecycle", "task_id", metadata.TaskID, "state", target)
+	return h.taskLifecycle.Transition(ctx, *metadata.TaskID, target, TaskMetadata{
+		TaskID: *metadata.TaskID,
+		UserID: metadata.UserID,
+	})
+}

backend/pkg/lifecycle/vmtaskhook_test.go

@@ -0,0 +1,59 @@
+package lifecycle
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/google/uuid"
+	"github.com/redis/go-redis/v9"
+
+	"github.com/chaitin/MonkeyCode/backend/consts"
+)
+
+func TestVMTaskHook_OnStateChange_FailedTransitionsTaskToError(t *testing.T) {
+	mr, err := miniredis.Run()
+	if err != nil {
+		t.Fatalf("miniredis.Run() error = %v", err)
+	}
+	defer mr.Close()
+
+	rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	defer rdb.Close()
+
+	taskLifecycle := NewManager[uuid.UUID, consts.TaskStatus, TaskMetadata](
+		rdb,
+		WithLogger[uuid.UUID, consts.TaskStatus, TaskMetadata](slog.New(slog.NewTextHandler(io.Discard, nil))),
+		WithTransitions[uuid.UUID, consts.TaskStatus, TaskMetadata](TaskTransitions()),
+	)
+
+	taskID := uuid.New()
+	userID := uuid.New()
+	meta := TaskMetadata{TaskID: taskID, UserID: userID}
+	if err := taskLifecycle.Transition(context.Background(), taskID, consts.TaskStatusPending, meta); err != nil {
+		t.Fatalf("taskLifecycle.Transition(pending) error = %v", err)
+	}
+
+	hook := &VMTaskHook{
+		taskLifecycle: taskLifecycle,
+		logger:        slog.New(slog.NewTextHandler(io.Discard, nil)),
+	}
+
+	if err := hook.OnStateChange(context.Background(), "vm-1", VMStatePending, VMStateFailed, VMMetadata{
+		VMID:   "vm-1",
+		TaskID: &taskID,
+		UserID: userID,
+	}); err != nil {
+		t.Fatalf("OnStateChange() error = %v", err)
+	}
+
+	state, err := taskLifecycle.GetState(context.Background(), taskID)
+	if err != nil {
+		t.Fatalf("taskLifecycle.GetState() error = %v", err)
+	}
+	if state != consts.TaskStatusError {
+		t.Fatalf("task state = %s, want %s", state, consts.TaskStatusError)
+	}
+}

backend/pkg/register.go

@@ -231,6 +231,7 @@ func RegisterInfra(i *do.Injector, w ...*web.Web) error {
 		)
 
 		lc.Register(
+			lifecycle.NewVMTaskHook(i),
 			lifecycle.NewVMRecycleHook(i),
 		)