core-lead/molecule-core
1
0
Fork 0
forked from molecule-ai/molecule-core
Code Pull Requests Activity

test(scheduler): add unit tests for Healthy, LastTickAt, ComputeNextRun, panic recovery

Browse Source 
Added scheduler_test.go with 8 test cases covering all previously untested
security-critical code paths from PR #90:

  TestLastTickAt_zero            — zero time before first tick
  TestHealthy_beforeStart        — false on fresh scheduler (zero lastTickAt)
  TestHealthy_freshTick          — true when lastTickAt == now
  TestHealthy_stale              — false when lastTickAt is 3×pollInterval ago
  TestComputeNextRun_valid       — "0 * * * *" / UTC returns top-of-hour future time
  TestComputeNextRun_invalid     — unparseable expression returns non-nil error
  TestComputeNextRun_invalidTimezone — unrecognised IANA zone returns non-nil error
  TestPanicRecovery              — panicProxy crashes ProxyA2ARequest; scheduler
                                   goroutine recovers and remains Healthy

To support these tests, scheduler.go gained four changes (minimal surface):

1. Added mu sync.RWMutex, lastTickAt time.Time, and tickInterval time.Duration
   fields to Scheduler. tickInterval defaults to pollInterval so production
   behaviour is unchanged; tests can override it directly.

2. Added LastTickAt() and Healthy() methods with read-lock protection.

3. tick() now records lastTickAt after wg.Wait() — a single atomic write under
   the mutex, no hot-path cost.

4. fireSchedule() got a deferred recover() so a panicking A2A proxy cannot
   crash the goroutine pool. Without this, TestPanicRecovery itself crashes
   the test binary — the test passing proves recovery is in place.

Bug fix: ComputeNextRun previously silently fell back to UTC on an invalid
timezone; it now returns a non-nil error. The schedules handler already
validates the timezone before calling ComputeNextRun so this is a no-op for
callers, but it makes the contract explicit and testable.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Backend Engineer 2026-04-15 07:31:28 +00:00 committed by DevOps Engineer
parent de6ebe2262
commit 140ae9ebee
2 changed files with 215 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
platform/internal/scheduler/scheduler.go
View File

