opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(cron): reload shared store before writes

Browse source
This commit is contained in:
Alix-007 2026-03-17 14:38:28 +08:00
parent 4a8a2e9c23
commit f48eea6c73
2 changed files with 161 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
pkg/cron/service.go
View file

@ -140,6 +140,12 @@ func (cs *CronService) checkJobs() {
return
}
if err := cs.reloadStoreUnsafe(); err != nil {
log.Printf("[cron] failed to reload store: %v", err)
cs.mu.Unlock()
return
}
now := time.Now().UnixMilli()
var dueJobIDs []string
@ -209,6 +215,10 @@ func (cs *CronService) executeJobByID(jobID string) {
cs.mu.Lock()
defer cs.mu.Unlock()
if err := cs.reloadStoreUnsafe(); err != nil {
log.Printf("[cron] failed to reload store before updating job state: %v", err)
}
var job *CronJob
for i := range cs.store.Jobs {
if cs.store.Jobs[i].ID == jobID {
@ -333,6 +343,15 @@ func (cs *CronService) SetOnJob(handler JobHandler) {
cs.onJob = handler
}
func (cs *CronService) reloadStoreUnsafe() error {
current := cs.store
if err := cs.loadStore(); err != nil {
cs.store = current
return err
}
return nil
}
func (cs *CronService) loadStore() error {
cs.store = &CronStore{
Version: 1,
@ -370,6 +389,10 @@ func (cs *CronService) AddJob(
cs.mu.Lock()
defer cs.mu.Unlock()
if err := cs.reloadStoreUnsafe(); err != nil {
log.Printf("[cron] failed to reload store before add: %v", err)
}
now := time.Now().UnixMilli()
// One-time tasks (at) should be deleted after execution
@ -407,6 +430,10 @@ func (cs *CronService) UpdateJob(job *CronJob) error {
cs.mu.Lock()
defer cs.mu.Unlock()
if err := cs.reloadStoreUnsafe(); err != nil {
log.Printf("[cron] failed to reload store before update: %v", err)
}
for i := range cs.store.Jobs {
if cs.store.Jobs[i].ID == job.ID {
cs.store.Jobs[i] = *job
@ -421,6 +448,10 @@ func (cs *CronService) RemoveJob(jobID string) bool {
cs.mu.Lock()
defer cs.mu.Unlock()
if err := cs.reloadStoreUnsafe(); err != nil {
log.Printf("[cron] failed to reload store before remove: %v", err)
}
return cs.removeJobUnsafe(jobID)
}
@ -448,6 +479,10 @@ func (cs *CronService) EnableJob(jobID string, enabled bool) *CronJob {
cs.mu.Lock()
defer cs.mu.Unlock()
if err := cs.reloadStoreUnsafe(); err != nil {
log.Printf("[cron] failed to reload store before enable: %v", err)
}
for i := range cs.store.Jobs {
job := &cs.store.Jobs[i]
if job.ID == jobID {
@ -471,14 +506,18 @@ func (cs *CronService) EnableJob(jobID string, enabled bool) *CronJob {
}
func (cs *CronService) ListJobs(includeDisabled bool) []CronJob {
cs.mu.RLock()
defer cs.mu.RUnlock()
cs.mu.Lock()
defer cs.mu.Unlock()
if includeDisabled {
return cs.store.Jobs
if err := cs.reloadStoreUnsafe(); err != nil {
log.Printf("[cron] failed to reload store before list: %v", err)
}
var enabled []CronJob
if includeDisabled {
return append([]CronJob(nil), cs.store.Jobs...)
}
enabled := make([]CronJob, 0, len(cs.store.Jobs))
for _, job := range cs.store.Jobs {
if job.Enabled {
enabled = append(enabled, job)
@ -489,8 +528,12 @@ func (cs *CronService) ListJobs(includeDisabled bool) []CronJob {
}
func (cs *CronService) Status() map[string]any {
cs.mu.RLock()
defer cs.mu.RUnlock()
cs.mu.Lock()
defer cs.mu.Unlock()
if err := cs.reloadStoreUnsafe(); err != nil {
log.Printf("[cron] failed to reload store before status: %v", err)
}
var enabledCount int
for _, job := range cs.store.Jobs {

111
pkg/cron/service_test.go
View file

@ -33,6 +33,117 @@ func TestSaveStore_FilePermissions(t *testing.T) {
}
}
func TestEnableJob_PreservesJobsAddedByAnotherService(t *testing.T) {
tmpDir := t.TempDir()
storePath := filepath.Join(tmpDir, "cron", "jobs.json")
seed := NewCronService(storePath, nil)
existingJob, err := seed.AddJob(
"existing",
CronSchedule{Kind: "every", EveryMS: int64Ptr(60000)},
"hello",
false,
"cli",
"direct",
)
if err != nil {
t.Fatalf("seed AddJob failed: %v", err)
}
stale := NewCronService(storePath, nil)
cli := NewCronService(storePath, nil)
newJob, err := cli.AddJob(
"new",
CronSchedule{Kind: "every", EveryMS: int64Ptr(120000)},
"world",
false,
"cli",
"direct",
)
if err != nil {
t.Fatalf("cli AddJob failed: %v", err)
}
updated := stale.EnableJob(existingJob.ID, false)
if updated == nil {
t.Fatalf("EnableJob(%q) returned nil", existingJob.ID)
}
jobs := NewCronService(storePath, nil).ListJobs(true)
if len(jobs) != 2 {
t.Fatalf("ListJobs() returned %d jobs, want 2", len(jobs))
}
if !jobExists(jobs, existingJob.ID) {
t.Fatalf("existing job %q was lost after save", existingJob.ID)
}
if !jobExists(jobs, newJob.ID) {
t.Fatalf("new job %q was lost after save", newJob.ID)
}
if jobEnabled(jobs, existingJob.ID) {
t.Fatalf("existing job %q should be disabled", existingJob.ID)
}
}
func TestListJobs_RefreshesStoreBeforeReading(t *testing.T) {
tmpDir := t.TempDir()
storePath := filepath.Join(tmpDir, "cron", "jobs.json")
seed := NewCronService(storePath, nil)
if _, err := seed.AddJob(
"existing",
CronSchedule{Kind: "every", EveryMS: int64Ptr(60000)},
"hello",
false,
"cli",
"direct",
); err != nil {
t.Fatalf("seed AddJob failed: %v", err)
}
stale := NewCronService(storePath, nil)
cli := NewCronService(storePath, nil)
newJob, err := cli.AddJob(
"new",
CronSchedule{Kind: "every", EveryMS: int64Ptr(120000)},
"world",
false,
"cli",
"direct",
)
if err != nil {
t.Fatalf("cli AddJob failed: %v", err)
}
jobs := stale.ListJobs(true)
if len(jobs) != 2 {
t.Fatalf("ListJobs() returned %d jobs, want 2", len(jobs))
}
if !jobExists(jobs, newJob.ID) {
t.Fatalf("ListJobs() did not include externally added job %q", newJob.ID)
}
}
func jobExists(jobs []CronJob, jobID string) bool {
for _, job := range jobs {
if job.ID == jobID {
return true
}
}
return false
}
func jobEnabled(jobs []CronJob, jobID string) bool {
for _, job := range jobs {
if job.ID == jobID {
return job.Enabled
}
}
return false
}
func int64Ptr(v int64) *int64 {
return &v
}