Documentation
¶
Overview ¶
Package memschedulestore is the in-memory reference port.ScheduleStore (scheduled-tasks issue #189, Phase 1b): the offline, clock-injected single-process schedule registry the conformance suite validates and that composition can wire as the default-store opt-in. It keeps per-schedule Spec+State records and per-fire ScheduleFire records in mutex-guarded maps and reads `now` from an injected port.Clock, so tests advance a fake clock to express "the schedule is now due" / "the slot was missed" without real sleeps.
It is the reference for the SAME port.ScheduleStore contract a future JSONL store, a gRPC-driver client, and a k8s-backed store implement; the contract is the port, the in-memory map is one adapter. The conformance suite (engine/adapter/scheduleconformance) is the dual-path contract-unification: the Go port is the contract, the wire/disk is one adapter. This is the SECOND adapter-validation pattern after leases (leaseconformance).
PARSER-FREE: the store NEVER interprets a cron expression. Claim's nextFire is caller-computed (composition, which has the cronparse dependency). The store does NOT import engine/adapter/cronparse — that is a deliberate layering choice: the store is the durable ground truth, the parser is a composition-time helper, and a store backend (e.g. a remote SQL table) need not ship a cron parser to satisfy the port.
MISFIRE-FREE: the store does NOT read ScheduleSpec.Misfire. Due returns any schedule whose NextFireAt <= now (plus Enabled + MaxFires); the misfire policy (fire-once-now vs skip) is a COMPOSITION concern applied at tick time.
Index ¶
- Variables
- type Store
- func (s *Store) Claim(_ context.Context, name string, now, nextFire time.Time) (port.Schedule, error)
- func (s *Store) ClaimNow(_ context.Context, name string, now, nextFire time.Time) (port.Schedule, error)
- func (s *Store) Create(_ context.Context, in port.Schedule) error
- func (s *Store) Delete(_ context.Context, name string) error
- func (s *Store) Due(_ context.Context, now time.Time) ([]port.Schedule, error)
- func (s *Store) List(_ context.Context) ([]port.Schedule, error)
- func (s *Store) ListFires(_ context.Context, scheduleName string) ([]port.ScheduleFire, error)
- func (s *Store) Load(_ context.Context, name string) (port.Schedule, error)
- func (s *Store) LoadFire(_ context.Context, fireID string) (port.ScheduleFire, error)
- func (s *Store) ReArmOneShot(_ context.Context, name string, nextFire time.Time) error
- func (s *Store) RecordFire(_ context.Context, f port.ScheduleFire) error
- func (s *Store) RecordFireProgress(_ context.Context, name string, fireID string, at time.Time) error
- func (s *Store) RecordFireStart(_ context.Context, name string, fire port.ScheduleFire) error
- func (s *Store) Save(_ context.Context, in port.Schedule) error
- func (s *Store) SetEnabled(_ context.Context, name string, enabled bool) error
Constants ¶
This section is empty.
Variables ¶
var ErrNotFound = fmt.Errorf("memschedulestore: schedule not found: %w", port.ErrScheduleNotFound)
ErrNotFound is returned by Load/Delete/Claim/RecordFire/LoadFire when no schedule (or fire) exists under the requested name/id. It wraps port.ErrScheduleNotFound so a consumer that may not import this adapter (e.g. engine/agent) can distinguish not-found from an infra failure via errors.Is, the same discipline memstore applies for ErrSessionNotFound.
Functions ¶
This section is empty.
Types ¶
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is a concurrency-safe in-memory ScheduleStore. The Claim path is the at-most-once atomic advance: under the mutex it re-checks the slot is still due (a peer's Claim may have advanced it between this caller's Due and Claim) and, only if still due, advances NextFireAt + LastFireAt, bumps FireCount, and stamps port.PendingFireSessionID on LastFireSessionID — all before the fire runs (claim-before-fire).
func New ¶
func New() *Store
New constructs an in-memory ScheduleStore. Every port.ScheduleStore method takes `now` as an explicit argument, so the store needs no injected clock of its own — time is caller-supplied. (An earlier draft injected a port.Clock reserved for a future self-pushing tick helper, but it was never read; it was dropped rather than carried as dead constructor surface. If a lookahead phase ever needs an internal `now`, re-add the clock then — YAGNI until it exists.)
func (*Store) Claim ¶
func (s *Store) Claim(_ context.Context, name string, now, nextFire time.Time) (port.Schedule, error)
Claim is the AT-MOST-ONCE atomic advance. Under the mutex it re-checks the slot is still due (NextFireAt <= now && Enabled && under MaxFires); if NOT (a peer's Claim already advanced it, or it was disabled, or it exhausted) it returns ErrScheduleNotFound — the slot is gone, the fail-safe interpretation the conformance suite pins (a second Claim at the same now does NOT re-claim). If still due, it atomically: sets LastFireAt=now, advances NextFireAt to nextFire, increments FireCount, stamps LastFireSessionID=port.PendingFireSessionID, and (for a zero nextFire: a one-shot or an exhausted cron) sets Enabled=false. It returns the claimed Schedule (with the advanced State).
func (*Store) ClaimNow ¶
func (s *Store) ClaimNow(_ context.Context, name string, now, nextFire time.Time) (port.Schedule, error)
ClaimNow is the manual-trigger variant of Claim (the FireNow primitive). It performs the SAME atomic advance as Claim but does NOT enforce the NextFireAt <= now due-check — it claims the slot regardless of whether it is due (a manual fire bypasses the cadence but still claims atomically for at-most-once). The Enabled + MaxFires checks STILL apply. The at-most-once fence is LastFireAt == now: a second ClaimNow at the same now (or a ClaimNow racing a tick-loop Claim at the same now) is rejected — the advance already happened. See port.ScheduleStore.ClaimNow for the crash-recoverability rationale.
func (*Store) Create ¶
Create atomically creates a NEW schedule under in.Spec.Name (review finding 5, issue #368): under the SAME mutex Save/Load/Claim already serialize on, it checks for an existing record and creates the new one in one lock acquisition — unlike the manager's prior two-call Load-then-Save, which raced across two separate acquisitions. A name already in use returns ErrScheduleAlreadyExists (wrapped) and leaves the existing record untouched.
func (*Store) Delete ¶
Delete removes the schedule stored under name. It is IDEMPOTENT: deleting an unknown name is success (the PrunableStore.Delete discipline).
func (*Store) Due ¶
Due returns the schedules whose NextFireAt <= now AND Enabled AND (when MaxFires > 0) FireCount < MaxFires. It is idempotent and side-effect-free; it does not advance state. The store does NOT read ScheduleSpec.Misfire — the misfire policy is a composition concern.
func (*Store) ListFires ¶
ListFires returns the fire records for a schedule, in no guaranteed order. The not-found case for the SCHEDULE wraps ErrScheduleNotFound; an empty fire list for an existing schedule is a successful empty slice (not an error).
func (*Store) Load ¶
Load returns the schedule stored under name. The not-found case wraps ErrScheduleNotFound. The returned Schedule is a DEEP COPY (the State is a struct, the Parts slice is copied) so a caller cannot mutate the store's record through the returned reference — the memstore precedent.
func (*Store) LoadFire ¶
LoadFire returns the fire record stored under fireID. The not-found case wraps ErrScheduleNotFound.
func (*Store) ReArmOneShot ¶
ReArmOneShot is the at-least-once re-arm primitive for a one-shot schedule (ADR 0059 Phase 2). It atomically: re-enables the schedule (Enabled=true), sets NextFireAt to nextFire, and increments OneShotRetryCount. The atomicity (the single mutex) is the re-arm fence: two concurrent re-arms cannot double-increment the counter or double-enable. The not-found case wraps ErrScheduleNotFound. The retry-budget gate (OneShotRetryCount < OneShotMaxRetries) is the CALLER's responsibility — the store does NOT enforce the budget, it only atomically advances the counter. One-shot-only; the caller never calls this on a cron schedule.
func (*Store) RecordFire ¶
RecordFire records the outcome of a fire (f) and updates the schedule's LastFireSessionID to f.SessionID (overwriting the port.PendingFireSessionID value Claim set). It is IDEMPOTENT per fire id: recording the same f.ID twice is a no-op (the second call returns nil without mutating state). The not-found case (the schedule was deleted between Claim and RecordFire) wraps ErrScheduleNotFound.
It FLIPS the fire terminal and clears the in-flight ScheduleState fields (LastFireStartedAt/LastFireProgressAt/FireDeadline) — a recorded (terminal) fire has no in-flight run (issue #386). It overwrites any in-flight fire record the same f.ID had under RecordFireStart with the terminal one.
func (*Store) RecordFireProgress ¶
func (s *Store) RecordFireProgress(_ context.Context, name string, fireID string, at time.Time) error
RecordFireProgress advances the in-flight fire's last-observed-progress instant (issue #386). It updates LastFireProgressAt on the state and ProgressAt on the in-flight fire record (when `at` is after the stored value — an earlier `at` is ignored so a reordered update cannot rewind progress). It is best-effort/idempotent: a missing in-flight fire record records on the state alone; a not-found schedule wraps ErrScheduleNotFound; a terminal fire is untouched. fireID targets the single in-flight fire record by its known id directly (review finding M1 — no scan); a terminal fire record is untouched (review finding M2 — never revert terminal → in-flight).
func (*Store) RecordFireStart ¶
RecordFireStart persists the IN-FLIGHT fire (issue #386): the fire's run has begun but not yet produced a terminal outcome. It writes the fire record (in-flight: Stop empty, StartedAt set) and stamps the schedule's LastFireSessionID to the REAL session id (overwriting the pending sentinel Claim set) + LastFireStartedAt (and seeds LastFireProgressAt to StartedAt when the caller passed a zero ProgressAt) + FireDeadline. It is IDEMPOTENT per fire id: a re-record of the same in-flight fire (same StartedAt) is a no-op for the in-flight record; a re-record for a fire id that is ALREADY terminal is a no-op (a terminal fire is not re-opened). The not-found case wraps ErrScheduleNotFound.
func (*Store) Save ¶
Save upserts the schedule by Spec.Name. A schedule with the same name is overwritten on the Spec half; the State half is PRESERVED on overwrite (a Save with a fresh zero State does not reset firing progress — call Delete + Save to reset, the port doc says so). A NEW schedule is initialised with Enabled=true (a new schedule is active by default; pause it by overwriting State.Enabled=false, which Save preserves on re-save).
func (*Store) SetEnabled ¶
SetEnabled atomically sets the schedule's Enabled flag WITHOUT touching any other State field (unlike Save, which preserves the State half on a Spec overwrite and so cannot mutate Enabled). It is the pause/resume primitive. The not-found case wraps ErrScheduleNotFound.
Source Files