增加虚拟机回收校验和审计记录

backend/biz/host/usecase/host.go

@@ -27,9 +27,9 @@ import (
 	"github.com/chaitin/MonkeyCode/backend/errcode"
 	"github.com/chaitin/MonkeyCode/backend/pkg/cvt"
 	"github.com/chaitin/MonkeyCode/backend/pkg/delayqueue"
-	"github.com/chaitin/MonkeyCode/backend/pkg/entx"
 	"github.com/chaitin/MonkeyCode/backend/pkg/random"
 	"github.com/chaitin/MonkeyCode/backend/pkg/taskflow"
+	"github.com/chaitin/MonkeyCode/backend/pkg/vmrecycle"
 	"github.com/chaitin/MonkeyCode/backend/pkg/vmstatus"
 	"github.com/chaitin/MonkeyCode/backend/templates"
 )
@@ -44,6 +44,7 @@ type HostUsecase struct {
 	userRepo         domain.UserRepo
 	girepo           domain.GitIdentityRepo
 	vmexpireQueue    *delayqueue.VMExpireQueue
+	recycler         vmrecycle.Recycler
 	privilegeChecker domain.PrivilegeChecker // 可选，由内部项目通过 WithPrivilegeChecker 注入
 	tokenProvider    *gituc.TokenProvider
 }
@@ -59,6 +60,7 @@ func NewHostUsecase(i *do.Injector) (domain.HostUsecase, error) {
 		userRepo:      do.MustInvoke[domain.UserRepo](i),
 		girepo:        do.MustInvoke[domain.GitIdentityRepo](i),
 		vmexpireQueue: do.MustInvoke[*delayqueue.VMExpireQueue](i),
+		recycler:      do.MustInvoke[vmrecycle.Recycler](i),
 		tokenProvider: do.MustInvoke[*gituc.TokenProvider](i),
 	}
 
