From 5513de0b98807b4d6865fc9d54c1b8740ba55f7c Mon Sep 17 00:00:00 2001
From: tong3jie <14191774+tong3jie@users.noreply.github.com>
Date: Thu, 12 Mar 2026 11:45:44 +0800
Subject: [PATCH] fix(cron): improve wake channel handling and enhance
 concurrency in tests

---
 pkg/cron/service.go      | 29 ++++++++++++++++++++---------
 pkg/cron/service_test.go |  3 ++-
 2 files changed, 22 insertions(+), 10 deletions(-)

diff --git a/pkg/cron/service.go b/pkg/cron/service.go
index d38009306..77a413133 100644
--- a/pkg/cron/service.go
+++ b/pkg/cron/service.go
@@ -74,6 +74,7 @@ func NewCronService(storePath string, onJob JobHandler) *CronService {
 		storePath: storePath,
 		onJob:     onJob,
 		gronx:     gronx.New(),
+		wakeChan:  make(chan struct{}),
 	}
 	// Initialize and load store on creation
 	cs.loadStore()
@@ -98,7 +99,9 @@ func (cs *CronService) Start() error {
 	}
 
 	cs.stopChan = make(chan struct{})
-	cs.wakeChan = make(chan struct{})
+	if cs.wakeChan == nil {
+		cs.wakeChan = make(chan struct{})
+	}
 	cs.running = true
 	go cs.runLoop(cs.stopChan)
 
@@ -332,6 +335,15 @@ func (cs *CronService) computeNextRun(schedule *CronSchedule, nowMS int64) *int6
 	}
 }
 
+// wake up the loop to re-evaluate next wake time immediately (e.g. after add/update/remove jobs)
+func (cs *CronService) notify() {
+	select {
+	case cs.wakeChan <- struct{}{}:
+	default:
+		// if the channel is full, it means the loop will wake up soon anyway, so we can skip sending
+	}
+}
+
 func (cs *CronService) recomputeNextRuns() {
 	now := time.Now().UnixMilli()
 	for i := range cs.store.Jobs {
@@ -433,10 +445,7 @@ func (cs *CronService) AddJob(
 		return nil, err
 	}
 
-	select {
-	case cs.wakeChan <- struct{}{}:
-	default:
-	}
+	cs.notify()
 
 	return &job, nil
 }
@@ -450,10 +459,7 @@ func (cs *CronService) UpdateJob(job *CronJob) error {
 			cs.store.Jobs[i] = *job
 			cs.store.Jobs[i].UpdatedAtMS = time.Now().UnixMilli()
 
-			select {
-			case cs.wakeChan <- struct{}{}:
-			default:
-			}
+			cs.notify()
 
 			return cs.saveStoreUnsafe()
 		}
@@ -485,6 +491,8 @@ func (cs *CronService) removeJobUnsafe(jobID string) bool {
 		}
 	}
 
+	cs.notify()
+
 	return removed
 }
 
@@ -507,6 +515,9 @@ func (cs *CronService) EnableJob(jobID string, enabled bool) *CronJob {
 			if err := cs.saveStoreUnsafe(); err != nil {
 				log.Printf("[cron] failed to save store after enable: %v", err)
 			}
+
+			cs.notify()
+
 			return job
 		}
 	}
diff --git a/pkg/cron/service_test.go b/pkg/cron/service_test.go
index c5cf288ac..c55e62174 100644
--- a/pkg/cron/service_test.go
+++ b/pkg/cron/service_test.go
@@ -214,6 +214,7 @@ func TestCronService_ConcurrentAccess(t *testing.T) {
 			for j := range iterations {
 				at := time.Now().Add(time.Hour).UnixMilli()
 				cs.AddJob(fmt.Sprintf("Job-%d-%d", id, j), CronSchedule{Kind: "at", AtMS: &at}, "", false, "", "")
+				time.Sleep(100 * time.Microsecond)
 			}
 		}(i)
 	}
@@ -227,7 +228,7 @@ func TestCronService_ConcurrentAccess(t *testing.T) {
 				if len(jobs) > 0 {
 					cs.EnableJob(jobs[0].ID, j%2 == 0)
 				}
-				time.Sleep(1 * time.Millisecond)
+				time.Sleep(100 * time.Microsecond)
 			}
 		}()
 	}