Optionally recover a recently-missed tick when an entry is re-registered late (env-gated, opt-in)

[s-kondamuri]

Problem

Entry.ScheduleFirst (called whenever an entry is (re-)registered while the
scheduler is already running, e.g. Cron.AddJob on a live Cron) computes:

func (e Entry) ScheduleFirst(now time.Time) time.Time {
	if !e.Prev.IsZero() {
		return e.Schedule.NextWithAfter(e.Prev, now)
	}
	...
}

NextWithAfter(t, after) starts its search from whichever of its two
arguments is later. Since now is virtually always later than Prev, the
search effectively always starts from now — so if the entry's natural next
tick (Schedule.Next(Prev)) already passed by the time it's (re-)registered,
it is silently skipped in favor of the following occurrence. No error, no
signal that anything was missed.

This is most likely to bite right after a process restart: a consumer's
in-memory "already registered" bookkeeping resets, so every entry has to be
freshly re-registered at once, and depending on how long that takes relative
to any given entry's next due tick, some entries can land in this exact gap
purely by timing.

We hit this concretely in segmentio/reverse-etl's scheduler-worker — a
customer's scheduled Reverse ETL sync silently didn't fire because its entry
was re-registered 34 seconds after its scheduled instant, following an
OOM-kill restart. Full writeup / evidence: SRC-4984.

Fix

Opt-in only, so existing consumers of this library are unaffected unless they
explicitly enable it:

• New env var RECOVERY_TIME_LIMIT (a time.ParseDuration string, e.g.
  20m). Unset/invalid → 0 → disabled, ScheduleFirst behaves exactly
  as before.
• When set and an entry's natural next tick (Schedule.Next(Prev)) is found
  to have already passed, but only within that limit, ScheduleFirst
  schedules it to fire ~30s from now instead of rolling forward silently to
  the next occurrence. Misses older than the limit are left untouched
  (rolled forward as before) — this intentionally never causes a "catch-up
  burst," it only recovers the single most recent miss.
• New Entry.RefID field + WithRefID(id string) EntryOption: an opaque,
  caller-defined identifier attached to an entry. This package never
  interprets it.
• New RecoveryObserver func(refID string, recovered bool, naturalNext, now time.Time)
  package var (nil/no-op by default). When RECOVERY_TIME_LIMIT is set,
  ScheduleFirst calls it once for every detected miss — recovered=true
  when it was fired immediately, recovered=false when it was too old and
  left to roll forward — passing along whatever RefID the caller attached
  via WithRefID. Lets a caller emit its own metrics/logs (e.g. a
  missed_tick_recovered/missed_tick_unrecovered counter tagged by
  refID) without this package needing to know what a "source" or
  "subscription" is. In reverse-etl's case RefID would be something like
  retl:<source_id>:<subscription_id>.

Tests

retl_recovery_test.go covers: recovers a recent miss, leaves an older miss
alone, confirms behavior is byte-for-byte unchanged when the env var is
unset (default), env var parsing, and RecoveryObserver firing correctly
(recovered / unrecovered / not-called-when-not-due), with RefID
passed through.

Compatibility

No breaking API changes — RefID/WithRefID/RecoveryObserver are
additive. No behavior change for any consumer that doesn't set
RECOVERY_TIME_LIMIT.

[s-kondamuri]

Flattened `ScheduleFirst`'s nested recovery check into a guard clause + a small `recoveryFireTime` helper (40c00d9) - pure refactor, no behavior change (verified branch-by-branch against the existing test suite, which passes unchanged).

🤖 Generated with Claude Code