@@ -109,36 +111,8 @@ func (h *HostUsecase) vmexpireConsumer() {
 		err := h.vmexpireQueue.StartConsumer(context.Background(), VM_EXPIRE_QUEUE_KEY, func(ctx context.Context, job *delayqueue.Job[*domain.VmExpireInfo]) error {
 			innerLogger := logger.With("job", job)
 			innerLogger.InfoContext(ctx, "received expired virtualmachine")
-
-			ctx = entx.SkipSoftDelete(ctx)
-			vm, err := h.repo.GetVirtualMachine(ctx, job.Payload.VmID)
-			if err != nil {
-				innerLogger.ErrorContext(ctx, "failed to get vm", "error", err)
-				return nil
-			}
-
-			if err := h.taskflow.VirtualMachiner().Delete(ctx, &taskflow.DeleteVirtualMachineReq{
-				UserID: vm.UserID.String(),
-				HostID: vm.HostID,
-				ID:     vm.EnvironmentID,
-			}); err != nil {
-				innerLogger.ErrorContext(ctx, "failed to delete vm", "error", err)
-			}
-
-			if err := h.repo.UpdateVirtualMachine(ctx, vm.ID, func(vmuo *db.VirtualMachineUpdateOne) error {
-				vmuo.SetIsRecycled(true)
-				return nil
-			}); err != nil {
-				innerLogger.ErrorContext(ctx, "failed to update vm", "error", err)
-				return err
-			}
-
-			if err := h.markRecycledTasksFinished(ctx, vm); err != nil {
-				innerLogger.ErrorContext(ctx, "failed to finish recycled tasks", "error", err)
-				return err
-			}
-
-			return nil
+			_, err := h.recycler.ForceRecycle(ctx, job.Payload.VmID, consts.VMRecycleMethodExpired)
+			return err
 		})
 
 		h.logger.With("error", err, "index", index).WarnContext(context.Background(), "start consumer error retrying...")
@@ -534,15 +508,13 @@ func (h *HostUsecase) CreateVM(ctx context.Context, user *domain.User, req *doma
 // DeleteVM 删除虚拟机
 func (h *HostUsecase) DeleteVM(ctx context.Context, uid uuid.UUID, hostID, vmID string) error {
 	h.logger.InfoContext(ctx, "delete vm", "vmID", vmID, "user_id", uid, "host_id", hostID)
+	if _, err := h.repo.GetVirtualMachineWithUser(ctx, uid, vmID); err != nil {
+		return err
+	}
+	if _, err := h.recycler.ForceRecycle(ctx, vmID, consts.VMRecycleMethodManualDelete); err != nil {
+		return err
+	}
 	return h.repo.DeleteVirtualMachine(ctx, uid, hostID, vmID, func(vm *db.VirtualMachine) error {
-		if err := h.taskflow.VirtualMachiner().Delete(ctx, &taskflow.DeleteVirtualMachineReq{
-			UserID: uid.String(),
-			HostID: vm.HostID,
-			ID:     vm.EnvironmentID,
-		}); err != nil {
-			h.logger.ErrorContext(ctx, "failed to delete vm", "error", err)
-		}
-
 		// 清理 expired_at 过期队列中的残留任务
 		_ = h.vmexpireQueue.Remove(ctx, VM_EXPIRE_QUEUE_KEY, vm.ID)
 

backend/biz/task/usecase/task.go

@@ -369,9 +369,10 @@ func (a *TaskUsecase) Stop(ctx context.Context, user *domain.User, id uuid.UUID)
 		// 通过 lifecycle 回收 VM
 		if vm := tk.VirtualMachine; vm != nil {
 			if err := a.vmLifecycle.Transition(ctx, vm.ID, lifecycle.VMStateRecycled, lifecycle.VMMetadata{
-				VMID:   vm.ID,
-				TaskID: &id,
-				UserID: user.ID,
+				VMID:          vm.ID,
+				TaskID:        &id,
+				UserID:        user.ID,
+				RecycleMethod: consts.VMRecycleMethodTaskStop,
 			}); err != nil {
 				a.logger.WarnContext(ctx, "vm recycle transition failed", "error", err, "vm_id", vm.ID)
 			}
@@ -1069,9 +1070,10 @@ func (a *TaskUsecase) Delete(ctx context.Context, user *domain.User, id uuid.UUI
 		vm := vms[0]
 		if !vm.IsRecycled {
 			if err := a.vmLifecycle.Transition(ctx, vm.ID, lifecycle.VMStateRecycled, lifecycle.VMMetadata{
-				VMID:   vm.ID,
-				TaskID: &id,
-				UserID: user.ID,
+				VMID:          vm.ID,
+				TaskID:        &id,
+				UserID:        user.ID,
+				RecycleMethod: consts.VMRecycleMethodTaskDelete,
 			}); err != nil {
 				a.logger.WarnContext(ctx, "vm recycle transition failed on delete", "error", err, "vm_id", vm.ID)
 			}

backend/biz/vmidle/usecase/policy_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/redis/go-redis/v9"
 
 	"github.com/chaitin/MonkeyCode/backend/config"
+	"github.com/chaitin/MonkeyCode/backend/consts"
 	"github.com/chaitin/MonkeyCode/backend/db"
 	"github.com/chaitin/MonkeyCode/backend/db/enttest"
 	"github.com/chaitin/MonkeyCode/backend/db/virtualmachine"
@@ -333,6 +334,133 @@ func TestRecycleJobRetriesMarkWithoutDeletingRemotelyAgain(t *testing.T) {
 	}
 }
 
+func TestRecycleJobReschedulesForRecentPostgresActivity(t *testing.T) {
+	now := time.Now()
+	taskID := uuid.New()
+	vm := &db.VirtualMachine{
+		ID: "vm-recent-pg",
+		Edges: db.VirtualMachineEdges{Tasks: []*db.Task{{
+			ID: taskID,
+		}}},
+	}
+	taskRepo := &recycleGuardTaskRepoStub{task: &db.Task{ID: taskID, LastActiveAt: now.Add(-time.Minute)}}
+	recycler := &idleRecyclerStub{}
+	r := newRecycleGuardRefresher(t, vm, taskRepo, &taskLogActivityStub{latest: now.Add(-2 * time.Hour)}, recycler)
+	if err := r.redis.Set(context.Background(), "vm:idle:debounce:"+vm.ID+":activity", "1", time.Minute).Err(); err != nil {
+		t.Fatal(err)
+	}
+
+	err := r.handleRecycleJob(context.Background(), &delayqueue.Job[*domain.VmIdleInfo]{
+		Payload: &domain.VmIdleInfo{VmID: vm.ID, RecycleMethod: consts.VMRecycleMethodIdle},
+	})
+	if !errors.Is(err, delayqueue.ErrJobRescheduled) {
+		t.Fatalf("error = %v, want rescheduled", err)
+	}
+	if recycler.calls != 0 || taskRepo.calls != 1 {
+		t.Fatalf("recycle calls = %d, task calls = %d", recycler.calls, taskRepo.calls)
+	}
+	if _, _, ok, err := r.recycleQueue.GetJobInfo(context.Background(), vmrecycle.RecycleQueueKey, vm.ID); err != nil || !ok {
+		t.Fatalf("rescheduled job ok = %v, err = %v", ok, err)
+	}
+}
+
+func TestRecycleJobReschedulesForRecentClickHouseActivity(t *testing.T) {
+	now := time.Now()
+	taskID := uuid.New()
+	vm := &db.VirtualMachine{
+		ID: "vm-recent-clickhouse",
+		Edges: db.VirtualMachineEdges{Tasks: []*db.Task{{
+			ID: taskID,
+		}}},
+	}
+	taskRepo := &recycleGuardTaskRepoStub{task: &db.Task{ID: taskID, LastActiveAt: now.Add(-2 * time.Hour)}}
+	recycler := &idleRecyclerStub{}
+	r := newRecycleGuardRefresher(t, vm, taskRepo, &taskLogActivityStub{latest: now.Add(-time.Minute)}, recycler)
+
+	err := r.handleRecycleJob(context.Background(), &delayqueue.Job[*domain.VmIdleInfo]{
+		Payload: &domain.VmIdleInfo{VmID: vm.ID, RecycleMethod: consts.VMRecycleMethodIdle},
+	})
+	if !errors.Is(err, delayqueue.ErrJobRescheduled) {
+		t.Fatalf("error = %v, want rescheduled", err)
+	}
+	if recycler.calls != 0 || taskRepo.calls != 0 {
+		t.Fatalf("recycle calls = %d, task calls = %d", recycler.calls, taskRepo.calls)
+	}
+}
+
+func TestRecycleJobKeepsRetryingWhenActivityCheckFails(t *testing.T) {
+	taskID := uuid.New()
+	vm := &db.VirtualMachine{
+		ID: "vm-activity-check-error",
+		Edges: db.VirtualMachineEdges{Tasks: []*db.Task{{
+			ID: taskID,
+		}}},
+	}
+	wantErr := errors.New("clickhouse unavailable")
+	recycler := &idleRecyclerStub{}
+	r := newRecycleGuardRefresher(t, vm, &recycleGuardTaskRepoStub{}, &taskLogActivityStub{err: wantErr}, recycler)
+
+	err := r.handleRecycleJob(context.Background(), &delayqueue.Job[*domain.VmIdleInfo]{
+		Payload: &domain.VmIdleInfo{VmID: vm.ID, RecycleMethod: consts.VMRecycleMethodIdle},
+	})
+	if !errors.Is(err, wantErr) || !errors.Is(err, delayqueue.ErrRetryAfterMaxAttempts) {
+		t.Fatalf("error = %v, want activity and persistent retry errors", err)
+	}
+	if recycler.calls != 0 {
+		t.Fatalf("recycle calls = %d, want 0", recycler.calls)
+	}
+}
+
+func TestRecycleJobContinuesWhenBothActivityChecksAreStale(t *testing.T) {
+	now := time.Now()
+	taskID := uuid.New()
+	vm := &db.VirtualMachine{
+		ID: "vm-stale-activity",
+		Edges: db.VirtualMachineEdges{Tasks: []*db.Task{{
+			ID: taskID,
+		}}},
+	}
+	recycler := &idleRecyclerStub{}
+	r := newRecycleGuardRefresher(t, vm,
+		&recycleGuardTaskRepoStub{task: &db.Task{ID: taskID, LastActiveAt: now.Add(-2 * time.Hour)}},
+		&taskLogActivityStub{latest: now.Add(-2 * time.Hour)},
+		recycler,
+	)
+
+	if err := r.handleRecycleJob(context.Background(), &delayqueue.Job[*domain.VmIdleInfo]{
+		Payload: &domain.VmIdleInfo{VmID: vm.ID, RecycleMethod: consts.VMRecycleMethodIdle},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if recycler.calls != 1 {
+		t.Fatalf("recycle calls = %d, want 1", recycler.calls)
+	}
+}
+
+func TestRecycleJobRepairsAlreadyRecycledVM(t *testing.T) {
+	vm := &db.VirtualMachine{ID: "vm-already-recycled", IsRecycled: true}
+	recycler := &idleRecyclerStub{}
+	r := newRecycleGuardRefresher(t, vm, &recycleGuardTaskRepoStub{}, &taskLogActivityStub{}, recycler)
+
+	if err := r.handleRecycleJob(context.Background(), &delayqueue.Job[*domain.VmIdleInfo]{
+		Payload: &domain.VmIdleInfo{VmID: vm.ID, RecycleMethod: consts.VMRecycleMethodIdle},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if recycler.calls != 1 {
+		t.Fatalf("recycle calls = %d, want repair call", recycler.calls)
+	}
+}
+
+func newRecycleGuardRefresher(t *testing.T, vm *db.VirtualMachine, taskRepo domain.TaskRepo, activity taskLogActivityReader, recycler vmrecycle.Recycler) *vmIdleRefresher {
+	t.Helper()
+	r := newQueueTestVMIdleRefresher(newTestRedis(t), &refreshHostRepoStub{vm: vm})
+	r.taskRepo = taskRepo
+	r.taskLogActivity = activity
+	r.recycler = recycler
+	return r
+}
+
 func newQueueTestVMIdleRefresher(redisClient *redis.Client, repo domain.HostRepo) *vmIdleRefresher {
 	logger := slog.Default()
 	return &vmIdleRefresher{
@@ -409,11 +537,43 @@ type idleRecyclerStub struct {
 	calls int
 }
 
-func (s *idleRecyclerStub) Recycle(_ context.Context, vmID string) (vmrecycle.Result, error) {
+type recycleGuardTaskRepoStub struct {
+	domain.TaskRepo
+	task  *db.Task
+	err   error
+	calls int
+}
+
+func (s *recycleGuardTaskRepoStub) GetByID(context.Context, uuid.UUID) (*db.Task, error) {
+	s.calls++
+	return s.task, s.err
+}
+
+type taskLogActivityStub struct {
+	latest time.Time
+	ok     bool
+	err    error
+}
+
+func (s *taskLogActivityStub) LatestTaskLogTime(context.Context, uuid.UUID) (time.Time, bool, error) {
+	if s.err != nil {
+		return time.Time{}, false, s.err
+	}
+	if !s.ok && s.latest.IsZero() {
+		return time.Time{}, false, nil
+	}
+	return s.latest, true, nil
+}
+
+func (s *idleRecyclerStub) Recycle(_ context.Context, vmID string, _ consts.VMRecycleMethod) (vmrecycle.Result, error) {
 	s.calls++
 	return vmrecycle.Result{VMID: vmID}, s.err
 }
 
+func (s *idleRecyclerStub) ForceRecycle(context.Context, string, consts.VMRecycleMethod) (vmrecycle.Result, error) {
+	return vmrecycle.Result{}, errors.New("not implemented")
+}
+
 type refreshTeamPolicyRepoStub struct {
 	team *db.Team
 }