[fix][evaluation] clean up runs rejected before dispatch

backend/modules/evaluation/domain/service/expt_manage_execution_impl.go

@@ -303,12 +303,48 @@ func withRetryYieldExt(ext map[string]string, enabled bool) map[string]string {
 	return ext
 }
 
-func (e *ExptMangerImpl) Run(ctx context.Context, exptID, runID, spaceID int64, itemRetryNum int, session *entity.Session, runMode entity.ExptRunMode, ext map[string]string) error {
-	if err := NewQuotaService(e.quotaRepo, e.configer).AllowExptRun(ctx, exptID, spaceID, session); err != nil {
-		return err
+func (e *ExptMangerImpl) prepareRun(ctx context.Context, exptID, runID, spaceID int64, session *entity.Session) (expt *entity.Experiment, err error) {
+	defer func() {
+		if err != nil {
+			e.cleanupUnscheduledRun(ctx, exptID, runID, err)
+		}
+	}()
+
+	// Read dependencies before reserving quota so read failures cannot leak a quota slot.
+	expt, err = e.GetDetail(ctx, exptID, spaceID, session)
+	if err != nil {
+		return nil, err
 	}
+	err = NewQuotaService(e.quotaRepo, e.configer).AllowExptRun(ctx, exptID, spaceID, session)
+	return expt, err
+}
 
-	expt, err := e.GetDetail(ctx, exptID, spaceID, session)
+func (e *ExptMangerImpl) cleanupUnscheduledRun(ctx context.Context, exptID, runID int64, cause error) {
+	if runID <= 0 {
+		logs.CtxError(ctx, "[ExptEval][RunLock] invalid cleanup run [expt_id=%v run_id=%v]", exptID, runID)
+		return
+	}
+
+	// Only the owned run before MQ publication may be unlocked here.
+	// Attempt this bare DEL synchronously once; a delayed retry could delete a new run's lock.
+	ctx = context.WithoutCancel(ctx)
+	stateCtx, cancelState := context.WithTimeout(ctx, exptRunLogPersistTimeout)
+	stateErr := e.runLogRepo.Update(stateCtx, exptID, runID, map[string]any{
+		"status": int64(entity.ExptStatus_Failed), "status_message": []byte(cause.Error()),
+	})
+	cancelState()
+
+	unlockCtx, cancelUnlock := context.WithTimeout(ctx, 2*time.Second)
+	released, unlockErr := e.mutex.UnlockForce(unlockCtx, e.makeExptMutexLockKey(exptID))
+	cancelUnlock()
+	if stateErr != nil || unlockErr != nil {
+		logs.CtxError(ctx, "[ExptEval][RunLock] cleanup failed [expt_id=%v run_id=%v state_err=%v unlock_err=%v]", exptID, runID, stateErr, unlockErr)
+	}
+	logs.CtxInfo(ctx, "[ExptEval][RunLock] cleanup result [expt_id=%v run_id=%v released=%v]", exptID, runID, released)
+}
+
+func (e *ExptMangerImpl) Run(ctx context.Context, exptID, runID, spaceID int64, itemRetryNum int, session *entity.Session, runMode entity.ExptRunMode, ext map[string]string) error {
+	expt, err := e.prepareRun(ctx, exptID, runID, spaceID, session)
 	if err != nil {
 		return err
 	}
@@ -414,11 +450,7 @@ func buildExptNotifyParam(expt *entity.Experiment, toStatus entity.ExptStatus) (
 }
 
 func (e *ExptMangerImpl) RetryItems(ctx context.Context, exptID, runID, spaceID int64, itemRetryNum int, itemIDs []int64, session *entity.Session, ext map[string]string) error {
-	if err := NewQuotaService(e.quotaRepo, e.configer).AllowExptRun(ctx, exptID, spaceID, session); err != nil {
-		return err
-	}
-
-	expt, err := e.GetDetail(ctx, exptID, spaceID, session)
+	expt, err := e.prepareRun(ctx, exptID, runID, spaceID, session)
 	if err != nil {
 		return err
 	}
@@ -1679,15 +1711,22 @@ func (e *ExptMangerImpl) unlockCompletingRun(ctx context.Context, exptID, exptRu
 	return err
 }
 
-func (e *ExptMangerImpl) LogRun(ctx context.Context, exptID, exptRunID int64, mode entity.ExptRunMode, spaceID int64, itemIDs []int64, session *entity.Session) error {
+func (e *ExptMangerImpl) LogRun(ctx context.Context, exptID, exptRunID int64, mode entity.ExptRunMode, spaceID int64, itemIDs []int64, session *entity.Session) (err error) {
 	duration := time.Duration(e.configer.GetExptExecConf(ctx, spaceID).GetZombieIntervalSecond()) * time.Second
 	locked, err := e.mutex.LockBackoff(ctx, e.makeExptMutexLockKey(exptID), duration, time.Second)
 	if err != nil {
 		return err
 	}
 	if !locked {
+		logs.CtxInfo(ctx, "[ExptEval][RunLock] lock occupied [expt_id=%v run_id=%v]", exptID, exptRunID)
 		return errorx.NewByCode(errno.ExperimentRunningExistedCode)
 	}
+	ownedRunID := exptRunID
+	defer func(runID int64) {
+		if err != nil {
+			e.cleanupUnscheduledRun(ctx, exptID, runID, err)
+		}
+	}(ownedRunID)
 
 	defer e.mtr.EmitExptExecRun(spaceID, int64(mode))
 
@@ -1733,16 +1772,25 @@ func (e *ExptMangerImpl) LogRetryItemsRun(ctx context.Context, exptID int64, mod
 	if err != nil {
 		return 0, false, err
 	}
+	if locked {
+		ownedRunID := runID
+		defer func(runID int64) {
+			if err != nil {
+				e.cleanupUnscheduledRun(ctx, exptID, runID, err)
+			}
+		}(ownedRunID)
+	}
 
 	var rl *entity.ExptRunLog
 	retried = !locked
 
 	if retried {
 		runID, err = strconv.ParseInt(existedRunID, 10, 64)
 		if err != nil {
-			logs.CtxError(ctx, "parsing expt run lock value to runid failed, raw: %v", existedRunID)
+			logs.CtxInfo(ctx, "[ExptEval][RunLock] lock occupied [expt_id=%v holder=%s]", exptID, existedRunID)
 			return 0, false, errorx.NewByCode(errno.ExperimentRunningExistedCode)
 		}
+		logs.CtxDebug(ctx, "[ExptEval][RunLock] joining existing run [expt_id=%v run_id=%v]", exptID, runID)
 
 		completing, err := e.ExistCompletingRunLock(ctx, exptID, runID, spaceID)
 		if err != nil {

backend/modules/evaluation/domain/service/expt_manage_execution_impl_test.go

@@ -153,6 +153,8 @@ func TestExptMangerImpl_Run(t *testing.T) {
 					EXPECT().
 					CreateOrUpdate(ctx, int64(789), gomock.Any(), session).
 					Return(errors.New("quota exceeded"))
+				mgr.runLogRepo.(*repoMocks.MockIExptRunLogRepo).EXPECT().Update(gomock.Any(), int64(123), int64(456), gomock.Any()).Return(nil)
+				mgr.mutex.(*lockMocks.MockILocker).EXPECT().UnlockForce(gomock.Any(), mgr.makeExptMutexLockKey(123)).Return(true, nil)
 			},
 			wantErr: true,
 		},
@@ -622,6 +624,8 @@ func TestExptMangerImpl_LogRun(t *testing.T) {
 					EXPECT().
 					Create(ctx, gomock.Any()).
 					Return(errors.New("create failed"))
+				mgr.runLogRepo.(*repoMocks.MockIExptRunLogRepo).EXPECT().Update(gomock.Any(), int64(123), int64(456), gomock.Any()).Return(nil)
+				mgr.mutex.(*lockMocks.MockILocker).EXPECT().UnlockForce(gomock.Any(), mgr.makeExptMutexLockKey(123)).Return(true, nil)
 			},
 			wantErr: true,
 		},
@@ -802,6 +806,8 @@ func TestExptMangerImpl_LogRetryItemsRun(t *testing.T) {
 					Return(true, "1006", nil)
 				mgr.runLogRepo.(*repoMocks.MockIExptRunLogRepo).
 					EXPECT().Save(ctx, gomock.Any()).Return(errors.New("save failed"))
+				mgr.runLogRepo.(*repoMocks.MockIExptRunLogRepo).EXPECT().Update(gomock.Any(), exptID, int64(1006), gomock.Any()).Return(nil)
+				mgr.mutex.(*lockMocks.MockILocker).EXPECT().UnlockForce(gomock.Any(), mgr.makeExptMutexLockKey(exptID)).Return(true, nil)
 			},
 			wantErr: true,
 		},
@@ -898,6 +904,8 @@ func TestExptMangerImpl_RetryItems(t *testing.T) {
 			setup: func() {
 				mgr.quotaRepo.(*repoMocks.MockQuotaRepo).
 					EXPECT().CreateOrUpdate(ctx, spaceID, gomock.Any(), session).Return(errors.New("quota exceeded"))
+				mgr.runLogRepo.(*repoMocks.MockIExptRunLogRepo).EXPECT().Update(gomock.Any(), exptID, runID, gomock.Any()).Return(nil)
+				mgr.mutex.(*lockMocks.MockILocker).EXPECT().UnlockForce(gomock.Any(), mgr.makeExptMutexLockKey(exptID)).Return(true, nil)
 				mgr.configer.(*componentMocks.MockIConfiger).
 					EXPECT().GetExptExecConf(ctx, spaceID).AnyTimes().
 					Return(&entity.ExptExecConf{SpaceExptConcurLimit: 10})

backend/modules/evaluation/domain/service/expt_run_admission_cleanup_test.go

@@ -0,0 +1,232 @@
+// Copyright (c) 2025 coze-dev Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package service
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+
+	idgenMocks "github.com/coze-dev/coze-loop/backend/infra/idgen/mocks"
+	lockMocks "github.com/coze-dev/coze-loop/backend/infra/lock/mocks"
+	lwtMocks "github.com/coze-dev/coze-loop/backend/infra/platestwrite/mocks"
+	metricsMocks "github.com/coze-dev/coze-loop/backend/modules/evaluation/domain/component/metrics/mocks"
+	componentMocks "github.com/coze-dev/coze-loop/backend/modules/evaluation/domain/component/mocks"
+	"github.com/coze-dev/coze-loop/backend/modules/evaluation/domain/entity"
+	eventsMocks "github.com/coze-dev/coze-loop/backend/modules/evaluation/domain/events/mocks"
+	repoMocks "github.com/coze-dev/coze-loop/backend/modules/evaluation/domain/repo/mocks"
+	svcMocks "github.com/coze-dev/coze-loop/backend/modules/evaluation/domain/service/mocks"
+	"github.com/coze-dev/coze-loop/backend/modules/evaluation/pkg/errno"
+	"github.com/coze-dev/coze-loop/backend/pkg/errorx"
+)
+
+func TestRunAdmissionCleanup(t *testing.T) {
+	for _, tc := range []struct {
+		name        string
+		mode        entity.ExptRunMode
+		publishFail bool
+	}{
+		{"submit", entity.EvaluationModeSubmit, false},
+		{"retry_all", entity.EvaluationModeRetryAll, false},
+		{"retry_failure", entity.EvaluationModeFailRetry, false},
+		{"retry_items", entity.EvaluationModeRetryItems, false},
+		{"publish_failure_preserves_run", entity.EvaluationModeFailRetry, true},
+		{"publish_failure_preserves_retry_items_run", entity.EvaluationModeRetryItems, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			mgr := newTestExptManager(ctrl)
+			ctx := context.Background()
+			session := &entity.Session{UserID: "u"}
+			const exptID, runID, spaceID = int64(1), int64(2), int64(3)
+			limit, logCalls := 2, 2
+			publishErr := errors.New("publish timed out")
+			if tc.publishFail {
+				limit, logCalls = 3, 1
+				mgr.publisher.(*eventsMocks.MockExptEventPublisher).EXPECT().PublishExptScheduleEvent(ctx, gomock.Any(), gomock.Any()).Return(publishErr)
+			}
+			configer := componentMocks.NewMockIConfiger(ctrl)
+			configer.EXPECT().GetExptExecConf(ctx, spaceID).Return(&entity.ExptExecConf{
+				SpaceExptConcurLimit: limit, ZombieIntervalSecond: 72 * 60 * 60,
+			}).AnyTimes()
+			configer.EXPECT().GetRetryYieldEnabled(ctx, spaceID).Return(false).AnyTimes()
+			mgr.configer = configer
+			quota := &entity.QuotaSpaceExpt{ExptID2RunTime: map[int64]int64{10: time.Now().Unix(), 11: time.Now().Unix()}}
+			mgr.quotaRepo.(*repoMocks.MockQuotaRepo).EXPECT().CreateOrUpdate(ctx, spaceID, gomock.Any(), session).
+				DoAndReturn(func(_ context.Context, _ int64, update func(*entity.QuotaSpaceExpt) (*entity.QuotaSpaceExpt, bool, error), _ *entity.Session) error {
+					_, changed, err := update(quota)
+					require.Equal(t, tc.publishFail, changed)
+					return err
+				})
+			mgr.lwt.(*lwtMocks.MockILatestWriteTracker).EXPECT().CheckWriteFlagByID(ctx, gomock.Any(), exptID).Return(false).AnyTimes()
+			mgr.exptRepo.(*repoMocks.MockIExperimentRepo).EXPECT().MGetByID(ctx, []int64{exptID}, spaceID).
+				Return([]*entity.Experiment{{ID: exptID, SpaceID: spaceID, Status: entity.ExptStatus_Failed}}, nil).AnyTimes()
+			mgr.evaluationSetService.(*svcMocks.MockIEvaluationSetService).EXPECT().
+				GetEvaluationSet(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Nil()).Return(&entity.EvaluationSet{}, nil).AnyTimes()
+			mgr.exptResultService.(*svcMocks.MockExptResultService).EXPECT().
+				MGetStats(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
+			mgr.exptAggrResultService.(*svcMocks.MockExptAggrResultService).EXPECT().
+				BatchGetExptAggrResultByExperimentIDs(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
+
+			var runLog *entity.ExptRunLog
+			locked := false
+			lockKey := mgr.makeExptMutexLockKey(exptID)
+			mgr.mtr.(*metricsMocks.MockExptMetric).EXPECT().EmitExptExecRun(spaceID, int64(tc.mode)).Times(logCalls)
+			var latestRunID int64
+			mgr.exptRepo.(*repoMocks.MockIExperimentRepo).EXPECT().Update(ctx, gomock.Any()).
+				DoAndReturn(func(_ context.Context, expt *entity.Experiment) error { latestRunID = expt.LatestRunID; return nil }).Times(logCalls)
+			if tc.mode == entity.EvaluationModeRetryItems {
+				mgr.idgenerator.(*idgenMocks.MockIIDGenerator).EXPECT().GenID(ctx).Return(runID, nil).Times(logCalls)
+				mgr.mutex.(*lockMocks.MockILocker).EXPECT().BackoffLockWithValue(ctx, lockKey, strconv.FormatInt(runID, 10), gomock.Any(), gomock.Any()).
+					DoAndReturn(func(context.Context, string, string, time.Duration, time.Duration) (bool, string, error) {
+						if locked {
+							return false, strconv.FormatInt(runID, 10), nil
+						}
+						locked = true
+						return true, "", nil
+					}).Times(logCalls)
+				mgr.runLogRepo.(*repoMocks.MockIExptRunLogRepo).EXPECT().Save(gomock.Any(), gomock.Any()).
+					DoAndReturn(func(_ context.Context, got *entity.ExptRunLog) error { runLog = got; return nil }).Times(logCalls)
+			} else {
+				mgr.mutex.(*lockMocks.MockILocker).EXPECT().LockBackoff(ctx, lockKey, gomock.Any(), gomock.Any()).
+					DoAndReturn(func(context.Context, string, time.Duration, time.Duration) (bool, error) {
+						if locked {
+							return false, nil
+						}
+						locked = true
+						return true, nil
+					}).Times(logCalls)
+				mgr.runLogRepo.(*repoMocks.MockIExptRunLogRepo).EXPECT().Create(ctx, gomock.Any()).
+					DoAndReturn(func(_ context.Context, got *entity.ExptRunLog) error { runLog = got; return nil }).Times(logCalls)
+			}
+			if !tc.publishFail {
+				mgr.runLogRepo.(*repoMocks.MockIExptRunLogRepo).EXPECT().Update(gomock.Any(), exptID, runID, gomock.Any()).
+					DoAndReturn(func(_ context.Context, _, _ int64, fields map[string]any) error {
+						runLog.Status = fields["status"].(int64)
+						runLog.StatusMessage = fields["status_message"].([]byte)
+						return nil
+					})
+				mgr.mutex.(*lockMocks.MockILocker).EXPECT().UnlockForce(gomock.Any(), lockKey).
+					DoAndReturn(func(context.Context, string) (bool, error) {
+						require.Equal(t, int64(entity.ExptStatus_Failed), runLog.Status)
+						locked = false
+						return true, nil
+					})
+			}
+
+			var err error
+			if tc.mode == entity.EvaluationModeRetryItems {
+				_, retried, logErr := mgr.LogRetryItemsRun(ctx, exptID, tc.mode, spaceID, []int64{4}, session)
+				require.NoError(t, logErr)
+				require.False(t, retried)
+				err = mgr.RetryItems(ctx, exptID, runID, spaceID, 0, []int64{4}, session, nil)
+			} else {
+				require.NoError(t, mgr.LogRun(ctx, exptID, runID, tc.mode, spaceID, nil, session))
+				err = mgr.Run(ctx, exptID, runID, spaceID, 0, session, tc.mode, nil)
+			}
+			if tc.publishFail {
+				require.ErrorIs(t, err, publishErr)
+				require.True(t, locked)
+				require.Equal(t, int64(entity.ExptStatus_Pending), runLog.Status)
+				return
+			}
+			status, ok := errorx.FromStatusError(err)
+			require.True(t, ok)
+			require.Equal(t, int32(errno.ExperimentRunningCountLimitCode), status.Code())
+			require.Len(t, quota.ExptID2RunTime, 2)
+			require.NotContains(t, quota.ExptID2RunTime, exptID)
+			require.False(t, locked)
+			require.Equal(t, int64(entity.ExptStatus_Failed), runLog.Status)
+			require.Contains(t, string(runLog.StatusMessage), "max limit: 2")
+			require.Equal(t, runID, latestRunID, "admission cleanup must leave the latest run pointing at the failed attempt")
+			if tc.mode == entity.EvaluationModeRetryItems {
+				_, retried, logErr := mgr.LogRetryItemsRun(ctx, exptID, tc.mode, spaceID, []int64{4}, session)
+				require.NoError(t, logErr)
+				require.False(t, retried)
+			} else {
+				require.NoError(t, mgr.LogRun(ctx, exptID, runID+1, tc.mode, spaceID, nil, session))
+			}
+			require.True(t, locked)
+		})
+	}
+}
+
+func TestRunPreparationReadFailure(t *testing.T) {
+	for _, tc := range []struct {
+		name                                              string
+		updateFails, updateTimesOut, updateCommittedError bool
+		lockPresent, released                             bool
+		unlockErr                                         error
+	}{
+		{name: "success", lockPresent: true, released: true},
+		{name: "update_failure_still_unlocks", updateFails: true, lockPresent: true, released: true},
+		{name: "update_committed_then_error_still_unlocks", updateCommittedError: true, lockPresent: true, released: true},
+		{name: "update_timeout_leaves_full_unlock_budget", updateTimesOut: true, lockPresent: true, released: true},
+		{name: "lock_already_absent"},
+		{name: "unlock_error_without_delete_preserves_original", lockPresent: true, unlockErr: errors.New("redis unavailable")},
+		{name: "update_and_unlock_fail_preserves_original", updateFails: true, lockPresent: true, unlockErr: errors.New("redis unavailable")},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			mgr := newTestExptManager(ctrl)
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			const exptID, runID, spaceID = int64(1), int64(2), int64(3)
+			mgr.lwt.(*lwtMocks.MockILatestWriteTracker).EXPECT().CheckWriteFlagByID(ctx, gomock.Any(), exptID).Return(false)
+			mgr.exptRepo.(*repoMocks.MockIExperimentRepo).EXPECT().MGetByID(ctx, []int64{exptID}, spaceID).
+				DoAndReturn(func(context.Context, []int64, int64) ([]*entity.Experiment, error) {
+					cancel()
+					return nil, context.Canceled
+				})
+			runLog := &entity.ExptRunLog{ExptID: exptID, ExptRunID: runID, Status: int64(entity.ExptStatus_Pending)}
+			locked := tc.lockPresent
+			var updateCtx context.Context
+			mgr.runLogRepo.(*repoMocks.MockIExptRunLogRepo).EXPECT().Update(gomock.Any(), exptID, runID, gomock.Any()).
+				DoAndReturn(func(cleanupCtx context.Context, _, _ int64, fields map[string]any) error {
+					updateCtx = cleanupCtx
+					require.NoError(t, cleanupCtx.Err())
+					deadline, ok := cleanupCtx.Deadline()
+					require.True(t, ok)
+					require.InDelta(t, 5, time.Until(deadline).Seconds(), 1)
+					if tc.updateTimesOut {
+						<-cleanupCtx.Done()
+						return cleanupCtx.Err()
+					}
+					if tc.updateFails {
+						return errors.New("run log unavailable")
+					}
+					runLog.Status = fields["status"].(int64)
+					runLog.StatusMessage = fields["status_message"].([]byte)
+					if tc.updateCommittedError {
+						return errors.New("run log response lost")
+					}
+					return nil
+				})
+			mgr.mutex.(*lockMocks.MockILocker).EXPECT().UnlockForce(gomock.Any(), mgr.makeExptMutexLockKey(exptID)).
+				DoAndReturn(func(unlockCtx context.Context, _ string) (bool, error) {
+					require.NotSame(t, updateCtx, unlockCtx)
+					require.NoError(t, unlockCtx.Err())
+					deadline, ok := unlockCtx.Deadline()
+					require.True(t, ok)
+					require.InDelta(t, 2, time.Until(deadline).Seconds(), 0.5)
+					if tc.unlockErr == nil {
+						locked = false
+					}
+					return tc.released, tc.unlockErr
+				})
+			err := mgr.Run(ctx, exptID, runID, spaceID, 0, &entity.Session{UserID: "u"}, entity.EvaluationModeFailRetry, nil)
+			require.ErrorIs(t, err, context.Canceled)
+			require.Equal(t, tc.lockPresent && tc.unlockErr != nil, locked)
+			if !tc.updateFails && !tc.updateTimesOut {
+				require.Equal(t, int64(entity.ExptStatus_Failed), runLog.Status)
+				require.Equal(t, context.Canceled.Error(), string(runLog.StatusMessage))
+			}
+		})
+	}
+}