mirror of
https://github.com/go-gitea/gitea.git
synced 2026-11-02 13:44:16 +09:00
fix(actions): default-branch push discards a due schedule occurrence (#40073)
Default-branch pushes and Actions unit saves recreated the repository's schedules with `Next` recomputed from that moment, so a push between a cron's due minute and the scheduler's pass skipped the occurrence, and `@every` timers restarted. Re-registration now updates schedule rows in place and keeps the spec rows of unchanged crons, so their cursor survives and an in-flight scheduler pass still finds its schedule. Re-detection also clears schedules when no scheduled workflow remains. Fixes https://github.com/go-gitea/gitea/issues/40062 Written with AI assistance. --------- Co-authored-by: silverwind <me@silverwind.io> Co-authored-by: Claude (Opus 5.5) <noreply@anthropic.com>
This commit is contained in:
6 files changed
+111
-27
No files matched your search
@@ -41,26 +41,57 @@ func init() {
|
||||
db.RegisterModel(new(ActionSchedule))
|
||||
}
|
||||
|
||||
// CreateScheduleTaskBySchedules creates new schedule task.
|
||||
func CreateScheduleTaskBySchedules(ctx context.Context, rows []*ActionSchedule) error {
|
||||
// Return early if there are no rows to insert
|
||||
if len(rows) == 0 {
|
||||
return nil
|
||||
}
|
||||
type scheduleSpecKey struct {
|
||||
scheduleID int64
|
||||
spec string
|
||||
}
|
||||
|
||||
// ReplaceScheduleTasksByRepo updates schedules in place and keeps the specs of unchanged crons, so re-registering cannot skip a due occurrence.
|
||||
func ReplaceScheduleTasksByRepo(ctx context.Context, repoID int64, rows []*ActionSchedule) error {
|
||||
return db.WithTx(ctx, func(ctx context.Context) error {
|
||||
var oldSchedules []*ActionSchedule
|
||||
if err := db.GetEngine(ctx).Cols("id", "workflow_id").Where("repo_id = ?", repoID).Find(&oldSchedules); err != nil {
|
||||
return err
|
||||
}
|
||||
var oldSpecs []*ActionScheduleSpec
|
||||
if err := db.GetEngine(ctx).Cols("id", "schedule_id", "spec").Where("repo_id = ?", repoID).Find(&oldSpecs); err != nil {
|
||||
return err
|
||||
}
|
||||
scheduleIDs := make(map[string]int64, len(oldSchedules))
|
||||
for _, schedule := range oldSchedules {
|
||||
scheduleIDs[schedule.WorkflowID] = schedule.ID
|
||||
}
|
||||
specIDs := make(map[scheduleSpecKey]int64, len(oldSpecs))
|
||||
for _, spec := range oldSpecs {
|
||||
specIDs[scheduleSpecKey{spec.ScheduleID, spec.Spec}] = spec.ID
|
||||
}
|
||||
|
||||
var keptScheduleIDs, keptSpecIDs []int64
|
||||
|
||||
// Loop through each schedule row
|
||||
for _, row := range rows {
|
||||
row.Title = util.EllipsisDisplayString(row.Title, 255)
|
||||
// Create new schedule row
|
||||
if err := db.Insert(ctx, row); err != nil {
|
||||
if id, ok := scheduleIDs[row.WorkflowID]; ok {
|
||||
row.ID = id
|
||||
if _, err := db.GetEngine(ctx).ID(id).AllCols().Update(row); err != nil {
|
||||
return err
|
||||
}
|
||||
} else if err := db.Insert(ctx, row); err != nil {
|
||||
return err
|
||||
}
|
||||
keptScheduleIDs = append(keptScheduleIDs, row.ID)
|
||||
|
||||
// Loop through each schedule spec and create a new spec row
|
||||
now := time.Now()
|
||||
|
||||
for _, spec := range row.Specs {
|
||||
key := scheduleSpecKey{row.ID, spec}
|
||||
if id, ok := specIDs[key]; ok {
|
||||
delete(specIDs, key)
|
||||
keptSpecIDs = append(keptSpecIDs, id)
|
||||
continue
|
||||
}
|
||||
|
||||
specRow := &ActionScheduleSpec{
|
||||
RepoID: row.RepoID,
|
||||
ScheduleID: row.ID,
|
||||
@@ -81,9 +112,16 @@ func CreateScheduleTaskBySchedules(ctx context.Context, rows []*ActionSchedule)
|
||||
if err = db.Insert(ctx, specRow); err != nil {
|
||||
return err
|
||||
}
|
||||
keptSpecIDs = append(keptSpecIDs, specRow.ID)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
||||
// an empty NotIn matches every row
|
||||
if _, err := db.GetEngine(ctx).Where("repo_id = ?", repoID).NotIn("id", keptScheduleIDs).Delete(&ActionSchedule{}); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err := db.GetEngine(ctx).Where("repo_id = ?", repoID).NotIn("id", keptSpecIDs).Delete(&ActionScheduleSpec{})
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
// Copyright 2026 The Gitea Authors. All rights reserved.
|
||||
// SPDX-License-Identifier: MIT
|
||||
|
||||
package actions
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gitea.dev/models/unittest"
|
||||
"gitea.dev/modules/timeutil"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestReplaceScheduleTasksByRepo(t *testing.T) {
|
||||
require.NoError(t, unittest.PrepareTestDatabase())
|
||||
|
||||
const repoID = 4711
|
||||
register := func(commitSHA, spec string) {
|
||||
require.NoError(t, ReplaceScheduleTasksByRepo(t.Context(), repoID, []*ActionSchedule{
|
||||
{RepoID: repoID, WorkflowID: "nightly.yml", CommitSHA: commitSHA, Specs: []string{spec}},
|
||||
}))
|
||||
}
|
||||
|
||||
register("c1", "17 */6 * * *")
|
||||
spec := unittest.AssertExistsAndLoadBean(t, &ActionScheduleSpec{RepoID: repoID})
|
||||
spec.Next = timeutil.TimeStamp(time.Now().Add(-time.Minute).Unix())
|
||||
require.NoError(t, UpdateScheduleSpec(t.Context(), spec, "next"))
|
||||
|
||||
register("c2", "17 */6 * * *")
|
||||
unittest.AssertExistsAndLoadBean(t, &ActionSchedule{ID: spec.ScheduleID, CommitSHA: "c2"})
|
||||
unittest.AssertExistsAndLoadBean(t, &ActionScheduleSpec{ID: spec.ID, ScheduleID: spec.ScheduleID, Next: spec.Next})
|
||||
|
||||
register("c3", "23 */6 * * *")
|
||||
unittest.AssertNotExistsBean(t, &ActionScheduleSpec{ID: spec.ID})
|
||||
assert.Greater(t, unittest.AssertExistsAndLoadBean(t, &ActionScheduleSpec{RepoID: repoID}).Next, spec.Next)
|
||||
|
||||
require.NoError(t, ReplaceScheduleTasksByRepo(t.Context(), repoID, nil))
|
||||
unittest.AssertNotExistsBean(t, &ActionSchedule{RepoID: repoID})
|
||||
unittest.AssertNotExistsBean(t, &ActionScheduleSpec{RepoID: repoID})
|
||||
}
|
||||
@@ -595,18 +595,17 @@ func handleSchedules(
|
||||
return nil
|
||||
}
|
||||
|
||||
if count, err := db.Count[actions_model.ActionSchedule](ctx, actions_model.FindScheduleOptions{RepoID: input.Repo.ID}); err != nil {
|
||||
log.Error("CountSchedules: %v", err)
|
||||
return err
|
||||
} else if count > 0 {
|
||||
// replacing the schedules leaves their in-progress runs alone, as GitHub does
|
||||
if err := actions_model.DeleteScheduleTaskByRepo(ctx, input.Repo.ID); err != nil {
|
||||
log.Error("DeleteScheduleTaskByRepo: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
if len(detectedWorkflows) == 0 {
|
||||
log.Trace("repo %s with commit %s couldn't find schedules", input.Repo.FullName(), commit.ID)
|
||||
if count, err := db.Count[actions_model.ActionSchedule](ctx, actions_model.FindScheduleOptions{RepoID: input.Repo.ID}); err != nil {
|
||||
log.Error("CountSchedules: %v", err)
|
||||
return err
|
||||
} else if count > 0 {
|
||||
// dropping the schedules leaves their in-progress runs alone, as GitHub does
|
||||
if err := actions_model.DeleteScheduleTaskByRepo(ctx, input.Repo.ID); err != nil {
|
||||
log.Error("DeleteScheduleTaskByRepo: %v", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -648,7 +647,7 @@ func handleSchedules(
|
||||
crons = append(crons, run)
|
||||
}
|
||||
|
||||
return actions_model.CreateScheduleTaskBySchedules(ctx, crons)
|
||||
return actions_model.ReplaceScheduleTasksByRepo(ctx, input.Repo.ID, crons)
|
||||
}
|
||||
|
||||
// DetectAndHandleSchedules detects the schedule workflows on the default branch and create schedule tasks
|
||||
@@ -672,9 +671,6 @@ func DetectAndHandleSchedules(ctx context.Context, repo *repo_model.Repository)
|
||||
if err != nil {
|
||||
return fmt.Errorf("detect schedule workflows: %w", err)
|
||||
}
|
||||
if len(scheduleWorkflows) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// We need a notifyInput to call handleSchedules
|
||||
// if repo is a mirror, commit author maybe an external user,
|
||||
|
||||
@@ -128,3 +128,12 @@ func TestFilteredWorkflowCommitStatusForForkPullRequest(t *testing.T) {
|
||||
TriggerEvent: &jobparser.Event{Name: actions_module.GithubEventPullRequest},
|
||||
}))
|
||||
}
|
||||
|
||||
func TestDetectAndHandleSchedulesDropsRemovedWorkflows(t *testing.T) {
|
||||
require.NoError(t, unittest.PrepareTestDatabase())
|
||||
repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 1})
|
||||
require.NoError(t, actions_model.ReplaceScheduleTasksByRepo(t.Context(), repo.ID, []*actions_model.ActionSchedule{{RepoID: repo.ID, WorkflowID: "removed.yml"}}))
|
||||
|
||||
require.NoError(t, DetectAndHandleSchedules(t.Context(), repo))
|
||||
unittest.AssertNotExistsBean(t, &actions_model.ActionSchedule{RepoID: repo.ID})
|
||||
}
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"context"
|
||||
"slices"
|
||||
|
||||
actions_model "gitea.dev/models/actions"
|
||||
"gitea.dev/models/db"
|
||||
repo_model "gitea.dev/models/repo"
|
||||
"gitea.dev/models/unit"
|
||||
@@ -27,9 +26,6 @@ func UpdateRepositoryUnits(ctx context.Context, repo *repo_model.Repository, uni
|
||||
|
||||
for _, u := range units {
|
||||
if u.Type == unit.TypeActions {
|
||||
if err := actions_model.DeleteScheduleTaskByRepo(ctx, repo.ID); err != nil {
|
||||
log.Error("DeleteScheduleTaskByRepo: %v", err)
|
||||
}
|
||||
if err := actions_service.DetectAndHandleSchedules(ctx, repo); err != nil {
|
||||
log.Error("DetectAndHandleSchedules: %v", err)
|
||||
}
|
||||
|
||||
@@ -219,10 +219,12 @@ func testScheduleUpdateArchiveAndUnarchive(t *testing.T) {
|
||||
func testScheduleUpdateDisableAndEnableActionsUnit(t *testing.T) {
|
||||
doTestScheduleUpdate(t, func(t *testing.T, u *url.URL, testContext APITestContext, user *user_model.User, repo *repo_model.Repository) (expectedSpec string) {
|
||||
run := createScheduleRun(t, repo)
|
||||
spec := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionScheduleSpec{RepoID: repo.ID})
|
||||
doAPIEditRepository(testContext, &api.EditRepoOption{
|
||||
HasActions: new(true),
|
||||
})(t)
|
||||
assertRunStatus(t, run, actions_model.StatusWaiting)
|
||||
unittest.AssertExistsAndLoadBean(t, &actions_model.ActionScheduleSpec{ID: spec.ID, Next: spec.Next})
|
||||
|
||||
doAPIEditRepository(testContext, &api.EditRepoOption{
|
||||
HasActions: new(false),
|
||||
|
||||
Reference in new issue
Block a user