From 1a7fc564fff2e564561890a21513df0ae74f635b Mon Sep 17 00:00:00 2001 From: Jose Cely Date: Tue, 23 Jun 2026 18:15:08 -0500 Subject: [PATCH] scheduler --- controllers/schedsample.go | 15 ++++++++------- register-scheduler.go | 2 +- run-auto-migrations.go | 1 + 3 files changed, 10 insertions(+), 8 deletions(-) diff --git a/controllers/schedsample.go b/controllers/schedsample.go index 1480a19..1837c3f 100644 --- a/controllers/schedsample.go +++ b/controllers/schedsample.go @@ -35,31 +35,32 @@ func SchedulerSample(c *core.Context) *core.Response { } c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued send_email task id=%d (maxRuns=1, thread=0)", item1.ID)) - // ── Enqueue a "flaky_task" (runs 3 times, sequential) ────────────────── + // ── Enqueue a "flaky_task" (runs 3 times, concurrent) ────────────────── flakyPayload, _ := json.Marshal(map[string]interface{}{ "description": "This task will fail and the error will be recorded", }) - item2, err := store.Enqueue(ctx, "flaky_task", string(flakyPayload), 0, 3, 0) + item2, err := store.Enqueue(ctx, "flaky_task", string(flakyPayload), 0, 3, 1) if err != nil { c.GetLogger().Error(fmt.Sprintf("failed to enqueue flaky_task: %v", err)) return c.Response.SetStatusCode(500).Json(`{"message": "failed to enqueue flaky_task"}`) } - c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued flaky_task id=%d (maxRuns=3, thread=0)", item2.ID)) + c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued flaky_task id=%d (maxRuns=3, thread=1/concurrent)", item2.ID)) - // ── Enqueue a high-priority "send_email" (repeats indefinitely, concurrent) ── + // ── Enqueue a high-priority "send_email" (repeats indefinitely, sequential) ── urgentPayload, _ := json.Marshal(map[string]interface{}{ "to": "urgent@example.com", "subject": "URGENT: Priority message", }) - item3, err := store.Enqueue(ctx, "send_email", string(urgentPayload), 10, -1, 1) + item3, err := store.Enqueue(ctx, "send_email", string(urgentPayload), 10, -1, 0) if err != nil { c.GetLogger().Error(fmt.Sprintf("failed to enqueue urgent send_email: %v", err)) return c.Response.SetStatusCode(500).Json(`{"message": "failed to enqueue urgent send_email"}`) } - c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued high-priority send_email id=%d (maxRuns=-1/infinite, thread=1/concurrent)", item3.ID)) + c.GetLogger().Info(fmt.Sprintf("[scheduler] enqueued high-priority send_email id=%d (maxRuns=-1/infinite, thread=0)", item3.ID)) - message := fmt.Sprintf(`{"message": "Enqueued 3 scheduler tasks (send_email x2, flaky_task x1)", "tasks": [%d, %d, %d]}`, + message := fmt.Sprintf(`{"message": "Enqueued 3 scheduler tasks (send_email, flaky_task)", "tasks": [%d, %d, %d]}`, item1.ID, item2.ID, item3.ID) + return c.Response.Json(message) } diff --git a/register-scheduler.go b/register-scheduler.go index 8067a76..9132e08 100644 --- a/register-scheduler.go +++ b/register-scheduler.go @@ -28,7 +28,7 @@ func handleSendEmail(ctx context.Context, item *scheduler.QueueItem) error { return fmt.Errorf("invalid send_email payload: %w", err) } log.Printf("[scheduler] [send_email] → to=%s subject=%q", payload.To, payload.Subject) - time.Sleep(50 * time.Millisecond) // simulate I/O + time.Sleep(2000 * time.Millisecond) // simulate I/O return nil } diff --git a/run-auto-migrations.go b/run-auto-migrations.go index 9a84d87..85250ce 100644 --- a/run-auto-migrations.go +++ b/run-auto-migrations.go @@ -30,6 +30,7 @@ func RunAutoMigrations() { // Simple scheduler (sche) tables — enabled when scheduler is active db.AutoMigrate(&scheduler.QueueItem{}) db.AutoMigrate(&scheduler.ProcessedItem{}) + db.AutoMigrate(&scheduler.SchedulerMeta{}) // End your auto migrations