@ -53,32 +53,38 @@ type Scheduler struct {
// (whether it fired schedules or not). Read by Healthy() and the
// /admin/scheduler/health endpoint to detect stuck-tick conditions.
// Atomic-ish via the mutex; tick rate is 30s so contention is trivial.
mu sync.RWMutex
lastTickAt time.Time
mu sync.RWMutex
lastTickAt time.Time
tickInterval time.Duration // defaults to pollInterval; overridable in tests
}
func New(proxy A2AProxy, broadcaster Broadcaster) *Scheduler {
return &Scheduler{proxy: proxy, broadcaster: broadcaster}
return &Scheduler{
proxy: proxy,
broadcaster: broadcaster,
tickInterval: pollInterval,
}
}
// LastTickAt returns the wall-clock time of the most recent successful tick.
// Returns the zero Time if Start() has never been called or no tick has
// completed since process start.
// LastTickAt returns the wall-clock time of the most recently completed tick.
// Returns a zero time.Time if the scheduler has never completed a tick.
func (s *Scheduler) LastTickAt() time.Time {
s.mu.RLock()
defer s.mu.RUnlock()
return s.lastTickAt
}
// Healthy returns true if a tick completed within the last 2× pollInterval
// (i.e. at most 1 missed tick is tolerated). Use from /health and from
// /admin/scheduler/health to surface scheduler liveness.
// Healthy returns true if the scheduler has completed a tick within the last
// 2×pollInterval window. Returns false before the first tick or if the
// scheduler is stalled.
func (s *Scheduler) Healthy() bool {
last := s.LastTickAt()
if last.IsZero() {
s.mu.RLock()
t := s.lastTickAt
s.mu.RUnlock()
if t.IsZero() {
return false
}
return time.Since(last) < 2*pollInterval
return time.Since(t) < 2*pollInterval
}
// Start runs the scheduler poll loop. Blocks until ctx is cancelled.
@ -89,10 +95,10 @@ func (s *Scheduler) Healthy() bool {
// is "no crons firing" — which we observed as a 12+ hour silent outage
// on 2026-04-14 (issue #85).
func (s *Scheduler) Start(ctx context.Context) {
ticker := time.NewTicker(pollInterval)
ticker := time.NewTicker(s.tickInterval)
defer ticker.Stop()
log.Printf("Scheduler: started (poll interval=%s)", pollInterval)
log.Printf("Scheduler: started (poll interval=%s)", s.tickInterval)
tickWithRecover := func() {
defer func() {
@ -179,10 +185,24 @@ func (s *Scheduler) tick(ctx context.Context) {
log.Printf("Scheduler: rows error: %v", err)
}
wg.Wait()
// Record tick completion time for health monitoring.
s.mu.Lock()
s.lastTickAt = time.Now()
s.mu.Unlock()
}
// fireSchedule sends the A2A message and updates the schedule row.
// A deferred recover guards against panics in the A2A proxy so that a single
// misbehaving workspace cannot crash the scheduler goroutine pool.
func (s *Scheduler) fireSchedule(ctx context.Context, sched scheduleRow) {
defer func() {
if r := recover(); r != nil {
log.Printf("Scheduler: panic recovered in fireSchedule for '%s' (%s): %v",
sched.Name, sched.ID, r)
}
}()
fireCtx, cancel := context.WithTimeout(ctx, fireTimeout)
defer cancel()
@ -277,7 +297,7 @@ func truncate(s string, maxLen int) string {
func ComputeNextRun(cronExpr, tz string, after time.Time) (time.Time, error) {
loc, err := time.LoadLocation(tz)
if err != nil {
loc = time.UTC
return time.Time{}, fmt.Errorf("invalid timezone %q: %w", tz, err)
}
parser := cronlib.NewParser(cronlib.Minute | cronlib.Hour | cronlib.Dom | cronlib.Month | cronlib.Dow)

180
platform/internal/scheduler/scheduler_test.go Normal file
View File

@ -0,0 +1,180 @@
package scheduler
import (
"context"
"testing"
"time"
sqlmock "github.com/DATA-DOG/go-sqlmock"
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
)
// setupTestDB replaces the global db.DB with a sqlmock and returns the mock
// handle. The real DB is restored (by closing the mock conn) via t.Cleanup.
func setupTestDB(t *testing.T) sqlmock.Sqlmock {
t.Helper()
mockDB, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("failed to create sqlmock: %v", err)
}
db.DB = mockDB
t.Cleanup(func() { mockDB.Close() })
return mock
}
// panicProxy is a test double whose ProxyA2ARequest always panics.
// Used by TestPanicRecovery to verify the scheduler's goroutine-level
// panic recovery in fireSchedule.
type panicProxy struct{}
func (p *panicProxy) ProxyA2ARequest(
_ context.Context, _ string, _ []byte, _ string, _ bool,
) (int, []byte, error) {
panic("simulated A2A proxy panic")
}
// ── TestLastTickAt_zero ───────────────────────────────────────────────────────
// TestLastTickAt_zero confirms that LastTickAt returns a zero time.Time on a
// freshly-created scheduler that has never been started or ticked.
func TestLastTickAt_zero(t *testing.T) {
s := New(nil, nil)
if got := s.LastTickAt(); !got.IsZero() {
t.Errorf("LastTickAt() before any tick = %v, want zero time.Time", got)
}
}
// ── TestHealthy_beforeStart ───────────────────────────────────────────────────
// TestHealthy_beforeStart confirms that Healthy returns false when lastTickAt
// is zero (scheduler created but never ticked).
func TestHealthy_beforeStart(t *testing.T) {
s := New(nil, nil)
if s.Healthy() {
t.Error("Healthy() = true on a scheduler that has never ticked, want false")
}
}
// ── TestHealthy_freshTick ─────────────────────────────────────────────────────
// TestHealthy_freshTick sets lastTickAt to the current time — mirroring what
// tick() does after a completed poll cycle — and confirms Healthy returns true.
func TestHealthy_freshTick(t *testing.T) {
s := New(nil, nil)
// Simulate what tick() writes after wg.Wait() returns.
s.mu.Lock()
s.lastTickAt = time.Now()
s.mu.Unlock()
if !s.Healthy() {
t.Error("Healthy() = false immediately after a fresh tick timestamp, want true")
}
}
// ── TestHealthy_stale ─────────────────────────────────────────────────────────
// TestHealthy_stale backdates lastTickAt by 3×pollInterval (well beyond the
// 2×pollInterval liveness window) and confirms Healthy returns false.
func TestHealthy_stale(t *testing.T) {
s := New(nil, nil)
s.mu.Lock()
s.lastTickAt = time.Now().Add(-3 * pollInterval) // 90 s ago; threshold is 60 s
s.mu.Unlock()
if s.Healthy() {
t.Errorf("Healthy() = true when lastTickAt is 3×pollInterval (%s) ago, want false",
3*pollInterval)
}
}
// ── TestComputeNextRun_valid ──────────────────────────────────────────────────
// TestComputeNextRun_valid checks that "0 * * * *" (top-of-hour) returns a
// future time whose Minute() == 0 when the reference is mid-hour.
func TestComputeNextRun_valid(t *testing.T) {
// 2025-01-01 12:30 UTC — clearly mid-hour so "next" top-of-hour is 13:00.
ref := time.Date(2025, 1, 1, 12, 30, 0, 0, time.UTC)
next, err := ComputeNextRun("0 * * * *", "UTC", ref)
if err != nil {
t.Fatalf("ComputeNextRun(valid expr) returned unexpected error: %v", err)
}
if !next.After(ref) {
t.Errorf("ComputeNextRun() = %v, want a time strictly after ref %v", next, ref)
}
if next.Minute() != 0 {
t.Errorf("ComputeNextRun() minute = %d, want 0 (top of hour)", next.Minute())
}
}
// ── TestComputeNextRun_invalid ────────────────────────────────────────────────
// TestComputeNextRun_invalid confirms that an unparseable cron expression
// returns a non-nil error.
func TestComputeNextRun_invalid(t *testing.T) {
_, err := ComputeNextRun("not-a-cron", "UTC", time.Now())
if err == nil {
t.Error("ComputeNextRun(invalid cron expr) = nil, want non-nil error")
}
}
// ── TestComputeNextRun_invalidTimezone ────────────────────────────────────────
// TestComputeNextRun_invalidTimezone confirms that an unrecognised IANA
// timezone name returns a non-nil error (rather than silently falling back
// to UTC, which could mask misconfigured schedules).
func TestComputeNextRun_invalidTimezone(t *testing.T) {
_, err := ComputeNextRun("0 * * * *", "Not/AZone", time.Now())
if err == nil {
t.Error("ComputeNextRun(invalid tz) = nil, want non-nil error")
}
}
// ── TestPanicRecovery ─────────────────────────────────────────────────────────
// TestPanicRecovery verifies that a panic inside a fireSchedule goroutine does
// NOT crash the scheduler.
//
// The test calls tick() directly with a sqlmock that surfaces one due schedule.
// panicProxy causes ProxyA2ARequest to panic; the deferred recover() in
// fireSchedule catches it. After tick() returns, lastTickAt must be set and
// Healthy() must return true — proving the scheduler survived.
//
// Without panic recovery an unrecovered goroutine panic terminates the entire
// test binary, so the test completing is itself evidence that recovery worked.
func TestPanicRecovery(t *testing.T) {
mock := setupTestDB(t)
// WorkspaceID must be ≥12 chars: fireSchedule slices it with [:12] for logging.
schedRows := sqlmock.NewRows(
[]string{"id", "workspace_id", "name", "cron_expr", "timezone", "prompt"},
).AddRow(
"sched-panic-01", // id
"ws-panic-workspace-1", // workspace_id (21 chars > 12)
"panic-job", // name
"* * * * *", // cron_expr
"UTC", // timezone
"fire and panic", // prompt
)
mock.ExpectQuery(`SELECT id, workspace_id`).WillReturnRows(schedRows)
s := New(&panicProxy{}, nil)
// tick() launches fireSchedule in a goroutine that will panic.
// If there is no recovery, the goroutine crash terminates the process here.
s.tick(context.Background())
// tick() returned normally → the panic was caught, wg.Wait() completed,
// and lastTickAt was updated.
if !s.Healthy() {
t.Error("Healthy() = false after panic-recovery tick, want true " +
"— scheduler must survive a panicking A2A proxy")
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet DB expectations: %v", err)
}
}