mirror of
https://github.com/QuantumNous/new-api.git
synced 2026-09-12 15:21:09 +00:00
fix(model): do not treat no-op system task state writes as lock loss (#7135)
This commit is contained in:
+17
-1
@@ -322,7 +322,23 @@ func UpdateSystemTaskState(taskID string, lockedBy string, state any) error {
|
|||||||
if result.Error != nil {
|
if result.Error != nil {
|
||||||
return result.Error
|
return result.Error
|
||||||
}
|
}
|
||||||
if result.RowsAffected == 0 {
|
if result.RowsAffected > 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
// MySQL counts changed rows, not matched rows. A no-op persist of the same
|
||||||
|
// state in the same second therefore returns RowsAffected == 0 even while
|
||||||
|
// the lease is still held. Confirm the lock before treating this as loss.
|
||||||
|
// Reuse `now` from the UPDATE so a clock tick cannot reintroduce false
|
||||||
|
// lock-loss; a lease that expires during the write is caught by the next heartbeat.
|
||||||
|
var held int64
|
||||||
|
err = DB.Model(&SystemTask{}).
|
||||||
|
Where("task_id = ? AND status = ? AND locked_by = ?", taskID, SystemTaskStatusRunning, lockedBy).
|
||||||
|
Where("EXISTS (SELECT 1 FROM system_task_locks WHERE system_task_locks.task_id = system_tasks.task_id AND system_task_locks.locked_by = ? AND system_task_locks.locked_until >= ?)", lockedBy, now).
|
||||||
|
Count(&held).Error
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if held == 0 {
|
||||||
return ErrSystemTaskLockLost
|
return ErrSystemTaskLockLost
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -350,3 +350,34 @@ func TestSystemTaskUpdatesRequireUnexpiredLock(t *testing.T) {
|
|||||||
assert.Equal(t, SystemTaskStatusRunning, reloaded.Status)
|
assert.Equal(t, SystemTaskStatusRunning, reloaded.Status)
|
||||||
assert.Empty(t, reloaded.State)
|
assert.Empty(t, reloaded.State)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestUpdateSystemTaskStateIdenticalPayloadDoesNotLoseLock(t *testing.T) {
|
||||||
|
// SQLite reports matched rows for unchanged UPDATEs, so this case passed
|
||||||
|
// even before the fix. The MySQL regression is covered by
|
||||||
|
// TestUpdateSystemTaskStateIdenticalPayloadDoesNotLoseLockConfiguredDatabases.
|
||||||
|
truncateTables(t)
|
||||||
|
runUpdateSystemTaskStateIdenticalPayloadKeepsLock(t, SystemTaskTypeLogCleanup)
|
||||||
|
}
|
||||||
|
|
||||||
|
func runUpdateSystemTaskStateIdenticalPayloadKeepsLock(t *testing.T, taskType string) {
|
||||||
|
t.Helper()
|
||||||
|
// Two persists in the same second so MySQL's unchanged-row UPDATE returns 0.
|
||||||
|
|
||||||
|
task, err := CreateSystemTask(taskType, nil, nil)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
runnerID := "runner-a"
|
||||||
|
_, claimed, err := ClaimSystemTask(task.ID, taskType, runnerID, common.GetTimestamp()+60)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.True(t, claimed)
|
||||||
|
|
||||||
|
state := testSystemTaskState{Total: 10, Processed: 10, Progress: 100, Remaining: 0}
|
||||||
|
require.NoError(t, UpdateSystemTaskState(task.TaskID, runnerID, state))
|
||||||
|
require.NoError(t, UpdateSystemTaskState(task.TaskID, runnerID, state), "identical state persist must not be treated as lock loss")
|
||||||
|
|
||||||
|
require.NoError(t, FinishSystemTask(task.TaskID, runnerID, SystemTaskStatusSucceeded, map[string]int64{"deleted_count": 10}, ""))
|
||||||
|
finished, err := GetSystemTaskByTaskID(task.TaskID)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NotNil(t, finished)
|
||||||
|
assert.Equal(t, SystemTaskStatusSucceeded, finished.Status)
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user