From 746f32822c500d3001ce9531e7f57f3548a74282 Mon Sep 17 00:00:00 2001 From: Tequila Sunset Date: Sat, 8 Aug 2026 18:57:12 +0800 Subject: [PATCH 1/5] Document reminder timing design --- design/persistence.md | 30 +++++++++++++- design/timers.md | 92 ++++++++++++++++++++++++++++++++++++------- 2 files changed, 105 insertions(+), 17 deletions(-) diff --git a/design/persistence.md b/design/persistence.md index e8aeb62..367ef07 100644 --- a/design/persistence.md +++ b/design/persistence.md @@ -309,7 +309,33 @@ The state-write baseline is recorded at each tier, because a single number would ## The Reminder table ``` -reminder(grain_type, grain_key, name, method, due_at, interval, etag) +reminder(grain_type, grain_key, name, method, first_tick_time, due_at, interval, etag) ``` -This table does not go through the `Store` interface — scanning due rows, CAS claiming, and row deletion do not fit into "read and write one state per GrainId". It has its own interface; details in [timers.md](timers.md). +The table above is the logical Reminder schema. SQLite may retain the existing +physical `schedule` table when that preserves a non-destructive migration. The +contract is the `first_tick_time` field, the old-row fallback, and the tests. + +`first_tick_time` stores the first due time for the current Reminder setting. +`due_at` stores the next due time. `interval` stores the period. A zero +`interval` means one-shot. + +This table does not go through the `Store` interface — scanning due rows, CAS +claiming, and row deletion do not fit into "read and write one state per +GrainId". It has its own interface, named `ReminderStore` in the public API; +details are in [timers.md](timers.md). + +### SQLite Reminder migration + +When SQLite opens an existing coordination database, it must add the +`first_tick_time INTEGER` column to the existing coordination table during +open. The migration must also handle old rows that have no value in this +column. For each such row, it must use the row's current `due_at` as +`first_tick_time`. + +This fallback keeps old rows readable, but it cannot recover the original first +time. The original first time is not reconstructible from an old row. The +migration must document and preserve that limit. + +The migration must be covered by a storage test that opens an old database, +checks the fallback value, and checks that a new open keeps the value. diff --git a/design/timers.md b/design/timers.md index 9835461..820e678 100644 --- a/design/timers.md +++ b/design/timers.md @@ -33,6 +33,13 @@ func (a *account) ApplyInterest(ctx context.Context, tick gor.TickStatus) error ```go type Reminder[T any] struct { /* bound to one GrainId */ } +func NewReminder[T any](b *Binder) Reminder[T] + +type ReminderTime struct { /* first delay and period */ } + +func After(delay time.Duration) ReminderTime +func Every(period time.Duration) ReminderTime + type TickStatus struct { FirstTickTime time.Time Period time.Duration @@ -46,6 +53,11 @@ func Handle[T any](m func(T, context.Context, TickStatus) error) MethodHandle[T] func (s Reminder[T]) Set(ctx context.Context, name string, when ReminderTime, m MethodHandle[T]) error ``` +The public Reminder API is `Reminder[T]`, `ReminderTime`, `NewReminder`, +`ReminderStore`, `ReminderInvocation`, and `TickStatus`. `Handle` accepts +`func(T, context.Context, TickStatus) error`. `After` creates a one-shot +Reminder. `Every` creates a periodic Reminder. + `Account.ApplyInterest` is a Go method expression. The compiler checks that `ApplyInterest` is a method of `Account` and that its signature is `func(Account, context.Context, TickStatus) error`. A typo, rename, or @@ -57,6 +69,13 @@ The name the table stores is read off the method expression once, when `Handle` The called method must be in the Grain's interface. The generated dispatch table must accept the `TickStatus` value at delivery time. +The generator must also expose a typed `newReminderCall(method, TickStatus)` +factory for each Grain interface. The factory creates the normal typed request +and reply values for the Reminder method. The timer passes those values to +`Runtime.Invoke`, so local and forwarded Calls use the same path. The timer +must not use reflection. The public API must not expose a string dispatch path; +the stored method name is only an internal identifier used by generated code. + **Why a method expression, not a generated handle value.** Reminders are set from inside Grain methods, which live in the Grain package. The Grain package cannot import the package generated from its own interfaces: that package imports the Grain package for the interface types used in its proxies and dispatch, so the import is a cycle — the same reason generated artifacts land in their own package that the Grain package does not import (see [codegen.md](codegen.md)). A per-method handle symbol emitted by the generator therefore cannot be named from the code that sets a reminder. The method expression is the only compile-time-checked way to name a method from that code using just the `gor` package and the interface declared in the Grain package, so the handle carries no generated symbol. The generator changes nothing for this. "One unified entry point" is rejected. Orleans has Grains implement `ReceiveReminder(name)` and switch on the name themselves — that is bringing back the hand-written dispatcher deleted at [step 3](../ROADMAP.md#3-typed-proxy-code-generation), and in user code of all places. gor's selling point is compile-time typing; it must not open a string-dispatch loophole here. @@ -96,34 +115,52 @@ gor.After(d) // one-shot, fires once after d gor.Every(d) // periodic, fires every d ``` -`Set` overwrites by name. One Grain has only one Reminder with a given name. -Setting it again changes its time. `Cancel(ctx, name)` deletes it. +`Set` replaces the row with the same name. One Grain has only one Reminder +with a given name. The replacement resets `FirstTickTime` to the new first due +time. It also resets `DueAt` to that same first due time. `Cancel(ctx, name)` +deletes the Reminder. -**Missed windows are not made up.** If the process is down for three periods, it fires once on return and then tracks to the next future time. Making up three firings is a trap — what users want is almost never "run everything that piled up", and how much piles up depends on the downtime, making the behavior unpredictable. If catch-up is truly wanted, users compute it in the method from the last execution time. +A one-shot Reminder uses `Period = 0` in its `TickStatus`. A periodic +Reminder keeps its `FirstTickTime` when the poller claims it. The claim reports +the claimed old `DueAt` as `CurrentTickTime`. The poller computes the next +`DueAt` strictly in the future. It does not catch up missed periods. + +**Missed windows are not made up.** If the process is down for three periods, +it fires once on return and then tracks to the next future time. Making up +three firings is a trap — what users want is almost never "run everything that +piled up", and how much piles up depends on the downtime, making the behavior +unpredictable. If catch-up is truly wanted, users compute it in the method +from the last execution time. **Precision is the polling interval.** Persisted Reminders should never promise milliseconds. ## The table ``` -reminder(grain_type, grain_key, name, method, due_at, interval, etag) +reminder(grain_type, grain_key, name, method, first_tick_time, due_at, interval, etag) ``` -A zero `interval` means one-shot. +`first_tick_time` stores the first due time for the current setting. `due_at` +stores the next due time. A zero `interval` means one-shot and maps to +`Period = 0` in `TickStatus`. The primary key is (GrainType, GrainKey, name). `name` identifies the -Reminder. `method` is the method to call when due. One method can back +Reminder. `method` is the internal method identifier. One method can back several Reminders with different periods. -## The table's interface +## ReminderStore -Four operations, aligned with what the poller and the user each need to do: +`ReminderStore` has four operations, aligned with what the poller and the user +need to do: - **List due** — rows with `due_at <= now`; `now` comes in as a parameter. - **Claim one row** — CAS with the row's etag, pushing `due_at` to the given next time; a zero time deletes the row. Exactly one claiming node wins. - **Write one row** — unconditional overwrite; the user's `Set` goes here. - **Delete one row** — unconditional; the user's `Cancel` goes here. +`ReminderStore` persists `FirstTickTime` with each row and returns it to the +poller when it lists a due row. + **The etag exists only for claiming.** The user's `Set` / `Cancel` carries no etag: the user does not have one anyway, and an explicit reschedule or cancel is his to win. The claim that got overwritten simply delivered one fewer time; at-most-once still holds. **The next due time is computed by the poller, not the table.** "No catch-up for missed" is policy; the table is only responsible for getting the CAS right. One-shot Reminders use the zero time for "no next" — the same convention as a zero `interval`. @@ -136,15 +173,24 @@ The poller scans rows with `due_at <= now` and, for each row: 1. **Claim** — CAS to push `due_at` to the next period (delete the row for one-shot Reminders). -2. Deliver the call only after winning the claim. +2. Build the typed request with the generated `newReminderCall` factory. +3. Deliver the ordinary Call through `Runtime.Invoke`, only after winning the + claim. -**Push to the first time still in the future**, not `due_at + interval`. After three periods of downtime, adding one interval still lands in the past; the next scan hits the same row again, and "no catch-up" becomes catch-up. +The periodic claim keeps `FirstTickTime` and reports the claimed old `DueAt` as +`CurrentTickTime`. It pushes `due_at` to the first time strictly in the future, +not to `due_at + interval`. After three periods of downtime, adding one +interval still lands in the past; the next scan hits the same row again, and +"no catch-up" becomes catch-up. -The reverse order causes repeated firing on a crash. Crash after the claim but before delivery misses one firing — a deliberate trade-off: +The reverse order causes repeated firing on a crash. A failure after the claim +can miss one delivery. This is a deliberate trade-off: **gor promises at-most-once delivery, not exactly-once execution.** -Delivery failures are not retried. Only the user knows whether retrying is safe; the runtime does not decide for him — the same stance as on `State.Set()` conflicts. +Delivery failures are not retried. Only the user knows whether retrying is safe; +the runtime does not decide for the user — the same stance as on `State.Set()` +conflicts. **But no retry does not mean silence.** A Reminder delivery has no caller waiting; the error returned by the method is sent by the runtime to the configured error sink, and dropped only when no sink is configured. The runtime does not retry for the user; whether to alert remains the user's decision. @@ -255,9 +301,25 @@ The step-4 skeleton must hold this one: Crashes, claim failures, two pollers scanning at the same time — none of these may break it. +## Minimum tests + +The minimum failure, restart, and claim tests must cover these cases: + +- A due Reminder is found and delivered after a process restart. +- Two pollers claim the same row at the same time; one claim wins. +- A failed claim does not deliver the Reminder and leaves the row available. +- A process failure after a successful claim and before delivery may miss one delivery. +- A failed Reminder method reaches `OnError`; the runtime does not retry it. +- `Set` replaces by name, resets `FirstTickTime`, and resets the first due time. +- A periodic Reminder after downtime reports the old due time, keeps `FirstTickTime`, and does not replay missed periods. +- A one-shot Reminder reports `Period = 0`. + ## Gap The typed Reminder method handle is implemented in the current code. The -public naming migration to `Reminder` and `NewReminder` remains part of the -0.1.0 API work. The method name is read from the expression once. The table, -poller, and restart recovery use the method-name string. +public naming migration to `Reminder`, `ReminderTime`, `NewReminder`, and +`ReminderStore` remains part of the 0.1.0 API work. The `first_tick_time` row +field and the generated typed `newReminderCall` factory are also part of that +work. This design batch does not rename the Go implementation. The method +name is read from the expression once. The table, poller, and restart recovery +use the method-name string as an internal identifier. From 7e7e216cc971aba6b719db236033c4e39b51b7b4 Mon Sep 17 00:00:00 2001 From: Tequila Sunset Date: Sat, 8 Aug 2026 19:26:12 +0800 Subject: [PATCH 2/5] Implement reminders and typed reminder calls --- background_error_test.go | 63 ++++++---- benchmark_test.go | 4 +- cluster_runtime_test.go | 2 +- .../testfixture/endtoend/domain/domain.go | 5 + .../testfixture/endtoend/gorgen/generated.go | 28 ++++- .../endtoend/gorgen/runtime_test.go | 21 ++++ cycle_test.go | 2 +- examples/shadow/cmd/load/main.go | 2 +- examples/shadow/cmd/load/main_test.go | 2 +- examples/shadow/cmd/shadow/main_test.go | 4 +- examples/shadow/domain/domain.go | 8 +- examples/shadow/domain/gorgen/generated.go | 31 ++++- examples/shadow/http_test.go | 6 +- examples/shadow/runtime.go | 2 +- examples/shadow/shadow_test.go | 22 ++-- forward_teardown_test.go | 2 +- forward_test.go | 4 +- gor.go | 112 +++++++++-------- gor_test.go | 6 +- internal/codegen/load.go | 26 +++- internal/codegen/model.go | 7 +- internal/codegen/render.go | 13 +- .../testfixture/generated/generated.go | 18 ++- method_handle_test.go | 12 +- observability_test.go | 8 +- schedule.go | 114 ++++++++--------- schedule_test.go | 95 ++++++++------ sim/cluster.go | 2 +- sim/cluster_test.go | 2 +- sim/cycle_test.go | 2 +- sim/fault_test.go | 2 +- sim/log.go | 2 +- sim/network.go | 14 +-- sim/replay_test.go | 2 +- sim/sim.go | 29 +++-- sim/store.go | 36 +++--- state.go | 2 +- state_test.go | 4 +- store/durability_test.go | 2 +- store/migration_test.go | 27 ++-- store/schedule.go | 116 +++++++++--------- store/schedule_sqlite.go | 85 +++++++------ store/schedule_test.go | 75 ++++++++--- store/sqlite.go | 45 ++++++- store/store.go | 12 +- store/store_test.go | 4 +- timer/timer.go | 47 ++++--- timer/timer_test.go | 113 ++++++++++++++--- 48 files changed, 790 insertions(+), 452 deletions(-) diff --git a/background_error_test.go b/background_error_test.go index 05c3f26..fc5fb3f 100644 --- a/background_error_test.go +++ b/background_error_test.go @@ -13,23 +13,23 @@ import ( "github.com/suraciii/gor/store" ) -// failingScheduleStore fails schedule listing or claiming on demand. It exists +// failingReminderStore fails schedule listing or claiming on demand. It exists // to pin the boundary of the background error exit: list and claim failures // are scheduler state, not application callback failures. -type failingScheduleStore struct { +type failingReminderStore struct { *store.Memory failList atomic.Bool failClaim atomic.Bool } -func (s *failingScheduleStore) ListDue(ctx context.Context, now time.Time) ([]store.Schedule, error) { +func (s *failingReminderStore) ListDue(ctx context.Context, now time.Time) ([]store.Reminder, error) { if s.failList.Load() { return nil, errors.New("simulated list failure") } return s.Memory.ListDue(ctx, now) } -func (s *failingScheduleStore) Claim(ctx context.Context, schedule store.Schedule, nextDueAt time.Time) (bool, error) { +func (s *failingReminderStore) Claim(ctx context.Context, schedule store.Reminder, nextDueAt time.Time) (bool, error) { if s.failClaim.Load() { return false, errors.New("simulated claim failure") } @@ -39,19 +39,21 @@ func (s *failingScheduleStore) Claim(ctx context.Context, schedule store.Schedul // misnamedSchedule is an entity whose interface contains a method literally // named OnDeactivate. Its signature differs from the Deactivatable hook, so it // does not implement Deactivatable; a scheduled failure of this method must -// still be reported as a ScheduledInvocation, not as a Deactivation. +// still be reported as a ReminderInvocation, not as a Deactivation. type misnamedSchedule interface { Arm(context.Context) error - OnDeactivate(context.Context) error + OnDeactivate(context.Context, TickStatus) error } type misnamedScheduleArmRequest struct{} type misnamedScheduleArmReply struct{} -type misnamedScheduleDeactivateRequest struct{} +type misnamedScheduleDeactivateRequest struct { + A0 TickStatus +} type misnamedScheduleDeactivateReply struct{} type misnamedScheduleEntity struct { - schedule Schedule[misnamedSchedule] + schedule Reminder[misnamedSchedule] wakeErr error } @@ -64,7 +66,7 @@ func (e *misnamedScheduleEntity) Arm(ctx context.Context) error { return e.schedule.Set(ctx, "wake", After(12*time.Second), Handle(misnamedSchedule.OnDeactivate)) } -func (e *misnamedScheduleEntity) OnDeactivate(context.Context) error { +func (e *misnamedScheduleEntity) OnDeactivate(context.Context, TickStatus) error { return e.wakeErr } @@ -72,16 +74,16 @@ func (p *misnamedScheduleProxy) Arm(ctx context.Context) error { return p.invoker.Invoke(ctx, p.id, "Arm", &misnamedScheduleArmRequest{}, &misnamedScheduleArmReply{}) } -func (p *misnamedScheduleProxy) OnDeactivate(ctx context.Context) error { - return p.invoker.Invoke(ctx, p.id, "OnDeactivate", &misnamedScheduleDeactivateRequest{}, &misnamedScheduleDeactivateReply{}) +func (p *misnamedScheduleProxy) OnDeactivate(ctx context.Context, tick TickStatus) error { + return p.invoker.Invoke(ctx, p.id, "OnDeactivate", &misnamedScheduleDeactivateRequest{A0: tick}, &misnamedScheduleDeactivateReply{}) } -func dispatchMisnamedSchedule(ctx context.Context, instance misnamedSchedule, method string, _ any, _ any) error { +func dispatchMisnamedSchedule(ctx context.Context, instance misnamedSchedule, method string, args any, _ any) error { switch method { case "Arm": return instance.Arm(ctx) case "OnDeactivate": - return instance.OnDeactivate(ctx) + return instance.OnDeactivate(ctx, args.(*misnamedScheduleDeactivateRequest).A0) default: return fmt.Errorf("unknown method %q", method) } @@ -98,15 +100,22 @@ func newMisnamedScheduleCall(method string) (any, any) { } } +func newMisnamedScheduleReminderCall(method string, status TickStatus) (any, any) { + if method == "OnDeactivate" { + return &misnamedScheduleDeactivateRequest{A0: status}, &misnamedScheduleDeactivateReply{} + } + return nil, nil +} + func installMisnamedSchedule(t *testing.T, rt *Runtime, wakeErr error) { t.Helper() if err := InstallType[misnamedSchedule](rt, dispatchMisnamedSchedule, func(invoker Invoker, id GrainId) misnamedSchedule { return &misnamedScheduleProxy{invoker: invoker, id: id} - }, newMisnamedScheduleCall); err != nil { + }, newMisnamedScheduleCall, newMisnamedScheduleReminderCall); err != nil { t.Fatal(err) } if err := Register[misnamedSchedule](rt, func(b *Binder) misnamedSchedule { - return &misnamedScheduleEntity{schedule: NewSchedule[misnamedSchedule](b), wakeErr: wakeErr} + return &misnamedScheduleEntity{schedule: NewReminder[misnamedSchedule](b), wakeErr: wakeErr} }); err != nil { t.Fatal(err) } @@ -115,7 +124,7 @@ func installMisnamedSchedule(t *testing.T, rt *Runtime, wakeErr error) { // TestBackgroundError_ScheduledMethodNamedOnDeactivate pins the sealed-source // contract against the exact trap the old method-string API had: a scheduled // method deliberately named "OnDeactivate" must still be reported as a -// ScheduledInvocation. +// ReminderInvocation. func TestBackgroundError_ScheduledMethodNamedOnDeactivate(t *testing.T) { synctest.Test(t, func(t *testing.T) { start := time.Unix(0, 0).UTC() @@ -126,7 +135,7 @@ func TestBackgroundError_ScheduledMethodNamedOnDeactivate(t *testing.T) { WithClock(fakeClock), WithIdleTimeout(0), WithEvictionInterval(0), - WithScheduleInterval(time.Second), + WithReminderInterval(time.Second), OnError(func(event BackgroundError) { errorsSeen <- event }), @@ -143,12 +152,12 @@ func TestBackgroundError_ScheduledMethodNamedOnDeactivate(t *testing.T) { select { case got := <-errorsSeen: wantID := GrainId{GrainType: TypeName[misnamedSchedule](), GrainKey: "alice"} - source, ok := got.Source.(ScheduledInvocation) + source, ok := got.Source.(ReminderInvocation) if !ok { - t.Fatalf("source = %#v, want ScheduledInvocation", got.Source) + t.Fatalf("source = %#v, want ReminderInvocation", got.Source) } if got.GrainId != wantID || source.Method != "OnDeactivate" || !errors.Is(got.Err, wakeErr) { - t.Fatalf("event = %#v, want identity %v, ScheduledInvocation{Method: OnDeactivate}, error %v", got, wantID, wakeErr) + t.Fatalf("event = %#v, want identity %v, ReminderInvocation{Method: OnDeactivate}, error %v", got, wantID, wakeErr) } if _, isDeactivation := got.Source.(Deactivation); isDeactivation { t.Fatalf("source = %#v, must not be Deactivation", got.Source) @@ -172,7 +181,7 @@ func TestBackgroundError_CancelShapedErrorFromLivePoller(t *testing.T) { WithClock(fakeClock), WithIdleTimeout(0), WithEvictionInterval(0), - WithScheduleInterval(time.Second), + WithReminderInterval(time.Second), OnError(func(event BackgroundError) { errorsSeen <- event }), @@ -188,12 +197,12 @@ func TestBackgroundError_CancelShapedErrorFromLivePoller(t *testing.T) { select { case got := <-errorsSeen: - source, ok := got.Source.(ScheduledInvocation) + source, ok := got.Source.(ReminderInvocation) if !ok { - t.Fatalf("source = %#v, want ScheduledInvocation", got.Source) + t.Fatalf("source = %#v, want ReminderInvocation", got.Source) } if source.Method != "Wake" || !errors.Is(got.Err, context.Canceled) { - t.Fatalf("event = %#v, want ScheduledInvocation{Method: Wake} with a cancel-shaped error", got) + t.Fatalf("event = %#v, want ReminderInvocation{Method: Wake} with a cancel-shaped error", got) } default: t.Fatal("cancel-shaped error from a live poller did not reach OnError") @@ -247,14 +256,14 @@ func TestBackgroundError_ScheduleFaultsAreSilent(t *testing.T) { synctest.Test(t, func(t *testing.T) { start := time.Unix(0, 0).UTC() fakeClock := clock.NewFake(start) - backend := &failingScheduleStore{Memory: store.NewMemory()} + backend := &failingReminderStore{Memory: store.NewMemory()} errorsSeen := make(chan BackgroundError, 1) rt := mustNew(t, WithStore(backend), WithClock(fakeClock), WithIdleTimeout(0), WithEvictionInterval(0), - WithScheduleInterval(time.Second), + WithReminderInterval(time.Second), OnError(func(event BackgroundError) { errorsSeen <- event }), @@ -316,7 +325,7 @@ func TestBackgroundError_CanceledDeliveryNotReported(t *testing.T) { WithClock(fakeClock), WithIdleTimeout(0), WithEvictionInterval(0), - WithScheduleInterval(time.Second), + WithReminderInterval(time.Second), OnError(func(event BackgroundError) { errorsSeen <- event }), diff --git a/benchmark_test.go b/benchmark_test.go index 859182b..0532eb2 100644 --- a/benchmark_test.go +++ b/benchmark_test.go @@ -94,7 +94,7 @@ func installBenchmarkEntity(b *testing.B, rt *Runtime) { b.Helper() if err := InstallType[benchmarkEntity](rt, dispatchBenchmarkEntity, func(invoker Invoker, id GrainId) benchmarkEntity { return &benchmarkEntityProxy{invoker: invoker, id: id} - }, newBenchmarkEntityCall); err != nil { + }, newBenchmarkEntityCall, nil); err != nil { rt.Close() b.Fatal(err) } @@ -194,7 +194,7 @@ func newBenchmarkForwardingRuntimes(b *testing.B) (*Runtime, *Runtime, GrainId, WithViewInterval(time.Hour), WithIdleTimeout(0), WithEvictionInterval(0), - WithScheduleInterval(0), + WithReminderInterval(0), WithTransport(nodeTransport), ) if err != nil { diff --git a/cluster_runtime_test.go b/cluster_runtime_test.go index 6c853c8..4e9a174 100644 --- a/cluster_runtime_test.go +++ b/cluster_runtime_test.go @@ -464,7 +464,7 @@ func installSideEffectEntity(t *testing.T, rt *Runtime, calls *atomic.Int32) { t.Helper() if err := InstallType[sideEffectEntity](rt, dispatchSideEffectEntity, func(invoker Invoker, id GrainId) sideEffectEntity { return &sideEffectEntityProxy{invoker: invoker, id: id} - }, newSideEffectEntityCall); err != nil { + }, newSideEffectEntityCall, nil); err != nil { t.Fatal(err) } if err := Register[sideEffectEntity](rt, func(b *Binder) sideEffectEntity { diff --git a/cmd/gorgen/testfixture/endtoend/domain/domain.go b/cmd/gorgen/testfixture/endtoend/domain/domain.go index e7cb1ae..b39b985 100644 --- a/cmd/gorgen/testfixture/endtoend/domain/domain.go +++ b/cmd/gorgen/testfixture/endtoend/domain/domain.go @@ -11,6 +11,7 @@ type Account interface { Deposit(ctx context.Context, amount int64) (int64, error) Snapshot(ctx context.Context) (int64, string, error) Reset(ctx context.Context) error + Tick(ctx context.Context, status gor.TickStatus) error } type account struct { @@ -36,3 +37,7 @@ func (a *account) Snapshot(context.Context) (int64, string, error) { func (a *account) Reset(ctx context.Context) error { return a.balance.Set(ctx, 0) } + +func (a *account) Tick(context.Context, gor.TickStatus) error { + return nil +} diff --git a/cmd/gorgen/testfixture/endtoend/gorgen/generated.go b/cmd/gorgen/testfixture/endtoend/gorgen/generated.go index 0a863fd..39c1980 100644 --- a/cmd/gorgen/testfixture/endtoend/gorgen/generated.go +++ b/cmd/gorgen/testfixture/endtoend/gorgen/generated.go @@ -47,6 +47,17 @@ func (p *accountProxy) Snapshot(ctx context.Context) (int64, string, error) { return reply.R0, reply.R1, err } +type accountTickRequest struct { + A0 gor.TickStatus +} +type accountTickReply struct{} + +func (p *accountProxy) Tick(ctx context.Context, status gor.TickStatus) error { + var reply accountTickReply + err := p.rt.Invoke(ctx, p.id, "Tick", &accountTickRequest{A0: status}, &reply) + return err +} + func dispatchAccount(ctx context.Context, instance domain.Account, method string, args any, reply any) error { switch method { case "Deposit": @@ -64,6 +75,10 @@ func dispatchAccount(ctx context.Context, instance domain.Account, method string typedReply.R0 = r0 typedReply.R1 = r1 return err + case "Tick": + typedArgs := args.(*accountTickRequest) + err := instance.Tick(ctx, typedArgs.A0) + return err default: return fmt.Errorf("unknown method %q", method) } @@ -77,6 +92,17 @@ func newAccountCall(method string) (args any, reply any) { return &accountResetRequest{}, &accountResetReply{} case "Snapshot": return &accountSnapshotRequest{}, &accountSnapshotReply{} + case "Tick": + return &accountTickRequest{}, &accountTickReply{} + default: + return nil, nil + } +} + +func newAccountReminderCall(method string, status gor.TickStatus) (args any, reply any) { + switch method { + case "Tick": + return &accountTickRequest{A0: status}, &accountTickReply{} default: return nil, nil } @@ -91,7 +117,7 @@ func newAccountProxy(rt gor.Invoker, id gor.GrainId) domain.Account { // the generated Grain types. After it returns nil, gor.Register and gor.Ref // can use those types with rt. func Install(rt *gor.Runtime) error { - if err := gor.InstallType[domain.Account](rt, dispatchAccount, newAccountProxy, newAccountCall); err != nil { + if err := gor.InstallType[domain.Account](rt, dispatchAccount, newAccountProxy, newAccountCall, newAccountReminderCall); err != nil { return err } return nil diff --git a/cmd/gorgen/testfixture/endtoend/gorgen/runtime_test.go b/cmd/gorgen/testfixture/endtoend/gorgen/runtime_test.go index 4f2a23d..5fe3edb 100644 --- a/cmd/gorgen/testfixture/endtoend/gorgen/runtime_test.go +++ b/cmd/gorgen/testfixture/endtoend/gorgen/runtime_test.go @@ -3,6 +3,7 @@ package gorgen import ( "context" "testing" + "time" "github.com/suraciii/gor" "github.com/suraciii/gor/cmd/gorgen/testfixture/endtoend/domain" @@ -37,6 +38,26 @@ func TestGeneratedAccountPersistsAcrossRestart(t *testing.T) { } } +func TestNewAccountReminderCallBuildsTypedRequest(t *testing.T) { + status := gor.TickStatus{ + FirstTickTime: time.Unix(10, 0).UTC(), + Period: time.Minute, + CurrentTickTime: time.Unix(20, 0).UTC(), + } + args, reply := newAccountReminderCall("Tick", status) + typedArgs, ok := args.(*accountTickRequest) + if !ok || typedArgs.A0 != status { + t.Fatalf("newAccountReminderCall(Tick) args = %#v, want TickStatus %#v", args, status) + } + if _, ok := reply.(*accountTickReply); !ok { + t.Fatalf("newAccountReminderCall(Tick) reply = %T, want *accountTickReply", reply) + } + args, reply = newAccountReminderCall("Missing", status) + if args != nil || reply != nil { + t.Fatalf("newAccountReminderCall(Missing) = (%T, %T), want (nil, nil)", args, reply) + } +} + func TestNewAccountCallUnknownMethodReturnsNil(t *testing.T) { args, reply := newAccountCall("Missing") if args != nil || reply != nil { diff --git a/cycle_test.go b/cycle_test.go index c519ad6..094b36c 100644 --- a/cycle_test.go +++ b/cycle_test.go @@ -96,7 +96,7 @@ func installChainWithFactory(t *testing.T, rt *Runtime, factory func(*Binder) ch t.Helper() if err := InstallType[chainEntity](rt, dispatchChain, func(invoker Invoker, id GrainId) chainEntity { return &chainEntityProxy{invoker: invoker, id: id} - }, newChainCall); err != nil { + }, newChainCall, nil); err != nil { t.Fatal(err) } if err := Register[chainEntity](rt, factory); err != nil { diff --git a/examples/shadow/cmd/load/main.go b/examples/shadow/cmd/load/main.go index 5c91777..657ac59 100644 --- a/examples/shadow/cmd/load/main.go +++ b/examples/shadow/cmd/load/main.go @@ -69,7 +69,7 @@ func run(ctx context.Context, args []string) (runErr error) { gor.WithClock(sourceClock), gor.WithIdleTimeout(idleTimeout), gor.WithEvictionInterval(evictionInterval), - gor.WithScheduleInterval(time.Second), + gor.WithReminderInterval(time.Second), gor.OnError(shadow.LogBackgroundError), ) if err != nil { diff --git a/examples/shadow/cmd/load/main_test.go b/examples/shadow/cmd/load/main_test.go index a334ef4..f0b194c 100644 --- a/examples/shadow/cmd/load/main_test.go +++ b/examples/shadow/cmd/load/main_test.go @@ -17,7 +17,7 @@ func TestReportDevicesReturnsAfterSuccess(t *testing.T) { gor.WithClock(clock.NewFake(time.Unix(0, 0).UTC())), gor.WithIdleTimeout(0), gor.WithEvictionInterval(0), - gor.WithScheduleInterval(0), + gor.WithReminderInterval(0), ) if err != nil { t.Fatal(err) diff --git a/examples/shadow/cmd/shadow/main_test.go b/examples/shadow/cmd/shadow/main_test.go index e941cab..a6a2bf3 100644 --- a/examples/shadow/cmd/shadow/main_test.go +++ b/examples/shadow/cmd/shadow/main_test.go @@ -31,7 +31,7 @@ func TestNewRuntimeReportsScheduledFailure(t *testing.T) { gor.WithClock(sourceClock), gor.WithIdleTimeout(0), gor.WithEvictionInterval(0), - gor.WithScheduleInterval(time.Second), + gor.WithReminderInterval(time.Second), ) if err != nil { logger.SetOutput(previousWriter) @@ -47,7 +47,7 @@ func TestNewRuntimeReportsScheduledFailure(t *testing.T) { logger.SetOutput(previousWriter) }() - if err := backend.Put(context.Background(), store.Schedule{ + if err := backend.Put(context.Background(), store.Reminder{ GrainId: store.GrainId{GrainType: gor.TypeName[domain.Device](), GrainKey: "device-1"}, Name: "broken", Method: "NotAMethod", diff --git a/examples/shadow/domain/domain.go b/examples/shadow/domain/domain.go index 823d712..fe4be3d 100644 --- a/examples/shadow/domain/domain.go +++ b/examples/shadow/domain/domain.go @@ -36,7 +36,7 @@ type Device interface { Report(ctx context.Context, workshopID string, state string) error Configure(ctx context.Context, configuration string) error Shadow(ctx context.Context) (Shadow, error) - MarkOffline(ctx context.Context) error + MarkOffline(ctx context.Context, tick gor.TickStatus) error } //gor:grain @@ -50,7 +50,7 @@ type device struct { binder *gor.Binder id gor.GrainId shadow gor.State[Shadow] - schedule gor.Schedule[Device] + schedule gor.Reminder[Device] lifecycleEvents chan<- LifecycleEvent } @@ -67,7 +67,7 @@ func newDevice(b *gor.Binder, events chan<- LifecycleEvent) Device { binder: b, id: gor.Self(b), shadow: gor.NewState[Shadow](b, "shadow"), - schedule: gor.NewSchedule[Device](b), + schedule: gor.NewReminder[Device](b), lifecycleEvents: events, } } @@ -130,7 +130,7 @@ func (d *device) emitLifecycle(kind string) { } } -func (d *device) MarkOffline(ctx context.Context) error { +func (d *device) MarkOffline(ctx context.Context, _ gor.TickStatus) error { shadow := d.shadow.Get() shadow.Online = false if err := d.shadow.Set(ctx, shadow); err != nil { diff --git a/examples/shadow/domain/gorgen/generated.go b/examples/shadow/domain/gorgen/generated.go index b0b8fbc..b1bb216 100644 --- a/examples/shadow/domain/gorgen/generated.go +++ b/examples/shadow/domain/gorgen/generated.go @@ -24,12 +24,14 @@ func (p *deviceProxy) Configure(ctx context.Context, configuration string) error return err } -type deviceMarkOfflineRequest struct{} +type deviceMarkOfflineRequest struct { + A0 gor.TickStatus +} type deviceMarkOfflineReply struct{} -func (p *deviceProxy) MarkOffline(ctx context.Context) error { +func (p *deviceProxy) MarkOffline(ctx context.Context, tick gor.TickStatus) error { var reply deviceMarkOfflineReply - err := p.rt.Invoke(ctx, p.id, "MarkOffline", &deviceMarkOfflineRequest{}, &reply) + err := p.rt.Invoke(ctx, p.id, "MarkOffline", &deviceMarkOfflineRequest{A0: tick}, &reply) return err } @@ -63,7 +65,8 @@ func dispatchDevice(ctx context.Context, instance domain.Device, method string, err := instance.Configure(ctx, typedArgs.A0) return err case "MarkOffline": - err := instance.MarkOffline(ctx) + typedArgs := args.(*deviceMarkOfflineRequest) + err := instance.MarkOffline(ctx, typedArgs.A0) return err case "Report": typedArgs := args.(*deviceReportRequest) @@ -94,6 +97,15 @@ func newDeviceCall(method string) (args any, reply any) { } } +func newDeviceReminderCall(method string, status gor.TickStatus) (args any, reply any) { + switch method { + case "MarkOffline": + return &deviceMarkOfflineRequest{A0: status}, &deviceMarkOfflineReply{} + default: + return nil, nil + } +} + func newDeviceProxy(rt gor.Invoker, id gor.GrainId) domain.Device { return &deviceProxy{id: id, rt: rt} } @@ -169,6 +181,13 @@ func newWorkshopCall(method string) (args any, reply any) { } } +func newWorkshopReminderCall(method string, status gor.TickStatus) (args any, reply any) { + switch method { + default: + return nil, nil + } +} + func newWorkshopProxy(rt gor.Invoker, id gor.GrainId) domain.Workshop { return &workshopProxy{id: id, rt: rt} } @@ -178,10 +197,10 @@ func newWorkshopProxy(rt gor.Invoker, id gor.GrainId) domain.Workshop { // the generated Grain types. After it returns nil, gor.Register and gor.Ref // can use those types with rt. func Install(rt *gor.Runtime) error { - if err := gor.InstallType[domain.Device](rt, dispatchDevice, newDeviceProxy, newDeviceCall); err != nil { + if err := gor.InstallType[domain.Device](rt, dispatchDevice, newDeviceProxy, newDeviceCall, newDeviceReminderCall); err != nil { return err } - if err := gor.InstallType[domain.Workshop](rt, dispatchWorkshop, newWorkshopProxy, newWorkshopCall); err != nil { + if err := gor.InstallType[domain.Workshop](rt, dispatchWorkshop, newWorkshopProxy, newWorkshopCall, newWorkshopReminderCall); err != nil { return err } return nil diff --git a/examples/shadow/http_test.go b/examples/shadow/http_test.go index dead000..868522f 100644 --- a/examples/shadow/http_test.go +++ b/examples/shadow/http_test.go @@ -21,7 +21,7 @@ func TestHTTPReportsConfiguresAndReadsShadow(t *testing.T) { gor.WithClock(sourceClock), gor.WithIdleTimeout(0), gor.WithEvictionInterval(0), - gor.WithScheduleInterval(time.Second), + gor.WithReminderInterval(time.Second), ) if err != nil { t.Fatal(err) @@ -78,7 +78,7 @@ func TestHTTPReportsConfiguresAndReadsShadow(t *testing.T) { func TestHTTPRejectsMalformedJSON(t *testing.T) { sourceClock := clock.NewFake(time.Unix(0, 0).UTC()) - rt, err := gor.New(gor.WithStore(store.NewMemory()), gor.WithClock(sourceClock), gor.WithScheduleInterval(time.Second), gor.WithIdleTimeout(0), gor.WithEvictionInterval(0)) + rt, err := gor.New(gor.WithStore(store.NewMemory()), gor.WithClock(sourceClock), gor.WithReminderInterval(time.Second), gor.WithIdleTimeout(0), gor.WithEvictionInterval(0)) if err != nil { t.Fatal(err) } @@ -105,7 +105,7 @@ func TestHTTPReportWriteFailureIsNotBadRequest(t *testing.T) { rt, err := gor.New( gor.WithStore(backend), gor.WithClock(sourceClock), - gor.WithScheduleInterval(time.Second), + gor.WithReminderInterval(time.Second), gor.WithIdleTimeout(0), gor.WithEvictionInterval(0), ) diff --git a/examples/shadow/runtime.go b/examples/shadow/runtime.go index bc0dd3c..de2333f 100644 --- a/examples/shadow/runtime.go +++ b/examples/shadow/runtime.go @@ -10,7 +10,7 @@ import ( func LogBackgroundError(event gor.BackgroundError) { switch source := event.Source.(type) { - case gor.ScheduledInvocation: + case gor.ReminderInvocation: log.Printf("%s/%s.%s failed: %v", event.GrainId.GrainType, event.GrainId.GrainKey, source.Method, event.Err) case gor.Deactivation: log.Printf("%s/%s deactivation (%v) failed: %v", event.GrainId.GrainType, event.GrainId.GrainKey, source.Reason, event.Err) diff --git a/examples/shadow/shadow_test.go b/examples/shadow/shadow_test.go index 8d9bc8b..fc1ea28 100644 --- a/examples/shadow/shadow_test.go +++ b/examples/shadow/shadow_test.go @@ -26,7 +26,7 @@ func TestDeviceShadowTracksReportsAndWorkshopPresence(t *testing.T) { gor.WithClock(sourceClock), gor.WithIdleTimeout(0), gor.WithEvictionInterval(0), - gor.WithScheduleInterval(time.Second), + gor.WithReminderInterval(time.Second), ) if err != nil { t.Fatal(err) @@ -71,12 +71,12 @@ func TestDeviceShadowTracksReportsAndWorkshopPresence(t *testing.T) { if err := device.Report(ctx, "assembly", "temperature=21"); err != nil { t.Fatal(err) } - schedules, err := backend.ListDue(ctx, start.Add(59*time.Second)) + reminders, err := backend.ListDue(ctx, start.Add(59*time.Second)) if err != nil { t.Fatal(err) } - if len(schedules) != 1 || !schedules[0].DueAt.Equal(start.Add(59*time.Second)) { - t.Fatalf("offline schedule after second report = %#v, want one schedule due at 59s", schedules) + if len(reminders) != 1 || !reminders[0].DueAt.Equal(start.Add(59*time.Second)) { + t.Fatalf("offline schedule after second report = %#v, want one schedule due at 59s", reminders) } sourceClock.Advance(2 * time.Second) @@ -93,12 +93,12 @@ func TestDeviceShadowTracksReportsAndWorkshopPresence(t *testing.T) { if got, err := workshop.OnlineCount(ctx); err != nil || got != 0 { t.Fatalf("online count after timeout = (%d, %v), want (0, nil)", got, err) } - schedules, err = backend.ListDue(ctx, sourceClock.Now().Add(domain.OfflineAfter)) + reminders, err = backend.ListDue(ctx, sourceClock.Now().Add(domain.OfflineAfter)) if err != nil { t.Fatal(err) } - if len(schedules) != 0 { - t.Fatalf("schedules after timeout = %#v, want none", schedules) + if len(reminders) != 0 { + t.Fatalf("reminders after timeout = %#v, want none", reminders) } if err := device.Report(ctx, "assembly", "temperature=22"); err != nil { @@ -119,7 +119,7 @@ func TestDeviceIdleEvictionRunsLifecycleAndReloadsState(t *testing.T) { gor.WithClock(sourceClock), gor.WithIdleTimeout(2*time.Second), gor.WithEvictionInterval(time.Second), - gor.WithScheduleInterval(0), + gor.WithReminderInterval(0), ) if err != nil { t.Fatal(err) @@ -177,11 +177,11 @@ func TestScheduledFailureReachesOnError(t *testing.T) { errorsSeen := make(chan gor.BackgroundError, 1) rt, err := gor.New( gor.WithStore(backend), - gor.WithScheduleStore(backend), + gor.WithReminderStore(backend), gor.WithClock(sourceClock), gor.WithIdleTimeout(0), gor.WithEvictionInterval(0), - gor.WithScheduleInterval(time.Second), + gor.WithReminderInterval(time.Second), gor.OnError(func(event gor.BackgroundError) { errorsSeen <- event }), @@ -205,7 +205,7 @@ func TestScheduledFailureReachesOnError(t *testing.T) { select { case got := <-errorsSeen: wantID := gor.GrainId{GrainType: gor.TypeName[domain.Device](), GrainKey: "device-1"} - source, ok := got.Source.(gor.ScheduledInvocation) + source, ok := got.Source.(gor.ReminderInvocation) if !ok || got.GrainId != wantID || source.Method != "MarkOffline" || !errors.Is(got.Err, errWorkshopWrite) { t.Fatalf("OnError event = %#v, want %v.MarkOffline with %v", got, wantID, errWorkshopWrite) } diff --git a/forward_teardown_test.go b/forward_teardown_test.go index 5360b70..506df3c 100644 --- a/forward_teardown_test.go +++ b/forward_teardown_test.go @@ -63,7 +63,7 @@ func installBlockingEntity(t *testing.T, rt *Runtime, entries, release chan stru t.Helper() if err := InstallType[blockingEntity](rt, dispatchBlockingEntity, func(invoker Invoker, id GrainId) blockingEntity { return &blockingEntityProxy{invoker: invoker, id: id} - }, newBlockingEntityCall); err != nil { + }, newBlockingEntityCall, nil); err != nil { t.Fatal(err) } if err := Register[blockingEntity](rt, func(b *Binder) blockingEntity { diff --git a/forward_test.go b/forward_test.go index 33781a7..d8395e5 100644 --- a/forward_test.go +++ b/forward_test.go @@ -702,7 +702,7 @@ func installEnvelopeAccount(t *testing.T, rt *Runtime) { t.Helper() if err := InstallType[envelopeAccount](rt, dispatchEnvelopeAccount, func(invoker Invoker, id GrainId) envelopeAccount { return &envelopeAccountProxy{invoker: invoker, id: id} - }, newEnvelopeAccountCall); err != nil { + }, newEnvelopeAccountCall, nil); err != nil { t.Fatal(err) } if err := Register[envelopeAccount](rt, func(*Binder) envelopeAccount { @@ -860,7 +860,7 @@ func installRoutedAccountWithFactory(t *testing.T, rt *Runtime, factory func(*Bi t.Helper() if err := InstallType[routedAccount](rt, dispatchRoutedAccount, func(invoker Invoker, id GrainId) routedAccount { return &routedAccountProxy{invoker: invoker, id: id} - }, newRoutedAccountCall); err != nil { + }, newRoutedAccountCall, nil); err != nil { t.Fatal(err) } if err := Register[routedAccount](rt, factory); err != nil { diff --git a/gor.go b/gor.go index ec3bd91..b39a4d9 100644 --- a/gor.go +++ b/gor.go @@ -65,8 +65,8 @@ type Scope interface { } // BackgroundError reports a failure of an application callback that has no -// caller waiting for its result: a claimed scheduled invocation, or a normal -// deactivation hook. GrainId is the affected entity, Err is the callback's +// caller waiting for its result: a claimed Reminder invocation, or a normal +// deactivation hook. GrainId is the affected Grain, Err is the callback's // error, and Source identifies which kind of callback failed. type BackgroundError struct { GrainId GrainId @@ -81,13 +81,13 @@ type ErrorSource interface { errorSource() } -// ScheduledInvocation is the source of a failure from a claimed scheduled -// invocation. Method is the entity method that was invoked. -type ScheduledInvocation struct { +// ReminderInvocation is the source of a failure from a claimed Reminder. +// Method is the Grain method that was invoked. +type ReminderInvocation struct { Method string } -func (ScheduledInvocation) errorSource() {} +func (ReminderInvocation) errorSource() {} // Deactivation is the source of a failure from a normal deactivation hook. // Reason is the reason of that deactivation. @@ -116,13 +116,13 @@ type Deactivatable interface { } // Runtime coordinates entity registration, activation, invocation, state, and -// schedules. Invocations for the same identity are serialized, and a runtime +// reminders. Invocations for the same identity are serialized, and a runtime // configured for a cluster can route an invocation to its current owner. // Create a Runtime with New and stop it with Close or Kill. type Runtime struct { engine *runtimepkg.Runtime store store.Store - scheduleStore store.ScheduleStore + reminderStore store.ReminderStore clock clock.Clock onError func(BackgroundError) onCall func(CallObservation) @@ -154,8 +154,8 @@ type Runtime struct { type Config struct { runtimepkg.Config Store store.Store - ScheduleStore store.ScheduleStore - ScheduleInterval time.Duration + ReminderStore store.ReminderStore + ReminderInterval time.Duration Transport transport.Transport OnError func(BackgroundError) OnCall func(CallObservation) @@ -188,9 +188,10 @@ type Invoker interface { var _ Invoker = (*Runtime)(nil) type typeRegistration struct { - dispatch runtimepkg.Dispatch - newProxy func(Invoker, GrainId) any - newCall func(string) (any, any) + dispatch runtimepkg.Dispatch + newProxy func(Invoker, GrainId) any + newCall func(string) (any, any) + newReminderCall func(string, TickStatus) (any, any) } // Option configures a Runtime created by New. New applies options in argument @@ -208,8 +209,8 @@ func (b *Binder) scopeRuntime() *Runtime { // New creates and starts a Runtime. // // By default, New uses clock.Real{}, store.NewMemory for entity state and -// schedules, a mailbox capacity of 16, a one-minute idle timeout, one-second -// eviction and schedule intervals, and one-second heartbeat and view +// reminders, a mailbox capacity of 16, a one-minute idle timeout, one-second +// eviction and Reminder intervals, and one-second heartbeat and view // intervals. A MemberStore and Transport must be configured together. In // clustered mode, ProbeInterval, ProbeTimeout, ProbeFailures, VoteTTL, // MaxTickGap, and MaxTableLatency default to one second, 500 ms, three, six @@ -227,7 +228,7 @@ func New(options ...Option) (*Runtime, error) { EvictionInterval: time.Second, }, Store: store.NewMemory(), - ScheduleInterval: time.Second, + ReminderInterval: time.Second, HeartbeatInterval: time.Second, ViewInterval: time.Second, } @@ -237,9 +238,9 @@ func New(options ...Option) (*Runtime, error) { if (config.MemberStore == nil) != (config.Transport == nil) { return nil, errors.New("member store and transport must be configured together") } - if config.ScheduleStore == nil { - if schedules, ok := config.Store.(store.ScheduleStore); ok { - config.ScheduleStore = schedules + if config.ReminderStore == nil { + if reminders, ok := config.Store.(store.ReminderStore); ok { + config.ReminderStore = reminders } } var ( @@ -271,7 +272,7 @@ func New(options ...Option) (*Runtime, error) { rt := &Runtime{ engine: runtimepkg.New(config.Config), store: config.Store, - scheduleStore: config.ScheduleStore, + reminderStore: config.ReminderStore, clock: config.Clock, onError: config.OnError, onCall: config.OnCall, @@ -286,8 +287,8 @@ func New(options ...Option) (*Runtime, error) { rt.clusterView.Store(&initialView) go rt.watchCluster() } - if config.ScheduleStore != nil && config.ScheduleInterval > 0 { - rt.poller = timer.New(config.ScheduleStore, config.Clock, config.ScheduleInterval, scheduleInvoker{runtime: rt}) + if config.ReminderStore != nil && config.ReminderInterval > 0 { + rt.poller = timer.New(config.ReminderStore, config.Clock, config.ReminderInterval, reminderInvoker{runtime: rt}, rt.newReminderCall) } if rt.clusterNode != nil && rt.transport != nil { rt.startTransport() @@ -330,30 +331,30 @@ func WithEvictionInterval(value time.Duration) Option { } } -// WithStore sets the store used for entity state. If omitted, New uses an -// in-memory store; when the selected store also implements ScheduleStore and -// no schedule store is supplied, New uses it for schedules too. +// WithStore sets the store used for Grain State. If omitted, New uses an +// in-memory store; when the selected store also implements ReminderStore and +// no Reminder store is supplied, New uses it for Reminders too. func WithStore(value store.Store) Option { return func(config *Config) { config.Store = value } } -// WithScheduleStore sets the store used for entity schedules. If omitted, New -// derives it from Store when Store implements ScheduleStore; otherwise schedule -// operations return ErrScheduleStoreUnavailable. -func WithScheduleStore(value store.ScheduleStore) Option { +// WithReminderStore sets the store used for Reminders. If omitted, New +// derives it from Store when Store implements ReminderStore; otherwise Reminder +// operations return ErrReminderStoreUnavailable. +func WithReminderStore(value store.ReminderStore) Option { return func(config *Config) { - config.ScheduleStore = value + config.ReminderStore = value } } -// WithScheduleInterval sets the interval for background schedule polling. If -// omitted, New polls once per second. A non-positive value keeps schedules +// WithReminderInterval sets the interval for background Reminder polling. If +// omitted, New polls once per second. A non-positive value keeps Reminders // persisted but disables automatic polling. -func WithScheduleInterval(value time.Duration) Option { +func WithReminderInterval(value time.Duration) Option { return func(config *Config) { - config.ScheduleInterval = value + config.ReminderInterval = value } } @@ -367,10 +368,10 @@ func WithTransport(value transport.Transport) Option { } // OnError sets the callback for failures of background application callbacks: -// claimed scheduled invocations and normal OnDeactivate hooks. If omitted, +// claimed Reminder invocations and normal OnDeactivate hooks. If omitted, // those errors are not reported. The callback may run asynchronously and // concurrently with application code; it is not called for ordinary foreground -// Invoke errors. A scheduled invocation whose delivery is canceled because the +// Invoke errors. A Reminder invocation whose delivery is canceled because the // poller's context was canceled during shutdown is not reported. ListDue and // Claim failures are not reported either. Event sources are sealed: branch on // the concrete type of Source, never on method-name strings. @@ -571,11 +572,11 @@ type boundInstance struct { binder *Binder } -// InstallType installs the dispatch and proxy factories required for T in rt. -// Generated Install code calls it; application code should use the generated -// installer rather than hand-writing this integration seam. It returns an -// error when T is already installed in rt. -func InstallType[T any](rt *Runtime, dispatch func(context.Context, T, string, any, any) error, newProxy func(Invoker, GrainId) T, newCall func(string) (any, any)) error { +// InstallType installs the dispatch, proxy, normal-call, and Reminder-call +// factories required for T in rt. Generated Install code calls it; application +// code should use the generated installer rather than hand-writing this seam. +// It returns an error when T is already installed in rt. +func InstallType[T any](rt *Runtime, dispatch func(context.Context, T, string, any, any) error, newProxy func(Invoker, GrainId) T, newCall func(string) (any, any), newReminderCall func(string, TickStatus) (any, any)) error { name := TypeName[T]() rt.typesMu.Lock() defer rt.typesMu.Unlock() @@ -589,7 +590,8 @@ func InstallType[T any](rt *Runtime, dispatch func(context.Context, T, string, a newProxy: func(invoker Invoker, id GrainId) any { return newProxy(invoker, id) }, - newCall: newCall, + newCall: newCall, + newReminderCall: newReminderCall, } return nil } @@ -674,6 +676,18 @@ func (rt *Runtime) typeRegistration(name string) (typeRegistration, bool) { return registration, ok } +func (rt *Runtime) newReminderCall(id store.GrainId, method string, firstTickTime time.Time, period time.Duration, currentTickTime time.Time) (any, any) { + registration, ok := rt.typeRegistration(id.GrainType) + if !ok || registration.newReminderCall == nil { + return nil, nil + } + return registration.newReminderCall(method, TickStatus{ + FirstTickTime: firstTickTime, + Period: period, + CurrentTickTime: currentTickTime, + }) +} + // Close begins an orderly shutdown. It stops admitting new entity calls, // lets calls already admitted finish, rejects queued calls without entering // their method bodies, and then waits for in-flight methods, normal @@ -946,27 +960,23 @@ func (rt *Runtime) deactivateMovedActivations(view cluster.View) { } } -type scheduleInvoker struct { +type reminderInvoker struct { runtime *Runtime } -func (i scheduleInvoker) Invoke(ctx context.Context, id store.GrainId, method string) error { - err := i.runtime.Invoke(ctx, GrainId(id), method, nil, nil) - // A delivery that failed because the poller's context was canceled is a - // clean shutdown, not a callback failure: reporting it would raise a false - // alarm on every orderly close. The judge is the poller context, not the - // shape of the error. +func (i reminderInvoker) Invoke(ctx context.Context, id store.GrainId, method string, args any, reply any) error { + err := i.runtime.Invoke(ctx, GrainId(id), method, args, reply) if err != nil && ctx.Err() == nil && i.runtime.onError != nil { i.runtime.onError(BackgroundError{ GrainId: GrainId(id), Err: err, - Source: ScheduledInvocation{Method: method}, + Source: ReminderInvocation{Method: method}, }) } return err } -func (i scheduleInvoker) Owns(id store.GrainId) bool { +func (i reminderInvoker) Owns(id store.GrainId) bool { return i.runtime.Owns(id) } diff --git a/gor_test.go b/gor_test.go index b412a58..265950e 100644 --- a/gor_test.go +++ b/gor_test.go @@ -182,7 +182,7 @@ func installLifecycleAccount(t *testing.T, rt *Runtime, factoryCalls *atomic.Int t.Helper() if err := InstallType[lifecycleAccount](rt, dispatchLifecycleAccount, func(invoker Invoker, id GrainId) lifecycleAccount { return &lifecycleAccountProxy{invoker: invoker, id: id} - }, newLifecycleAccountCall); err != nil { + }, newLifecycleAccountCall, nil); err != nil { t.Fatal(err) } if err := Register[lifecycleAccount](rt, func(b *Binder) lifecycleAccount { @@ -532,7 +532,7 @@ func TestBinderScope_ProvidesClockAndTypedReferences(t *testing.T) { } if err := InstallType[scopeAccount](rt, dispatchScopeAccount, func(invoker Invoker, id GrainId) scopeAccount { return &scopeAccountProxy{invoker: invoker, id: id} - }, newScopeAccountCall); err != nil { + }, newScopeAccountCall, nil); err != nil { t.Fatal(err) } if err := Register[scopeAccount](rt, func(b *Binder) scopeAccount { @@ -891,7 +891,7 @@ func installAccountWithDispatch(t *testing.T, rt *Runtime, dispatch func(context t.Helper() if err := InstallType[Account](rt, dispatch, func(invoker Invoker, id GrainId) Account { return &accountProxy{invoker: invoker, id: id} - }, newAccountCall); err != nil { + }, newAccountCall, nil); err != nil { t.Fatal(err) } } diff --git a/internal/codegen/load.go b/internal/codegen/load.go index 05091a7..c3dc349 100644 --- a/internal/codegen/load.go +++ b/internal/codegen/load.go @@ -102,9 +102,10 @@ type pendingInterface struct { } type pendingMethod struct { - name string - params []pendingParameter - results []types.Type + name string + reminder bool + params []pendingParameter + results []types.Type } type pendingParameter struct { @@ -138,7 +139,7 @@ func loadInterface(pkg *packages.Package, specification *ast.TypeSpec, imports m if signature.Results().Len() == 0 || !isError(signature.Results().At(signature.Results().Len()-1).Type()) { return pendingInterface{}, locatedError(pkg.Fset, method.Pos(), "%s.%s must have error as its last result", model.name, method.Name()) } - loaded := pendingMethod{name: method.Name()} + loaded := pendingMethod{name: method.Name(), reminder: isReminderMethod(signature)} for parameter := 0; parameter < signature.Params().Len(); parameter++ { variable := signature.Params().At(parameter) name := variable.Name() @@ -157,6 +158,21 @@ func loadInterface(pkg *packages.Package, specification *ast.TypeSpec, imports m return model, nil } +func isReminderMethod(signature *types.Signature) bool { + if signature.Params().Len() != 2 || signature.Results().Len() != 1 { + return false + } + return isTickStatus(signature.Params().At(1).Type()) && isError(signature.Results().At(0).Type()) +} + +func isTickStatus(value types.Type) bool { + named, ok := value.(*types.Named) + if !ok || named.Obj().Pkg() == nil { + return false + } + return named.Obj().Name() == "TickStatus" && named.Obj().Pkg().Path() == "github.com/suraciii/gor" +} + // recordImports adds every package referenced by value to imports, keyed by // path. The rendered string is discarded; only the referenced set matters. func recordImports(value types.Type, imports map[string]Import) { @@ -184,7 +200,7 @@ func materialize(entity pendingInterface, aliases map[string]string) Interface { } model := Interface{Name: entity.name, Methods: make([]Method, len(entity.methods))} for i, method := range entity.methods { - loaded := Method{Name: method.name} + loaded := Method{Name: method.name, Reminder: method.reminder} for _, parameter := range method.params { loaded.Params = append(loaded.Params, Parameter{Name: parameter.name, Type: types.TypeString(parameter.typ, qualifier)}) } diff --git a/internal/codegen/model.go b/internal/codegen/model.go index dbeb2c3..ef37cdd 100644 --- a/internal/codegen/model.go +++ b/internal/codegen/model.go @@ -20,9 +20,10 @@ type Interface struct { } type Method struct { - Name string - Params []Parameter - Results []string + Name string + Reminder bool + Params []Parameter + Results []string } type Parameter struct { diff --git a/internal/codegen/render.go b/internal/codegen/render.go index 87fe988..6b53938 100644 --- a/internal/codegen/render.go +++ b/internal/codegen/render.go @@ -35,6 +35,7 @@ type renderInterface struct { type renderMethod struct { Name string + Reminder bool Params string Results string ContextName string @@ -116,6 +117,7 @@ func prepareMethod(entity Interface, method Method) renderMethod { values := method.Results[:len(method.Results)-1] rendered := renderMethod{ Name: method.Name, + Reminder: method.Reminder, Params: joinParameters(method.Params), Results: joinResults(method.Results), ContextName: method.Params[0].Name, @@ -248,6 +250,15 @@ func new{{ $entity.Name }}Call(method string) (args any, reply any) { } } +func new{{ $entity.Name }}ReminderCall(method string, status gor.TickStatus) (args any, reply any) { + switch method { +{{range .Methods}}{{if .Reminder}} case "{{.Name}}": + return &{{.ArgsName}}{A0: status}, &{{.ReplyName}}{} +{{end}}{{end}} default: + return nil, nil + } +} + func {{$entity.ConstructorName}}(rt gor.Invoker, id gor.GrainId) {{$.SourcePackage}}.{{$entity.Name}} { return &{{.ProxyName}}{id: id, rt: rt} } @@ -258,7 +269,7 @@ func {{$entity.ConstructorName}}(rt gor.Invoker, id gor.GrainId) {{$.SourcePacka // the generated Grain types. After it returns nil, gor.Register and gor.Ref // can use those types with rt. func Install(rt *gor.Runtime) error { -{{range .Interfaces}} if err := gor.InstallType[{{$.SourcePackage}}.{{.Name}}](rt, {{.DispatchName}}, {{.ConstructorName}}, new{{.Name}}Call); err != nil { +{{range .Interfaces}} if err := gor.InstallType[{{$.SourcePackage}}.{{.Name}}](rt, {{.DispatchName}}, {{.ConstructorName}}, new{{.Name}}Call, new{{.Name}}ReminderCall); err != nil { return err } {{end}} return nil diff --git a/internal/codegen/testfixture/generated/generated.go b/internal/codegen/testfixture/generated/generated.go index 43fd263..cc77fee 100644 --- a/internal/codegen/testfixture/generated/generated.go +++ b/internal/codegen/testfixture/generated/generated.go @@ -64,6 +64,13 @@ func newAccountCall(method string) (args any, reply any) { } } +func newAccountReminderCall(method string, status gor.TickStatus) (args any, reply any) { + switch method { + default: + return nil, nil + } +} + func newAccountProxy(rt gor.Invoker, id gor.GrainId) domain.Account { return &accountProxy{id: id, rt: rt} } @@ -105,6 +112,13 @@ func newLedgerCall(method string) (args any, reply any) { } } +func newLedgerReminderCall(method string, status gor.TickStatus) (args any, reply any) { + switch method { + default: + return nil, nil + } +} + func newLedgerProxy(rt gor.Invoker, id gor.GrainId) domain.Ledger { return &ledgerProxy{id: id, rt: rt} } @@ -114,10 +128,10 @@ func newLedgerProxy(rt gor.Invoker, id gor.GrainId) domain.Ledger { // the generated Grain types. After it returns nil, gor.Register and gor.Ref // can use those types with rt. func Install(rt *gor.Runtime) error { - if err := gor.InstallType[domain.Account](rt, dispatchAccount, newAccountProxy, newAccountCall); err != nil { + if err := gor.InstallType[domain.Account](rt, dispatchAccount, newAccountProxy, newAccountCall, newAccountReminderCall); err != nil { return err } - if err := gor.InstallType[domain.Ledger](rt, dispatchLedger, newLedgerProxy, newLedgerCall); err != nil { + if err := gor.InstallType[domain.Ledger](rt, dispatchLedger, newLedgerProxy, newLedgerCall, newLedgerReminderCall); err != nil { return err } return nil diff --git a/method_handle_test.go b/method_handle_test.go index d377365..97b7a94 100644 --- a/method_handle_test.go +++ b/method_handle_test.go @@ -6,11 +6,11 @@ import ( ) // handleProbe is the entity interface the extraction tripwire locks its map -// on. Both methods must have the schedule shape func(handleProbe, -// context.Context) error, which is what Handle admits. +// on. Both methods must have the Reminder shape func(handleProbe, +// context.Context, TickStatus) error, which is what Handle admits. type handleProbe interface { - Wake(context.Context) error - Arm(context.Context) error + Wake(context.Context, TickStatus) error + Arm(context.Context, TickStatus) error } // TestHandle_ExtractsTrailingMethodName locks the map from an interface method @@ -18,7 +18,7 @@ type handleProbe interface { // reflect and runtime.FuncForPC, and FuncForPC's name format is not a // Go-documented contract — it is empirically stable, but Go is free to change // it. This test exists so that a Go upgrade which changes the encoding breaks -// the build instead of silently mis-naming schedules: a mis-read name would be +// the build instead of silently mis-naming reminders: a mis-read name would be // stored into the schedule table, match no dispatch case, and fail every // delivery with "unknown method". It is a tripwire for an external contract, // not a check of business logic — keep it as long as Handle reads names from @@ -26,7 +26,7 @@ type handleProbe interface { func TestHandle_ExtractsTrailingMethodName(t *testing.T) { handles := []struct { name string - method func(handleProbe, context.Context) error + method func(handleProbe, context.Context, TickStatus) error want string }{ {name: "Wake", method: handleProbe.Wake, want: "Wake"}, diff --git a/observability_test.go b/observability_test.go index d758e2b..8afda21 100644 --- a/observability_test.go +++ b/observability_test.go @@ -80,7 +80,7 @@ func newObservedRuntime(t *testing.T, sourceClock clock.Clock, events chan<- Cal WithClock(sourceClock), WithIdleTimeout(0), WithEvictionInterval(0), - WithScheduleInterval(0), + WithReminderInterval(0), } if events != nil { base = append(base, OnCall(func(observation CallObservation) { @@ -92,7 +92,7 @@ func newObservedRuntime(t *testing.T, sourceClock clock.Clock, events chan<- Cal return nil }, func(string) (any, any) { return nil, nil - }); err != nil { + }, nil); err != nil { t.Fatal(err) } if err := Register[observedAccount](rt, factory); err != nil { @@ -165,7 +165,7 @@ func TestOnCallDoesNotReportDeactivation(t *testing.T) { }) } -func TestOnCallReportsScheduledInvocation(t *testing.T) { +func TestOnCallReportsReminderInvocation(t *testing.T) { synctest.Test(t, func(t *testing.T) { start := time.Unix(0, 0).UTC() fakeClock := clock.NewFake(start) @@ -175,7 +175,7 @@ func TestOnCallReportsScheduledInvocation(t *testing.T) { WithClock(fakeClock), WithIdleTimeout(0), WithEvictionInterval(0), - WithScheduleInterval(time.Second), + WithReminderInterval(time.Second), OnCall(func(observation CallObservation) { events <- observation }), diff --git a/schedule.go b/schedule.go index 9b8515b..395bed6 100644 --- a/schedule.go +++ b/schedule.go @@ -12,107 +12,101 @@ import ( "github.com/suraciii/gor/store" ) -// ErrScheduleStoreUnavailable is returned by Schedule.Set and Schedule.Cancel -// when no schedule store is configured. It is a sentinel suitable for +// ErrReminderStoreUnavailable is returned by Reminder.Set and Reminder.Cancel +// when no reminder store is configured. It is a sentinel suitable for // errors.Is. -var ErrScheduleStoreUnavailable = errors.New("schedule store is not configured") +var ErrReminderStoreUnavailable = errors.New("reminder store is not configured") -// ScheduleTime describes when a schedule first runs and whether it repeats. -// Its zero value is valid and is equivalent to After(0): a one-shot schedule +// ReminderTime describes when a Reminder first runs and whether it repeats. +// Its zero value is valid and is equivalent to After(0): a one-shot Reminder // due at the clock time used by Set. -type ScheduleTime struct { +type ReminderTime struct { delay time.Duration interval time.Duration } -// After returns a one-shot schedule due delay after Set uses its clock. A zero +// After returns a one-shot Reminder due delay after Set uses its clock. A zero // delay is due immediately; a negative delay is due in the past and is -// eligible on the next scheduler poll. -func After(delay time.Duration) ScheduleTime { - return ScheduleTime{delay: delay} +// eligible on the next poller poll. +func After(delay time.Duration) ReminderTime { + return ReminderTime{delay: delay} } -// Every returns a schedule whose first due time is interval after Set uses its +// Every returns a Reminder whose first due time is interval after Set uses its // clock and whose subsequent due times use the same interval when interval is -// positive. Every(0) is accepted and produces a one-shot schedule, just like -// After(0). A negative interval is also accepted and stored; it places the -// schedule in the past so it remains due instead of producing a future -// recurring deadline. -func Every(interval time.Duration) ScheduleTime { - return ScheduleTime{delay: interval, interval: interval} +// positive. Every(0) is accepted and produces a one-shot Reminder, just like +// After(0). A negative interval is accepted and stored. +func Every(interval time.Duration) ReminderTime { + return ReminderTime{delay: interval, interval: interval} } -// MethodHandle names one method of T for a schedule. Build one with Handle +// TickStatus describes the time represented by one Reminder delivery. +type TickStatus struct { + FirstTickTime time.Time + Period time.Duration + CurrentTickTime time.Time +} + +// MethodHandle names one method of T for a Reminder. Build one with Handle // from a method expression on T's interface; the type parameter ties the -// handle to the entity interface, so a handle built from another interface -// does not assign and cannot reach this schedule. +// handle to the Grain interface. type MethodHandle[T any] struct { method string } // Handle builds a MethodHandle from a method expression on T's interface, // such as gor.Handle(Account.ApplyInterest). The method name is read off the -// expression once, at this call: reflect and runtime.FuncForPC yield the -// full function name, and its trailing segment is the method name. The -// expression must be a method expression on the interface — a hand-written -// closure of the same function type also compiles, but the name read off it -// is not a method name and delivery fails with "unknown method". -func Handle[T any](m func(T, context.Context) error) MethodHandle[T] { +// expression once, at this call. A closure of the same function type compiles, +// but its name is not a method name and delivery fails with "unknown method". +func Handle[T any](m func(T, context.Context, TickStatus) error) MethodHandle[T] { full := runtime.FuncForPC(reflect.ValueOf(m).Pointer()).Name() name := full[strings.LastIndexByte(full, '.')+1:] return MethodHandle[T]{method: name} } -// Schedule manages schedules for the entity bound to a Binder, typed to the -// entity's interface T. Obtain one with NewSchedule[T]; the zero value has no -// schedule store and its operations return ErrScheduleStoreUnavailable. -type Schedule[T any] struct { +// Reminder manages Reminders for the Grain bound to a Binder, typed to the +// Grain's interface T. Obtain one with NewReminder[T]; the zero value has no +// reminder store and its operations return ErrReminderStoreUnavailable. +type Reminder[T any] struct { identity store.GrainId - store store.ScheduleStore + store store.ReminderStore clock clock.Clock } -// NewSchedule returns a schedule manager bound to the entity represented by b +// NewReminder returns a Reminder manager bound to the Grain represented by b // and typed to its interface T. -func NewSchedule[T any](b *Binder) Schedule[T] { - return Schedule[T]{ +func NewReminder[T any](b *Binder) Reminder[T] { + return Reminder[T]{ identity: b.identity, - store: b.runtime.scheduleStore, + store: b.runtime.reminderStore, clock: b.runtime.clock, } } -// Set creates or replaces the named schedule for the bound entity. m must be -// a Handle built from a method expression on the entity's interface; the -// method name is read off the handle once, here, and stored. Set does not -// validate that the name has a dispatch case, so a handle built from a -// closure rather than a method expression is stored and fails with "unknown -// method" when the scheduler invokes it. A successful Set persists the -// schedule. Each due occurrence is delivered at most once, and an invocation -// that returns an error is not automatically retried. Setting the same name -// again replaces its method and timing. -// Set returns ErrScheduleStoreUnavailable when no schedule store is -// configured, or the error returned by the store. -func (s Schedule[T]) Set(ctx context.Context, name string, when ScheduleTime, m MethodHandle[T]) error { +// Set creates or replaces the named Reminder for the bound Grain. The new +// first due time is used as FirstTickTime and DueAt. A successful Set persists +// the Reminder. Each due occurrence is delivered at most once, and an +// invocation that returns an error is not automatically retried. +func (s Reminder[T]) Set(ctx context.Context, name string, when ReminderTime, m MethodHandle[T]) error { if s.store == nil { - return ErrScheduleStoreUnavailable + return ErrReminderStoreUnavailable } - return s.store.Put(ctx, store.Schedule{ - GrainId: s.identity, - Name: name, - Method: m.method, - DueAt: s.clock.Now().Add(when.delay), - Interval: when.interval, + firstTickTime := s.clock.Now().Add(when.delay) + return s.store.Put(ctx, store.Reminder{ + GrainId: s.identity, + Name: name, + Method: m.method, + FirstTickTime: firstTickTime, + DueAt: firstTickTime, + Interval: when.interval, }) } -// Cancel asks the schedule store to delete the named schedule for the bound -// entity. Canceling a name that does not exist succeeds as a no-op. It returns -// ErrScheduleStoreUnavailable when no schedule store is configured, or the -// error returned by the store. -func (s Schedule[T]) Cancel(ctx context.Context, name string) error { +// Cancel asks the reminder store to delete the named Reminder for the bound +// Grain. Canceling a name that does not exist succeeds as a no-op. +func (s Reminder[T]) Cancel(ctx context.Context, name string) error { if s.store == nil { - return ErrScheduleStoreUnavailable + return ErrReminderStoreUnavailable } return s.store.Delete(ctx, s.identity, name) } diff --git a/schedule_test.go b/schedule_test.go index 2a3d10d..5398738 100644 --- a/schedule_test.go +++ b/schedule_test.go @@ -15,7 +15,7 @@ import ( type scheduledAccount interface { Arm(context.Context) error - Wake(context.Context) error + Wake(context.Context, TickStatus) error Value(context.Context) (int64, error) } @@ -23,7 +23,9 @@ type scheduledAccountArmRequest struct{} type scheduledAccountArmReply struct{} -type scheduledAccountWakeRequest struct{} +type scheduledAccountWakeRequest struct { + A0 TickStatus +} type scheduledAccountWakeReply struct{} @@ -35,9 +37,10 @@ type scheduledAccountValueReply struct { type scheduledAccountEntity struct { value State[int64] - schedule Schedule[scheduledAccount] + schedule Reminder[scheduledAccount] wakeErr error wakeStarted chan struct{} + wakeCalls *atomic.Int32 cancelShapedErr bool } @@ -50,7 +53,10 @@ func (a *scheduledAccountEntity) Arm(ctx context.Context) error { return a.schedule.Set(ctx, "wake", After(12*time.Second), Handle(scheduledAccount.Wake)) } -func (a *scheduledAccountEntity) Wake(ctx context.Context) error { +func (a *scheduledAccountEntity) Wake(ctx context.Context, _ TickStatus) error { + if a.wakeCalls != nil { + a.wakeCalls.Add(1) + } if a.wakeStarted != nil { close(a.wakeStarted) <-ctx.Done() @@ -75,8 +81,8 @@ func (p *scheduledAccountProxy) Arm(ctx context.Context) error { return p.invoker.Invoke(ctx, p.id, "Arm", &scheduledAccountArmRequest{}, &scheduledAccountArmReply{}) } -func (p *scheduledAccountProxy) Wake(ctx context.Context) error { - return p.invoker.Invoke(ctx, p.id, "Wake", &scheduledAccountWakeRequest{}, &scheduledAccountWakeReply{}) +func (p *scheduledAccountProxy) Wake(ctx context.Context, tick TickStatus) error { + return p.invoker.Invoke(ctx, p.id, "Wake", &scheduledAccountWakeRequest{A0: tick}, &scheduledAccountWakeReply{}) } func (p *scheduledAccountProxy) Value(ctx context.Context) (int64, error) { @@ -85,12 +91,12 @@ func (p *scheduledAccountProxy) Value(ctx context.Context) (int64, error) { return reply.R0, err } -func dispatchScheduledAccount(ctx context.Context, instance scheduledAccount, method string, _ any, reply any) error { +func dispatchScheduledAccount(ctx context.Context, instance scheduledAccount, method string, args any, reply any) error { switch method { case "Arm": return instance.Arm(ctx) case "Wake": - return instance.Wake(ctx) + return instance.Wake(ctx, args.(*scheduledAccountWakeRequest).A0) case "Value": typedReply := reply.(*scheduledAccountValueReply) value, err := instance.Value(ctx) @@ -116,9 +122,17 @@ func newScheduledAccountCall(method string) (args any, reply any) { } } +func newScheduledAccountReminderCall(method string, status TickStatus) (args any, reply any) { + if method == "Wake" { + return &scheduledAccountWakeRequest{A0: status}, &scheduledAccountWakeReply{} + } + return nil, nil +} + type scheduledAccountConfig struct { wakeErr error wakeStarted chan struct{} + wakeCalls *atomic.Int32 cancelShapedErr bool } @@ -130,16 +144,17 @@ func installScheduledAccount(t *testing.T, rt *Runtime, factoryCalls *atomic.Int } if err := InstallType[scheduledAccount](rt, dispatchScheduledAccount, func(invoker Invoker, id GrainId) scheduledAccount { return &scheduledAccountProxy{invoker: invoker, id: id} - }, newScheduledAccountCall); err != nil { + }, newScheduledAccountCall, newScheduledAccountReminderCall); err != nil { t.Fatal(err) } if err := Register[scheduledAccount](rt, func(b *Binder) scheduledAccount { factoryCalls.Add(1) return &scheduledAccountEntity{ value: NewState[int64](b, "value"), - schedule: NewSchedule[scheduledAccount](b), + schedule: NewReminder[scheduledAccount](b), wakeErr: config.wakeErr, wakeStarted: config.wakeStarted, + wakeCalls: config.wakeCalls, cancelShapedErr: config.cancelShapedErr, } }); err != nil { @@ -153,19 +168,20 @@ func TestSchedule_OnErrorReceivesInvocationFailure(t *testing.T) { fakeClock := clock.NewFake(start) backend := store.NewMemory() wakeErr := errors.New("scheduled wake failed") + wakeCalls := new(atomic.Int32) errorsSeen := make(chan BackgroundError, 1) rt := mustNew(t, WithStore(backend), WithClock(fakeClock), WithIdleTimeout(5*time.Second), WithEvictionInterval(time.Second), - WithScheduleInterval(time.Second), + WithReminderInterval(time.Second), OnError(func(event BackgroundError) { errorsSeen <- event }), ) defer rt.Close() - installScheduledAccount(t, rt, new(atomic.Int32), scheduledAccountConfig{wakeErr: wakeErr}) + installScheduledAccount(t, rt, new(atomic.Int32), scheduledAccountConfig{wakeErr: wakeErr, wakeCalls: wakeCalls}) if err := Ref[scheduledAccount](rt, "alice").Arm(context.Background()); err != nil { t.Fatalf("Arm: %v", err) @@ -175,13 +191,16 @@ func TestSchedule_OnErrorReceivesInvocationFailure(t *testing.T) { select { case got := <-errorsSeen: - source, ok := got.Source.(ScheduledInvocation) + source, ok := got.Source.(ReminderInvocation) wantID := GrainId{GrainType: TypeName[scheduledAccount](), GrainKey: "alice"} if !ok || got.GrainId != wantID || source.Method != "Wake" || !errors.Is(got.Err, wakeErr) { - t.Fatalf("OnError event = %#v, want identity %v, ScheduledInvocation{Method: Wake}, error %v", got, wantID, wakeErr) + t.Fatalf("OnError event = %#v, want identity %v, ReminderInvocation{Method: Wake}, error %v", got, wantID, wakeErr) } default: - t.Fatal("OnError did not receive scheduled invocation failure") + t.Fatal("OnError did not receive Reminder invocation failure") + } + if got := wakeCalls.Load(); got != 1 { + t.Fatalf("Wake calls = %d, want 1 without automatic retry", got) } }) } @@ -206,7 +225,7 @@ func TestSchedule_DropsInvocationFailureWithoutOnError(t *testing.T) { t.Fatalf("ListDue: %v", err) } if len(rows) != 0 { - t.Fatalf("schedules after failed one-shot invocation = %#v, want none", rows) + t.Fatalf("reminders after failed one-shot invocation = %#v, want none", rows) } }) } @@ -223,7 +242,7 @@ func TestSchedule_DropsCancellationOnRuntimeClose(t *testing.T) { WithClock(fakeClock), WithIdleTimeout(5*time.Second), WithEvictionInterval(time.Second), - WithScheduleInterval(time.Second), + WithReminderInterval(time.Second), OnError(func(event BackgroundError) { errorsSeen <- event }), @@ -257,7 +276,7 @@ func newScheduledRuntime(t *testing.T, backend store.Store, sourceClock clock.Cl WithClock(sourceClock), WithIdleTimeout(5*time.Second), WithEvictionInterval(time.Second), - WithScheduleInterval(time.Second), + WithReminderInterval(time.Second), ) return rt } @@ -266,19 +285,19 @@ func TestSchedule_SetOverwritesAndCancelDeletes(t *testing.T) { start := time.Unix(0, 0).UTC() fakeClock := clock.NewFake(start) backend := store.NewMemory() - schedule := NewSchedule[scheduledAccount](newTestBinder(GrainId{GrainType: "account", GrainKey: "alice"}, backend, backend, fakeClock)) + schedule := NewReminder[scheduledAccount](newTestBinder(GrainId{GrainType: "account", GrainKey: "alice"}, backend, backend, fakeClock)) if err := schedule.Set(context.Background(), "wake", After(time.Second), Handle(scheduledAccount.Wake)); err != nil { t.Fatal(err) } - if err := schedule.Set(context.Background(), "wake", Every(2*time.Second), Handle(scheduledAccount.Arm)); err != nil { + if err := schedule.Set(context.Background(), "wake", Every(2*time.Second), Handle(scheduledAccount.Wake)); err != nil { t.Fatal(err) } rows, err := backend.ListDue(context.Background(), start.Add(3*time.Second)) if err != nil { t.Fatal(err) } - if len(rows) != 1 || rows[0].Method != "Arm" || rows[0].Interval != 2*time.Second || !rows[0].DueAt.Equal(start.Add(2*time.Second)) { + if len(rows) != 1 || rows[0].Method != "Wake" || rows[0].Interval != 2*time.Second || !rows[0].FirstTickTime.Equal(start.Add(2*time.Second)) || !rows[0].DueAt.Equal(start.Add(2*time.Second)) { t.Fatalf("overwritten schedule = %#v", rows) } @@ -290,38 +309,38 @@ func TestSchedule_SetOverwritesAndCancelDeletes(t *testing.T) { t.Fatal(err) } if len(rows) != 0 { - t.Fatalf("schedules after cancel = %#v, want none", rows) + t.Fatalf("reminders after cancel = %#v, want none", rows) } } -func TestSchedule_ReturnsUnavailableWithoutScheduleStore(t *testing.T) { - schedule := NewSchedule[scheduledAccount](newTestBinder(GrainId{GrainType: "account", GrainKey: "alice"}, failingWriteStore{}, nil, clock.Real{})) - if err := schedule.Set(context.Background(), "wake", After(time.Second), Handle(scheduledAccount.Wake)); !errors.Is(err, ErrScheduleStoreUnavailable) { - t.Fatalf("Set error = %v, want ErrScheduleStoreUnavailable", err) +func TestSchedule_ReturnsUnavailableWithoutReminderStore(t *testing.T) { + schedule := NewReminder[scheduledAccount](newTestBinder(GrainId{GrainType: "account", GrainKey: "alice"}, failingWriteStore{}, nil, clock.Real{})) + if err := schedule.Set(context.Background(), "wake", After(time.Second), Handle(scheduledAccount.Wake)); !errors.Is(err, ErrReminderStoreUnavailable) { + t.Fatalf("Set error = %v, want ErrReminderStoreUnavailable", err) } - if err := schedule.Cancel(context.Background(), "wake"); !errors.Is(err, ErrScheduleStoreUnavailable) { - t.Fatalf("Cancel error = %v, want ErrScheduleStoreUnavailable", err) + if err := schedule.Cancel(context.Background(), "wake"); !errors.Is(err, ErrReminderStoreUnavailable) { + t.Fatalf("Cancel error = %v, want ErrReminderStoreUnavailable", err) } } -func TestNew_ScheduleStoreOptionIsOrderIndependent(t *testing.T) { +func TestNew_ReminderStoreOptionIsOrderIndependent(t *testing.T) { explicit := store.NewMemory() - first := mustNew(t, WithScheduleStore(explicit), WithStore(failingWriteStore{}), WithScheduleInterval(0), WithEvictionInterval(0)) - if first.scheduleStore != explicit { - t.Fatal("WithStore replaced an earlier explicit ScheduleStore") + first := mustNew(t, WithReminderStore(explicit), WithStore(failingWriteStore{}), WithReminderInterval(0), WithEvictionInterval(0)) + if first.reminderStore != explicit { + t.Fatal("WithStore replaced an earlier explicit ReminderStore") } first.Close() - second := mustNew(t, WithStore(failingWriteStore{}), WithScheduleStore(explicit), WithScheduleInterval(0), WithEvictionInterval(0)) - if second.scheduleStore != explicit { - t.Fatal("WithScheduleStore did not replace the default ScheduleStore") + second := mustNew(t, WithStore(failingWriteStore{}), WithReminderStore(explicit), WithReminderInterval(0), WithEvictionInterval(0)) + if second.reminderStore != explicit { + t.Fatal("WithReminderStore did not replace the default ReminderStore") } second.Close() backend := store.NewMemory() - third := mustNew(t, WithStore(backend), WithScheduleInterval(0), WithEvictionInterval(0)) - if third.scheduleStore != backend { - t.Fatal("New did not derive ScheduleStore from Store") + third := mustNew(t, WithStore(backend), WithReminderInterval(0), WithEvictionInterval(0)) + if third.reminderStore != backend { + t.Fatal("New did not derive ReminderStore from Store") } third.Close() } diff --git a/sim/cluster.go b/sim/cluster.go index 3090918..a31c204 100644 --- a/sim/cluster.go +++ b/sim/cluster.go @@ -77,7 +77,7 @@ func (c *simulationCluster) newRuntime(id, generation int) (*gor.Runtime, error) c.tracker, gor.WithClock(c.clock), gor.WithMemberStore(members), - gor.WithScheduleStore(&nodeScheduleStore{backend: c.backend, addr: addr}), + gor.WithReminderStore(&nodeReminderStore{backend: c.backend, addr: addr}), gor.WithNodeAddr(addr), gor.WithGeneration(memberGeneration(id, generation)), gor.WithHeartbeatInterval(simulationStepDuration), diff --git a/sim/cluster_test.go b/sim/cluster_test.go index 357cc0b..96326d3 100644 --- a/sim/cluster_test.go +++ b/sim/cluster_test.go @@ -38,7 +38,7 @@ func (*dualActivationEntity) Disarm(context.Context, string) error { return nil } -func (*dualActivationEntity) Tick(context.Context) error { +func (*dualActivationEntity) Tick(context.Context, gor.TickStatus) error { return nil } diff --git a/sim/cycle_test.go b/sim/cycle_test.go index 8abf7af..6f11fef 100644 --- a/sim/cycle_test.go +++ b/sim/cycle_test.go @@ -67,7 +67,7 @@ func newCycleCallerCall(method string) (args any, reply any) { func installCycleCaller(rt *gor.Runtime) error { if err := gor.InstallType[cycleCaller](rt, dispatchCycleCaller, func(invoker gor.Invoker, id gor.GrainId) cycleCaller { return &cycleCallerProxy{invoker: invoker, id: id} - }, newCycleCallerCall); err != nil { + }, newCycleCallerCall, nil); err != nil { return err } return gor.Register[cycleCaller](rt, func(b *gor.Binder) cycleCaller { diff --git a/sim/fault_test.go b/sim/fault_test.go index f79755b..7a2fb87 100644 --- a/sim/fault_test.go +++ b/sim/fault_test.go @@ -36,7 +36,7 @@ func (*gatedCounterEntity) Disarm(context.Context, string) error { return nil } -func (*gatedCounterEntity) Tick(context.Context) error { +func (*gatedCounterEntity) Tick(context.Context, gor.TickStatus) error { return nil } diff --git a/sim/log.go b/sim/log.go index 9377ab6..f043c85 100644 --- a/sim/log.go +++ b/sim/log.go @@ -77,7 +77,7 @@ func (l *eventLog) addState(id store.GrainId, value int64) { } func (l *eventLog) addScheduleObservation(stats scheduleStats, deliveries int) { - l.add(" observe schedules list-calls=%d claim-won=%d claim-lost=%d deliveries=%d list-errors=%d list-delays=%d claim-errors=%d claim-applied-errors=%d", stats.listCalls, stats.claimWon, stats.claimLost, deliveries, stats.listErrors, stats.listDelays, stats.claimErrors, stats.claimAppliedErrors) + l.add(" observe reminders list-calls=%d claim-won=%d claim-lost=%d deliveries=%d list-errors=%d list-delays=%d claim-errors=%d claim-applied-errors=%d", stats.listCalls, stats.claimWon, stats.claimLost, deliveries, stats.listErrors, stats.listDelays, stats.claimErrors, stats.claimAppliedErrors) } func (l *eventLog) addMemberObservation(stats memberStats) { diff --git a/sim/network.go b/sim/network.go index 00f5d34..aacdce9 100644 --- a/sim/network.go +++ b/sim/network.go @@ -410,28 +410,28 @@ func (s *partitionedMemberStore) ListMembers(ctx context.Context) (store.MemberS return store.MemberSnapshot{Members: members, TableNow: s.backend.memberTableNow()}, nil } -// nodeScheduleStore attributes the poller's list calls to the owning node so +// nodeReminderStore attributes the poller's list calls to the owning node so // the schedule list fault can bind a node target, the same way // partitionedMemberStore attributes member list calls. -type nodeScheduleStore struct { +type nodeReminderStore struct { backend *fakeStore addr string } -var _ store.ScheduleStore = (*nodeScheduleStore)(nil) +var _ store.ReminderStore = (*nodeReminderStore)(nil) -func (s *nodeScheduleStore) ListDue(ctx context.Context, now time.Time) ([]store.Schedule, error) { +func (s *nodeReminderStore) ListDue(ctx context.Context, now time.Time) ([]store.Reminder, error) { return s.backend.listDueFor(ctx, s.addr, now) } -func (s *nodeScheduleStore) Claim(ctx context.Context, schedule store.Schedule, nextDueAt time.Time) (bool, error) { +func (s *nodeReminderStore) Claim(ctx context.Context, schedule store.Reminder, nextDueAt time.Time) (bool, error) { return s.backend.Claim(ctx, schedule, nextDueAt) } -func (s *nodeScheduleStore) Put(ctx context.Context, schedule store.Schedule) error { +func (s *nodeReminderStore) Put(ctx context.Context, schedule store.Reminder) error { return s.backend.Put(ctx, schedule) } -func (s *nodeScheduleStore) Delete(ctx context.Context, id store.GrainId, name string) error { +func (s *nodeReminderStore) Delete(ctx context.Context, id store.GrainId, name string) error { return s.backend.Delete(ctx, id, name) } diff --git a/sim/replay_test.go b/sim/replay_test.go index 057bd4a..bc0cd1a 100644 --- a/sim/replay_test.go +++ b/sim/replay_test.go @@ -166,7 +166,7 @@ func networkStatPositive(log, name string) bool { func scheduleStatPositive(log, name string) bool { for _, line := range strings.Split(log, "\n") { - if !strings.Contains(line, "observe schedules ") { + if !strings.Contains(line, "observe reminders ") { continue } for _, field := range strings.Fields(line) { diff --git a/sim/sim.go b/sim/sim.go index b1d41f6..812a21e 100644 --- a/sim/sim.go +++ b/sim/sim.go @@ -28,7 +28,7 @@ type counter interface { Add(context.Context, int64) (int64, error) Arm(context.Context, string, time.Duration, time.Duration) error Disarm(context.Context, string) error - Tick(context.Context) error + Tick(context.Context, gor.TickStatus) error } type counterAddRequest struct { @@ -53,14 +53,16 @@ type counterDisarmRequest struct { type counterDisarmReply struct{} -type counterTickRequest struct{} +type counterTickRequest struct { + A0 gor.TickStatus +} type counterTickReply struct{} type counterEntity struct { value gor.State[int64] id gor.GrainId - schedule gor.Schedule[counter] + schedule gor.Reminder[counter] tracker *timerTracker } @@ -84,7 +86,7 @@ func (c *counterEntity) Disarm(ctx context.Context, name string) error { return c.schedule.Cancel(ctx, name) } -func (c *counterEntity) Tick(context.Context) error { +func (c *counterEntity) Tick(context.Context, gor.TickStatus) error { c.tracker.deliver(storeIdentity(c.id)) return nil } @@ -108,8 +110,8 @@ func (p *counterProxy) Disarm(ctx context.Context, name string) error { return p.invoker.Invoke(ctx, p.id, "Disarm", &counterDisarmRequest{A0: name}, &counterDisarmReply{}) } -func (p *counterProxy) Tick(ctx context.Context) error { - return p.invoker.Invoke(ctx, p.id, "Tick", &counterTickRequest{}, &counterTickReply{}) +func (p *counterProxy) Tick(ctx context.Context, tick gor.TickStatus) error { + return p.invoker.Invoke(ctx, p.id, "Tick", &counterTickRequest{A0: tick}, &counterTickReply{}) } func dispatchCounter(ctx context.Context, instance counter, method string, args any, reply any) error { @@ -130,7 +132,7 @@ func dispatchCounter(ctx context.Context, instance counter, method string, args typedArgs := args.(*counterDisarmRequest) return instance.Disarm(ctx, typedArgs.A0) case "Tick": - return instance.Tick(ctx) + return instance.Tick(ctx, args.(*counterTickRequest).A0) default: return fmt.Errorf("unknown method %q", method) } @@ -151,10 +153,17 @@ func newCounterCall(method string) (args any, reply any) { } } +func newCounterReminderCall(method string, status gor.TickStatus) (args any, reply any) { + if method == "Tick" { + return &counterTickRequest{A0: status}, &counterTickReply{} + } + return nil, nil +} + func installCounterType(rt *gor.Runtime) error { return gor.InstallType[counter](rt, dispatchCounter, func(invoker gor.Invoker, id gor.GrainId) counter { return &counterProxy{invoker: invoker, id: id} - }, newCounterCall) + }, newCounterCall, newCounterReminderCall) } func registerCounter(rt *gor.Runtime, factory func(*gor.Binder) counter) error { @@ -169,7 +178,7 @@ func installCounterWithTracker(rt *gor.Runtime, tracker *timerTracker) error { return &counterEntity{ value: gor.NewState[int64](b, "value"), id: gor.Self(b), - schedule: gor.NewSchedule[counter](b), + schedule: gor.NewReminder[counter](b), tracker: tracker, } }) @@ -180,7 +189,7 @@ func baseRuntimeOptions(backend *fakeStore) []gor.Option { gor.WithStore(backend), gor.WithIdleTimeout(0), gor.WithEvictionInterval(0), - gor.WithScheduleInterval(simulationStepDuration), + gor.WithReminderInterval(simulationStepDuration), gor.WithMailboxCapacity(4), } } diff --git a/sim/store.go b/sim/store.go index 93dbd40..763d0bc 100644 --- a/sim/store.go +++ b/sim/store.go @@ -117,7 +117,7 @@ var ( errMemberAppliedFailure = errors.New("sim member write applied before failure") ) -type scheduleKey struct { +type reminderKey struct { identity store.GrainId name string } @@ -144,7 +144,7 @@ type fakeStore struct { readBarriers map[store.GrainId]readBarrier members map[fakeMemberKey]store.Member memberFault memberFaultSpec - schedules map[scheduleKey]store.Schedule + reminders map[reminderKey]store.Reminder scheduleListFault scheduleFaultSpec scheduleClaimFaults map[store.GrainId]scheduleFaultKind timerTracker *timerTracker @@ -159,7 +159,7 @@ type fakeStore struct { } var _ store.Store = (*fakeStore)(nil) -var _ store.ScheduleStore = (*fakeStore)(nil) +var _ store.ReminderStore = (*fakeStore)(nil) var _ store.MemberStore = (*fakeStore)(nil) func newFakeStore(tracker *timerTracker) *fakeStore { @@ -170,7 +170,7 @@ func newFakeStore(tracker *timerTracker) *fakeStore { plans: make(map[store.GrainId]faultPlan), readBarriers: make(map[store.GrainId]readBarrier), members: make(map[fakeMemberKey]store.Member), - schedules: make(map[scheduleKey]store.Schedule), + reminders: make(map[reminderKey]store.Reminder), scheduleClaimFaults: make(map[store.GrainId]scheduleFaultKind), timerTracker: tracker, memberClock: clock.Real{}, @@ -408,12 +408,12 @@ func (s *fakeStore) checkMemberStatuses() error { return nil } -func (s *fakeStore) snapshotDue(now time.Time, caller string) ([]store.Schedule, scheduleFaultSpec) { +func (s *fakeStore) snapshotDue(now time.Time, caller string) ([]store.Reminder, scheduleFaultSpec) { s.mu.Lock() defer s.mu.Unlock() s.stats.listCalls++ - result := make([]store.Schedule, 0) - for _, schedule := range s.schedules { + result := make([]store.Reminder, 0) + for _, schedule := range s.reminders { if !schedule.DueAt.After(now) { result = append(result, schedule) } @@ -447,11 +447,11 @@ func (s *fakeStore) snapshotDue(now time.Time, caller string) ([]store.Schedule, return result, fault } -func (s *fakeStore) ListDue(ctx context.Context, now time.Time) ([]store.Schedule, error) { +func (s *fakeStore) ListDue(ctx context.Context, now time.Time) ([]store.Reminder, error) { return s.listDueFor(ctx, "", now) } -func (s *fakeStore) listDueFor(_ context.Context, caller string, now time.Time) ([]store.Schedule, error) { +func (s *fakeStore) listDueFor(_ context.Context, caller string, now time.Time) ([]store.Reminder, error) { defer s.endOperation(s.beginOperation()) result, fault := s.snapshotDue(now, caller) if fault.kind == scheduleListError { @@ -464,10 +464,10 @@ func (s *fakeStore) listDueFor(_ context.Context, caller string, now time.Time) return result, nil } -func (s *fakeStore) Claim(_ context.Context, schedule store.Schedule, nextDueAt time.Time) (bool, error) { +func (s *fakeStore) Claim(_ context.Context, schedule store.Reminder, nextDueAt time.Time) (bool, error) { defer s.endOperation(s.beginOperation()) s.mu.Lock() - current, ok := s.schedules[scheduleKey{identity: schedule.GrainId, name: schedule.Name}] + current, ok := s.reminders[reminderKey{identity: schedule.GrainId, name: schedule.Name}] if !ok || current.ETag != schedule.ETag { s.stats.claimLost++ s.mu.Unlock() @@ -481,11 +481,11 @@ func (s *fakeStore) Claim(_ context.Context, schedule store.Schedule, nextDueAt return false, errScheduleClaimFailure } if nextDueAt.IsZero() { - delete(s.schedules, scheduleKey{identity: schedule.GrainId, name: schedule.Name}) + delete(s.reminders, reminderKey{identity: schedule.GrainId, name: schedule.Name}) } else { current.DueAt = nextDueAt current.ETag++ - s.schedules[scheduleKey{identity: schedule.GrainId, name: schedule.Name}] = current + s.reminders[reminderKey{identity: schedule.GrainId, name: schedule.Name}] = current } s.stats.claimWon++ if fault == scheduleClaimAppliedError { @@ -499,16 +499,16 @@ func (s *fakeStore) Claim(_ context.Context, schedule store.Schedule, nextDueAt return true, nil } -func (s *fakeStore) Put(_ context.Context, schedule store.Schedule) error { +func (s *fakeStore) Put(_ context.Context, schedule store.Reminder) error { defer s.endOperation(s.beginOperation()) s.mu.Lock() - key := scheduleKey{identity: schedule.GrainId, name: schedule.Name} - current, ok := s.schedules[key] + key := reminderKey{identity: schedule.GrainId, name: schedule.Name} + current, ok := s.reminders[key] schedule.ETag = 1 if ok { schedule.ETag = current.ETag + 1 } - s.schedules[key] = schedule + s.reminders[key] = schedule s.mu.Unlock() return nil } @@ -516,7 +516,7 @@ func (s *fakeStore) Put(_ context.Context, schedule store.Schedule) error { func (s *fakeStore) Delete(_ context.Context, id store.GrainId, name string) error { defer s.endOperation(s.beginOperation()) s.mu.Lock() - delete(s.schedules, scheduleKey{identity: id, name: name}) + delete(s.reminders, reminderKey{identity: id, name: name}) s.mu.Unlock() return nil } diff --git a/state.go b/state.go index 537e3de..e8b9cd6 100644 --- a/state.go +++ b/state.go @@ -10,7 +10,7 @@ import ( ) // Binder is the runtime-bound context passed to an entity factory. Entity code -// uses the Binder to create state, schedules, and references; application code +// uses the Binder to create state, reminders, and references; application code // should use the supplied Binder rather than construct one. type Binder struct { runtime *Runtime diff --git a/state_test.go b/state_test.go index dbcf688..abe8edc 100644 --- a/state_test.go +++ b/state_test.go @@ -10,8 +10,8 @@ import ( "github.com/suraciii/gor/store" ) -func newTestBinder(id GrainId, backend store.Store, schedules store.ScheduleStore, sourceClock clock.Clock) *Binder { - return newBinder(&Runtime{store: backend, scheduleStore: schedules, clock: sourceClock}, id) +func newTestBinder(id GrainId, backend store.Store, reminders store.ReminderStore, sourceClock clock.Clock) *Binder { + return newBinder(&Runtime{store: backend, reminderStore: reminders, clock: sourceClock}, id) } func TestState_PersistsAllRegisteredValuesAsOneRecord(t *testing.T) { diff --git a/store/durability_test.go b/store/durability_test.go index ba973db..a0f0bdd 100644 --- a/store/durability_test.go +++ b/store/durability_test.go @@ -87,7 +87,7 @@ func TestOpenSQLiteRelaxed_WritesPersistAcrossCloseAndFlush(t *testing.T) { if _, err := s.Write(context.Background(), id, []byte("stale"), first); !errors.Is(err, ErrConflict) { t.Fatalf("stale Write error = %v, want ErrConflict", err) } - if err := s.Put(context.Background(), Schedule{ + if err := s.Put(context.Background(), Reminder{ GrainId: GrainId{GrainType: "account", GrainKey: "alice"}, Name: "tick", Method: "Tick", diff --git a/store/migration_test.go b/store/migration_test.go index ebe9d3c..7f904f4 100644 --- a/store/migration_test.go +++ b/store/migration_test.go @@ -174,18 +174,18 @@ func TestMigrate_OldDatabaseReadsBackEveryConfirmedState(t *testing.T) { } } - schedules, err := s.ListDue(context.Background(), time.Unix(0, 100000).UTC()) + reminders, err := s.ListDue(context.Background(), time.Unix(0, 100000).UTC()) if err != nil { t.Fatalf("ListDue: %v", err) } - if len(schedules) != 2 { - t.Fatalf("ListDue returned %d schedules, want 2", len(schedules)) + if len(reminders) != 2 { + t.Fatalf("ListDue returned %d reminders, want 2", len(reminders)) } - if schedules[0].GrainId != (GrainId{GrainType: "account", GrainKey: "bob"}) || schedules[0].Name != "renew" || schedules[0].Method != "Renew" || schedules[0].ETag != 3 { - t.Fatalf("schedule[0] = %#v, want bob/renew/3", schedules[0]) + if reminders[0].GrainId != (GrainId{GrainType: "account", GrainKey: "bob"}) || reminders[0].Name != "renew" || reminders[0].Method != "Renew" || reminders[0].ETag != 3 || !reminders[0].FirstTickTime.Equal(reminders[0].DueAt) { + t.Fatalf("reminder[0] = %#v, want bob/renew/3 with FirstTickTime equal to DueAt", reminders[0]) } - if schedules[1].GrainId != (GrainId{GrainType: "account", GrainKey: "alice"}) || schedules[1].Name != "tick" || schedules[1].Method != "Tick" || schedules[1].ETag != 1 { - t.Fatalf("schedule[1] = %#v, want alice/tick/1", schedules[1]) + if reminders[1].GrainId != (GrainId{GrainType: "account", GrainKey: "alice"}) || reminders[1].Name != "tick" || reminders[1].Method != "Tick" || reminders[1].ETag != 1 || !reminders[1].FirstTickTime.Equal(reminders[1].DueAt) { + t.Fatalf("reminder[1] = %#v, want alice/tick/1 with FirstTickTime equal to DueAt", reminders[1]) } members, err := s.ListMembers(context.Background()) @@ -223,6 +223,13 @@ func TestMigrate_OldDatabaseReadsBackEveryConfirmedState(t *testing.T) { if string(record.Data) != "data-0" || record.ETag != 5 { t.Fatalf("record after reopen = %#v, want data-0 etag 5", record) } + reopenedReminders, err := reopened.ListDue(context.Background(), time.Unix(0, 100000).UTC()) + if err != nil { + t.Fatalf("ListDue after reopen: %v", err) + } + if len(reopenedReminders) != 2 || !reopenedReminders[0].FirstTickTime.Equal(reopenedReminders[0].DueAt) || !reopenedReminders[1].FirstTickTime.Equal(reopenedReminders[1].DueAt) { + t.Fatalf("reminders after reopen = %#v, want fallback FirstTickTime values", reopenedReminders) + } }) } } @@ -338,12 +345,12 @@ func TestMigrate_OldDatabaseWithoutStateRows(t *testing.T) { } defer s.Close() - schedules, err := s.ListDue(context.Background(), time.Unix(0, 100000).UTC()) + reminders, err := s.ListDue(context.Background(), time.Unix(0, 100000).UTC()) if err != nil { t.Fatalf("ListDue: %v", err) } - if len(schedules) != 2 { - t.Fatalf("ListDue returned %d schedules, want 2", len(schedules)) + if len(reminders) != 2 { + t.Fatalf("ListDue returned %d reminders, want 2", len(reminders)) } members, err := s.ListMembers(context.Background()) if err != nil { diff --git a/store/schedule.go b/store/schedule.go index f4da092..e680c92 100644 --- a/store/schedule.go +++ b/store/schedule.go @@ -6,130 +6,132 @@ import ( "time" ) -// Schedule describes one persisted invocation deadline. +// Reminder describes one persisted Reminder deadline. // // GrainId and Name identify the row. DueAt is inclusive when queried by -// ListDue. Interval is retained for the scheduler, and ETag is the version -// used by Claim. -type Schedule struct { - GrainId GrainId - Name string - Method string - DueAt time.Time - Interval time.Duration - ETag ETag +// ListDue. FirstTickTime is the first due time for the current setting. +// Interval is retained for the poller, and ETag is the version used by Claim. +type Reminder struct { + GrainId GrainId + Name string + Method string + FirstTickTime time.Time + DueAt time.Time + Interval time.Duration + ETag ETag } -// ScheduleStore persists schedules and atomically claims due rows. +// ReminderStore persists Reminders and atomically claims due rows. // // Implementations must support concurrent calls. Claim must compare the row's // identity, name, and ETag atomically so concurrent claimers using one // snapshot produce at most one winner. Put and Delete are unconditional // changes by design. -type ScheduleStore interface { - // ListDue returns every schedule whose DueAt is no later than now. - ListDue(context.Context, time.Time) ([]Schedule, error) - // Claim compares the schedule's identity, name, and ETag atomically. When +type ReminderStore interface { + // ListDue returns every Reminder whose DueAt is no later than now. + ListDue(context.Context, time.Time) ([]Reminder, error) + // Claim compares the Reminder identity, name, and ETag atomically. When // they match, a non-zero nextDueAt replaces DueAt and increments the stored - // ETag; a zero nextDueAt deletes the schedule. It returns true only when the + // ETag; a zero nextDueAt deletes the Reminder. It returns true only when the // update or deletion succeeds, and returns false with a nil error when the // row is absent or its ETag is stale. - Claim(context.Context, Schedule, time.Time) (bool, error) - // Put inserts or replaces a schedule without an ETag precondition. A new + Claim(context.Context, Reminder, time.Time) (bool, error) + // Put inserts or replaces a Reminder without an ETag precondition. A new // row receives ETag 1; replacing an existing row increments its current // ETag. The input ETag is ignored. - Put(context.Context, Schedule) error - // Delete unconditionally removes the named schedule, if it exists. + Put(context.Context, Reminder) error + // Delete unconditionally removes the named Reminder, if it exists. Delete(context.Context, GrainId, string) error } -type scheduleKey struct { +type reminderKey struct { identity GrainId name string } -func keyForSchedule(schedule Schedule) scheduleKey { - return scheduleKey{identity: schedule.GrainId, name: schedule.Name} +func keyForReminder(reminder Reminder) reminderKey { + return reminderKey{identity: reminder.GrainId, name: reminder.Name} } -func sortSchedules(schedules []Schedule) { - sort.Slice(schedules, func(i, j int) bool { - if schedules[i].DueAt != schedules[j].DueAt { - return schedules[i].DueAt.Before(schedules[j].DueAt) +func sortReminders(reminders []Reminder) { + sort.Slice(reminders, func(i, j int) bool { + if reminders[i].DueAt != reminders[j].DueAt { + return reminders[i].DueAt.Before(reminders[j].DueAt) } - if schedules[i].GrainId.GrainType != schedules[j].GrainId.GrainType { - return schedules[i].GrainId.GrainType < schedules[j].GrainId.GrainType + if reminders[i].GrainId.GrainType != reminders[j].GrainId.GrainType { + return reminders[i].GrainId.GrainType < reminders[j].GrainId.GrainType } - if schedules[i].GrainId.GrainKey != schedules[j].GrainId.GrainKey { - return schedules[i].GrainId.GrainKey < schedules[j].GrainId.GrainKey + if reminders[i].GrainId.GrainKey != reminders[j].GrainId.GrainKey { + return reminders[i].GrainId.GrainKey < reminders[j].GrainId.GrainKey } - return schedules[i].Name < schedules[j].Name + return reminders[i].Name < reminders[j].Name }) } -// ListDue returns due schedules in deterministic DueAt, identity, and name +// ListDue returns due Reminders in deterministic DueAt, identity, and name // order. -func (m *Memory) ListDue(ctx context.Context, now time.Time) ([]Schedule, error) { +func (m *Memory) ListDue(ctx context.Context, now time.Time) ([]Reminder, error) { if err := ctx.Err(); err != nil { return nil, err } m.mu.RLock() defer m.mu.RUnlock() - result := make([]Schedule, 0) - for _, schedule := range m.schedules { - if !schedule.DueAt.After(now) { - result = append(result, schedule) + result := make([]Reminder, 0) + for _, reminder := range m.reminders { + if !reminder.DueAt.After(now) { + result = append(result, reminder) } } - sortSchedules(result) + sortReminders(result) return result, nil } -// Claim atomically checks schedule's identity, name, and ETag. It returns true -// and advances the row when nextDueAt is non-zero, or deletes the row when -// nextDueAt is zero. It returns false and nil when the row is absent or stale. -func (m *Memory) Claim(ctx context.Context, schedule Schedule, nextDueAt time.Time) (bool, error) { +// Claim atomically checks the Reminder identity, name, and ETag. It returns +// true and advances the row when nextDueAt is non-zero, or deletes the row +// when nextDueAt is zero. It returns false and nil when the row is absent or +// stale. +func (m *Memory) Claim(ctx context.Context, reminder Reminder, nextDueAt time.Time) (bool, error) { if err := ctx.Err(); err != nil { return false, err } m.mu.Lock() defer m.mu.Unlock() - key := keyForSchedule(schedule) - current, ok := m.schedules[key] - if !ok || current.ETag != schedule.ETag { + key := keyForReminder(reminder) + current, ok := m.reminders[key] + if !ok || current.ETag != reminder.ETag { return false, nil } if nextDueAt.IsZero() { - delete(m.schedules, key) + delete(m.reminders, key) return true, nil } current.DueAt = nextDueAt current.ETag++ - m.schedules[key] = current + m.reminders[key] = current return true, nil } -// Put unconditionally inserts or replaces a schedule and assigns a new ETag. -func (m *Memory) Put(ctx context.Context, schedule Schedule) error { +// Put unconditionally inserts or replaces a Reminder and assigns a new ETag. +func (m *Memory) Put(ctx context.Context, reminder Reminder) error { if err := ctx.Err(); err != nil { return err } m.mu.Lock() defer m.mu.Unlock() - key := keyForSchedule(schedule) - current, ok := m.schedules[key] - schedule.ETag = 1 + key := keyForReminder(reminder) + current, ok := m.reminders[key] + reminder.ETag = 1 if ok { - schedule.ETag = current.ETag + 1 + reminder.ETag = current.ETag + 1 } - m.schedules[key] = schedule + m.reminders[key] = reminder return nil } -// Delete unconditionally removes the schedule identified by id and name. +// Delete unconditionally removes the Reminder identified by id and name. func (m *Memory) Delete(ctx context.Context, id GrainId, name string) error { if err := ctx.Err(); err != nil { return err @@ -137,6 +139,6 @@ func (m *Memory) Delete(ctx context.Context, id GrainId, name string) error { m.mu.Lock() defer m.mu.Unlock() - delete(m.schedules, scheduleKey{identity: id, name: name}) + delete(m.reminders, reminderKey{identity: id, name: name}) return nil } diff --git a/store/schedule_sqlite.go b/store/schedule_sqlite.go index 214bf03..78249e6 100644 --- a/store/schedule_sqlite.go +++ b/store/schedule_sqlite.go @@ -6,13 +6,13 @@ import ( "time" ) -var _ ScheduleStore = (*SQLite)(nil) +var _ ReminderStore = (*SQLite)(nil) -// ListDue returns due schedules in deterministic DueAt, identity, and name +// ListDue returns due Reminders in deterministic DueAt, identity, and name // order. -func (s *SQLite) ListDue(ctx context.Context, now time.Time) ([]Schedule, error) { +func (s *SQLite) ListDue(ctx context.Context, now time.Time) ([]Reminder, error) { rows, err := s.readDB.QueryContext(ctx, ` -SELECT entity_type, entity_key, name, method, due_at, interval, etag +SELECT entity_type, entity_key, name, method, first_tick_time, due_at, interval, etag FROM schedule WHERE due_at <= ? ORDER BY due_at, entity_type, entity_key, name`, timeValue(now)) @@ -21,27 +21,29 @@ ORDER BY due_at, entity_type, entity_key, name`, timeValue(now)) } defer rows.Close() - result := make([]Schedule, 0) + result := make([]Reminder, 0) for rows.Next() { var ( entityType string entityKey string name string method string + firstTick int64 dueAt int64 interval int64 etag int64 ) - if err := rows.Scan(&entityType, &entityKey, &name, &method, &dueAt, &interval, &etag); err != nil { + if err := rows.Scan(&entityType, &entityKey, &name, &method, &firstTick, &dueAt, &interval, &etag); err != nil { return nil, err } - result = append(result, Schedule{ - GrainId: GrainId{GrainType: entityType, GrainKey: entityKey}, - Name: name, - Method: method, - DueAt: timeFromValue(dueAt), - Interval: time.Duration(interval), - ETag: ETag(etag), + result = append(result, Reminder{ + GrainId: GrainId{GrainType: entityType, GrainKey: entityKey}, + Name: name, + Method: method, + FirstTickTime: timeFromValue(firstTick), + DueAt: timeFromValue(dueAt), + Interval: time.Duration(interval), + ETag: ETag(etag), }) } if err := rows.Err(); err != nil { @@ -50,10 +52,11 @@ ORDER BY due_at, entity_type, entity_key, name`, timeValue(now)) return result, nil } -// Claim atomically checks schedule's identity, name, and ETag. It returns true -// and advances the row when nextDueAt is non-zero, or deletes the row when -// nextDueAt is zero. It returns false and nil when the row is absent or stale. -func (s *SQLite) Claim(ctx context.Context, schedule Schedule, nextDueAt time.Time) (bool, error) { +// Claim atomically checks the Reminder identity, name, and ETag. It returns +// true and advances the row when nextDueAt is non-zero, or deletes the row +// when nextDueAt is zero. It returns false and nil when the row is absent or +// stale. +func (s *SQLite) Claim(ctx context.Context, reminder Reminder, nextDueAt time.Time) (bool, error) { var ( result sql.Result err error @@ -62,10 +65,10 @@ func (s *SQLite) Claim(ctx context.Context, schedule Schedule, nextDueAt time.Ti result, err = s.writeDB.ExecContext(ctx, ` DELETE FROM schedule WHERE entity_type = ? AND entity_key = ? AND name = ? AND etag = ?`, - schedule.GrainId.GrainType, - schedule.GrainId.GrainKey, - schedule.Name, - int64(schedule.ETag), + reminder.GrainId.GrainType, + reminder.GrainId.GrainKey, + reminder.Name, + int64(reminder.ETag), ) } else { result, err = s.writeDB.ExecContext(ctx, ` @@ -73,10 +76,10 @@ UPDATE schedule SET due_at = ?, etag = etag + 1 WHERE entity_type = ? AND entity_key = ? AND name = ? AND etag = ?`, timeValue(nextDueAt), - schedule.GrainId.GrainType, - schedule.GrainId.GrainKey, - schedule.Name, - int64(schedule.ETag), + reminder.GrainId.GrainType, + reminder.GrainId.GrainKey, + reminder.Name, + int64(reminder.ETag), ) } if err != nil { @@ -89,27 +92,29 @@ WHERE entity_type = ? AND entity_key = ? AND name = ? AND etag = ?`, return rows == 1, nil } -// Put unconditionally inserts or replaces a schedule and assigns a new ETag. -func (s *SQLite) Put(ctx context.Context, schedule Schedule) error { +// Put unconditionally inserts or replaces a Reminder and assigns a new ETag. +func (s *SQLite) Put(ctx context.Context, reminder Reminder) error { _, err := s.writeDB.ExecContext(ctx, ` -INSERT INTO schedule (entity_type, entity_key, name, method, due_at, interval, etag) -VALUES (?, ?, ?, ?, ?, ?, 1) +INSERT INTO schedule (entity_type, entity_key, name, method, first_tick_time, due_at, interval, etag) +VALUES (?, ?, ?, ?, ?, ?, ?, 1) ON CONFLICT (entity_type, entity_key, name) DO UPDATE SET - method = excluded.method, - due_at = excluded.due_at, - interval = excluded.interval, - etag = schedule.etag + 1`, - schedule.GrainId.GrainType, - schedule.GrainId.GrainKey, - schedule.Name, - schedule.Method, - timeValue(schedule.DueAt), - int64(schedule.Interval), +method = excluded.method, +first_tick_time = excluded.first_tick_time, +due_at = excluded.due_at, +interval = excluded.interval, +etag = schedule.etag + 1`, + reminder.GrainId.GrainType, + reminder.GrainId.GrainKey, + reminder.Name, + reminder.Method, + timeValue(reminder.FirstTickTime), + timeValue(reminder.DueAt), + int64(reminder.Interval), ) return err } -// Delete unconditionally removes the schedule identified by id and name. +// Delete unconditionally removes the Reminder identified by id and name. func (s *SQLite) Delete(ctx context.Context, id GrainId, name string) error { _, err := s.writeDB.ExecContext(ctx, ` DELETE FROM schedule diff --git a/store/schedule_test.go b/store/schedule_test.go index f7496c4..d92622c 100644 --- a/store/schedule_test.go +++ b/store/schedule_test.go @@ -2,30 +2,68 @@ package store import ( "context" + "path/filepath" "sync" "testing" "time" ) -func TestMemoryScheduleStore(t *testing.T) { - runScheduleStoreTests(t, NewMemory()) +func TestMemoryReminderStore(t *testing.T) { + runReminderStoreTests(t, NewMemory()) } -func TestSQLiteScheduleStore(t *testing.T) { - runScheduleStoreTests(t, newSQLiteTestStore(t)) +func TestSQLiteReminderStore(t *testing.T) { + runReminderStoreTests(t, newSQLiteTestStore(t)) } -func runScheduleStoreTests(t *testing.T, backend ScheduleStore) { +func TestSQLiteReminderStore_PreservesFirstTickAfterReopen(t *testing.T) { + path := filepath.Join(t.TempDir(), "reminders.db") + first, err := OpenSQLite(path) + if err != nil { + t.Fatalf("OpenSQLite first: %v", err) + } + row := Reminder{ + GrainId: GrainId{GrainType: "account", GrainKey: "alice"}, + Name: "wake", + Method: "Wake", + FirstTickTime: time.Unix(10, 0).UTC(), + DueAt: time.Unix(20, 0).UTC(), + Interval: time.Hour, + } + if err := first.Put(context.Background(), row); err != nil { + first.Close() + t.Fatalf("Put: %v", err) + } + if err := first.Close(); err != nil { + t.Fatalf("Close first: %v", err) + } + + second, err := OpenSQLite(path) + if err != nil { + t.Fatalf("OpenSQLite second: %v", err) + } + defer second.Close() + got, err := second.ListDue(context.Background(), row.DueAt) + if err != nil { + t.Fatalf("ListDue after reopen: %v", err) + } + if len(got) != 1 || !got[0].FirstTickTime.Equal(row.FirstTickTime) || !got[0].DueAt.Equal(row.DueAt) { + t.Fatalf("row after reopen = %#v, want FirstTickTime %s and DueAt %s", got, row.FirstTickTime, row.DueAt) + } +} + +func runReminderStoreTests(t *testing.T, backend ReminderStore) { t.Helper() t.Run("WriteAndListDue", func(t *testing.T) { ctx := context.Background() now := time.Unix(100, 0).UTC() - due := Schedule{ - GrainId: GrainId{GrainType: "account", GrainKey: "alice"}, - Name: "wake", - Method: "Wake", - DueAt: now.Add(-time.Second), - Interval: time.Hour, + due := Reminder{ + GrainId: GrainId{GrainType: "account", GrainKey: "alice"}, + Name: "wake", + Method: "Wake", + FirstTickTime: now.Add(-2 * time.Second), + DueAt: now.Add(-time.Second), + Interval: time.Hour, } if err := backend.Put(ctx, due); err != nil { t.Fatalf("Put due: %v", err) @@ -44,12 +82,13 @@ func runScheduleStoreTests(t *testing.T, backend ScheduleStore) { if len(got) != 1 { t.Fatalf("ListDue returned %d rows, want 1", len(got)) } - if got[0].GrainId != due.GrainId || got[0].Name != due.Name || got[0].Method != due.Method || !got[0].DueAt.Equal(due.DueAt) || got[0].Interval != due.Interval || got[0].ETag != 1 { + if got[0].GrainId != due.GrainId || got[0].Name != due.Name || got[0].Method != due.Method || !got[0].FirstTickTime.Equal(due.FirstTickTime) || !got[0].DueAt.Equal(due.DueAt) || got[0].Interval != due.Interval || got[0].ETag != 1 { t.Fatalf("due row = %#v, want %#v with ETag 1", got[0], due) } replacement := due replacement.Method = "WakeAgain" + replacement.FirstTickTime = now.Add(30 * time.Minute) replacement.DueAt = now.Add(time.Hour) if err := backend.Put(ctx, replacement); err != nil { t.Fatalf("Put replacement: %v", err) @@ -72,7 +111,7 @@ func runScheduleStoreTests(t *testing.T, backend ScheduleStore) { if err != nil { t.Fatalf("ListDue at replacement: %v", err) } - if len(got) != 1 || got[0].Method != replacement.Method || got[0].ETag != 2 { + if len(got) != 1 || got[0].Method != replacement.Method || !got[0].FirstTickTime.Equal(replacement.FirstTickTime) || got[0].ETag != 2 { t.Fatalf("replacement rows = %#v, want method %q and ETag 2", got, replacement.Method) } }) @@ -80,7 +119,7 @@ func runScheduleStoreTests(t *testing.T, backend ScheduleStore) { t.Run("ClaimCASAllowsExactlyOneWinner", func(t *testing.T) { ctx := context.Background() now := time.Unix(200, 0).UTC() - task := Schedule{ + task := Reminder{ GrainId: GrainId{GrainType: "account", GrainKey: "bob"}, Name: "wake", Method: "Wake", @@ -137,7 +176,7 @@ func runScheduleStoreTests(t *testing.T, backend ScheduleStore) { if err != nil { t.Fatalf("ListDue after Claim: %v", err) } - var claimed Schedule + var claimed Reminder found := false for _, candidate := range got { if candidate.GrainId == task.GrainId && candidate.Name == task.Name { @@ -146,7 +185,7 @@ func runScheduleStoreTests(t *testing.T, backend ScheduleStore) { break } } - if !found || claimed.ETag != task.ETag+1 || !claimed.DueAt.Equal(nextDueAt) { + if !found || claimed.ETag != task.ETag+1 || !claimed.FirstTickTime.Equal(task.FirstTickTime) || !claimed.DueAt.Equal(nextDueAt) { t.Fatalf("claimed row = %#v, want %s/%s at next due time and ETag %d", got, task.GrainId.GrainType, task.Name, task.ETag+1) } }) @@ -154,7 +193,7 @@ func runScheduleStoreTests(t *testing.T, backend ScheduleStore) { t.Run("OneShotClaimDeletes", func(t *testing.T) { ctx := context.Background() now := time.Unix(300, 0).UTC() - task := Schedule{ + task := Reminder{ GrainId: GrainId{GrainType: "account", GrainKey: "carol"}, Name: "once", Method: "Wake", @@ -186,7 +225,7 @@ func runScheduleStoreTests(t *testing.T, backend ScheduleStore) { t.Run("DeleteIsUnconditional", func(t *testing.T) { ctx := context.Background() now := time.Unix(400, 0).UTC() - task := Schedule{ + task := Reminder{ GrainId: GrainId{GrainType: "account", GrainKey: "dave"}, Name: "cancel", Method: "Wake", diff --git a/store/sqlite.go b/store/sqlite.go index afc6d36..7cfac64 100644 --- a/store/sqlite.go +++ b/store/sqlite.go @@ -19,7 +19,7 @@ const sqliteBusyTimeout = 5000 // crash (power loss, operating-system crash, hard reset). A clean restart // loses nothing at either tier. // -// The tier applies to entity state only; the schedule and membership tables +// The tier applies to Grain State only; the Reminder and membership tables // always run at DurabilityFull. type Durability int @@ -50,7 +50,7 @@ func WithDurability(d Durability) Option { } // SQLite is a SQLite-backed implementation of Store, MemberStore, and -// ScheduleStore. +// ReminderStore. // // Entity state lives in a database file derived from the named path by // inserting "-state" before the file extension; the schedule and membership @@ -110,6 +110,10 @@ func openSQLite(path string, memberClock clock.Clock, opts ...Option) (*SQLite, writeDB.Close() return nil, err } + if err := migrateReminderSchema(writeDB); err != nil { + writeDB.Close() + return nil, err + } readDB, err := sql.Open("sqlite", sqliteDSN(path, DurabilityFull)) if err != nil { @@ -283,6 +287,7 @@ CREATE TABLE IF NOT EXISTS schedule ( entity_key TEXT NOT NULL, name TEXT NOT NULL, method TEXT NOT NULL, + first_tick_time INTEGER NOT NULL, due_at INTEGER NOT NULL, interval INTEGER NOT NULL, etag INTEGER NOT NULL, @@ -301,6 +306,42 @@ CREATE TABLE IF NOT EXISTS member ( return err } +func migrateReminderSchema(db *sql.DB) error { + rows, err := db.Query(`PRAGMA table_info(schedule)`) + if err != nil { + return err + } + defer rows.Close() + + found := false + for rows.Next() { + var ( + cid int + name string + columnType string + notNull int + defaultVal any + primaryKey int + ) + if err := rows.Scan(&cid, &name, &columnType, ¬Null, &defaultVal, &primaryKey); err != nil { + return err + } + if name == "first_tick_time" { + found = true + } + } + if err := rows.Err(); err != nil { + return err + } + if !found { + if _, err := db.Exec(`ALTER TABLE schedule ADD COLUMN first_tick_time INTEGER`); err != nil { + return err + } + } + _, err = db.Exec(`UPDATE schedule SET first_tick_time = due_at WHERE first_tick_time IS NULL`) + return err +} + func createStateSchema(db *sql.DB) error { _, err := db.Exec(` CREATE TABLE IF NOT EXISTS records ( diff --git a/store/store.go b/store/store.go index a9fe549..32742be 100644 --- a/store/store.go +++ b/store/store.go @@ -2,7 +2,7 @@ // in-memory and SQLite implementations. // // Store implementations are part of the supported extension surface. They -// persist entity state and coordinate membership and schedules through +// persist Grain State and coordinate membership and Reminders through // compare-and-swap operations. package store @@ -21,7 +21,7 @@ type GrainId struct { GrainKey string } -// ETag identifies the version of a record, member, or schedule row. +// ETag identifies the version of a record, member, or Reminder row. // The zero value means that the row must not exist when it is first written. type ETag int64 @@ -64,17 +64,17 @@ func timeFromValue(value int64) time.Time { } // Memory is an in-memory implementation of Store, MemberStore, and -// ScheduleStore. +// ReminderStore. type Memory struct { mu sync.RWMutex records map[GrainId]Record - schedules map[scheduleKey]Schedule + reminders map[reminderKey]Reminder members map[memberKey]Member memberClock clock.Clock } var _ Store = (*Memory)(nil) -var _ ScheduleStore = (*Memory)(nil) +var _ ReminderStore = (*Memory)(nil) var _ MemberStore = (*Memory)(nil) // NewMemory returns an empty in-memory store. @@ -88,7 +88,7 @@ func NewMemory(memberClocks ...clock.Clock) *Memory { } return &Memory{ records: make(map[GrainId]Record), - schedules: make(map[scheduleKey]Schedule), + reminders: make(map[reminderKey]Reminder), members: make(map[memberKey]Member), memberClock: memberClock, } diff --git a/store/store_test.go b/store/store_test.go index 9bbba82..cc6bff2 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -166,9 +166,9 @@ func TestMemoryStore_MethodsHonorCanceledContext(t *testing.T) { wantCanceled(err) _, err = memory.ListDue(ctx, time.Time{}) wantCanceled(err) - _, err = memory.Claim(ctx, Schedule{}, time.Time{}) + _, err = memory.Claim(ctx, Reminder{}, time.Time{}) wantCanceled(err) - err = memory.Put(ctx, Schedule{}) + err = memory.Put(ctx, Reminder{}) wantCanceled(err) err = memory.Delete(ctx, GrainId{GrainType: "account", GrainKey: "alice"}, "daily") wantCanceled(err) diff --git a/timer/timer.go b/timer/timer.go index 1e47ec9..cd7d414 100644 --- a/timer/timer.go +++ b/timer/timer.go @@ -1,8 +1,8 @@ -// Package timer polls persisted schedules and delivers due entity calls for +// Package timer polls persisted Reminders and delivers due Grain Calls for // gor. // // It is an implementation package, not an application dependency. Create and -// manage schedules through the root gor package's Schedule APIs instead of +// manage Reminders through the root gor package's Reminder APIs instead of // importing timer directly. package timer @@ -15,33 +15,40 @@ import ( ) type Table interface { - ListDue(context.Context, time.Time) ([]store.Schedule, error) - Claim(context.Context, store.Schedule, time.Time) (bool, error) + ListDue(context.Context, time.Time) ([]store.Reminder, error) + Claim(context.Context, store.Reminder, time.Time) (bool, error) } type Invoker interface { Owns(store.GrainId) bool - Invoke(context.Context, store.GrainId, string) error + Invoke(context.Context, store.GrainId, string, any, any) error } +// ReminderCallFactory creates the normal typed request and reply values for a +// claimed Reminder. The root package converts these time values into its +// public TickStatus before calling generated code. +type ReminderCallFactory func(store.GrainId, string, time.Time, time.Duration, time.Time) (any, any) + type Poller struct { table Table clock clock.Clock interval time.Duration invoker Invoker + newCall ReminderCallFactory ctx context.Context cancel context.CancelFunc done chan struct{} } -func New(table Table, clock clock.Clock, interval time.Duration, invoker Invoker) *Poller { +func New(table Table, clock clock.Clock, interval time.Duration, invoker Invoker, newCall ReminderCallFactory) *Poller { ctx, cancel := context.WithCancel(context.Background()) poller := &Poller{ table: table, clock: clock, interval: interval, invoker: invoker, + newCall: newCall, ctx: ctx, cancel: cancel, done: make(chan struct{}), @@ -73,31 +80,37 @@ func (p *Poller) run(ticker clock.Ticker) { func (p *Poller) poll() { now := p.clock.Now() - schedules, err := p.table.ListDue(p.ctx, now) + reminders, err := p.table.ListDue(p.ctx, now) if err != nil { return } - for _, schedule := range schedules { + for _, reminder := range reminders { if p.ctx.Err() != nil { return } - if !p.invoker.Owns(schedule.GrainId) { + if !p.invoker.Owns(reminder.GrainId) { continue } - nextDueAt := nextDueAt(schedule, now) - claimed, err := p.table.Claim(p.ctx, schedule, nextDueAt) + nextDueAt := nextDueAt(reminder, now) + claimed, err := p.table.Claim(p.ctx, reminder, nextDueAt) if err != nil || !claimed { continue } - _ = p.invoker.Invoke(p.ctx, schedule.GrainId, schedule.Method) + if p.newCall == nil { + continue + } + args, reply := p.newCall(reminder.GrainId, reminder.Method, reminder.FirstTickTime, reminder.Interval, reminder.DueAt) + _ = p.invoker.Invoke(p.ctx, reminder.GrainId, reminder.Method, args, reply) } } -func nextDueAt(schedule store.Schedule, now time.Time) time.Time { - if schedule.Interval == 0 { +func nextDueAt(reminder store.Reminder, now time.Time) time.Time { + if reminder.Interval <= 0 { return time.Time{} } - elapsed := now.Sub(schedule.DueAt) - steps := elapsed/schedule.Interval + 1 - return schedule.DueAt.Add(steps * schedule.Interval) + next := reminder.DueAt.Add(reminder.Interval) + for !next.After(now) { + next = next.Add(reminder.Interval) + } + return next } diff --git a/timer/timer_test.go b/timer/timer_test.go index b7b0db0..1f8211d 100644 --- a/timer/timer_test.go +++ b/timer/timer_test.go @@ -14,19 +14,19 @@ import ( ) type fakeTable struct { - rows []store.Schedule + rows []store.Reminder claimWon bool recorder *stepRecorder nextDueAt []time.Time } -func (t *fakeTable) ListDue(context.Context, time.Time) ([]store.Schedule, error) { +func (t *fakeTable) ListDue(context.Context, time.Time) ([]store.Reminder, error) { t.recorder.record("list") - return append([]store.Schedule(nil), t.rows...), nil + return append([]store.Reminder(nil), t.rows...), nil } -func (t *fakeTable) Claim(_ context.Context, schedule store.Schedule, nextDueAt time.Time) (bool, error) { +func (t *fakeTable) Claim(_ context.Context, schedule store.Reminder, nextDueAt time.Time) (bool, error) { t.recorder.mu.Lock() t.recorder.steps = append(t.recorder.steps, "claim") t.nextDueAt = append(t.nextDueAt, nextDueAt) @@ -39,7 +39,11 @@ type recordingInvoker struct { calls []store.GrainId } -func (i *recordingInvoker) Invoke(_ context.Context, id store.GrainId, method string) error { +func testReminderCall(store.GrainId, string, time.Time, time.Duration, time.Time) (any, any) { + return &struct{}{}, &struct{}{} +} + +func (i *recordingInvoker) Invoke(_ context.Context, id store.GrainId, method string, _, _ any) error { i.recorder.mu.Lock() i.recorder.steps = append(i.recorder.steps, "invoke") i.calls = append(i.calls, id) @@ -67,7 +71,7 @@ func (r *stepRecorder) record(step string) { r.mu.Unlock() } -func (i *blockingInvoker) Invoke(ctx context.Context, _ store.GrainId, _ string) error { +func (i *blockingInvoker) Invoke(ctx context.Context, _ store.GrainId, _ string, _, _ any) error { close(i.started) <-ctx.Done() close(i.finished) @@ -87,7 +91,7 @@ func (i *ownershipInvoker) Owns(store.GrainId) bool { return i.owns } -func (i *ownershipInvoker) Invoke(context.Context, store.GrainId, string) error { +func (i *ownershipInvoker) Invoke(context.Context, store.GrainId, string, any, any) error { i.calls.Add(1) return nil } @@ -97,7 +101,7 @@ func TestPoller_ClaimsBeforeInvoking(t *testing.T) { start := time.Unix(100, 0).UTC() recorder := &stepRecorder{} backend := &fakeTable{ - rows: []store.Schedule{{ + rows: []store.Reminder{{ GrainId: store.GrainId{GrainType: "account", GrainKey: "alice"}, Name: "wake", Method: "Wake", @@ -110,7 +114,7 @@ func TestPoller_ClaimsBeforeInvoking(t *testing.T) { } fakeClock := clock.NewFake(start) invoker := &recordingInvoker{recorder: recorder} - poller := New(backend, fakeClock, time.Second, invoker) + poller := New(backend, fakeClock, time.Second, invoker, testReminderCall) synctest.Wait() fakeClock.Advance(time.Second) synctest.Wait() @@ -127,7 +131,7 @@ func TestPoller_AdvancesToFirstFutureTime(t *testing.T) { start := time.Unix(200, 0).UTC() interval := time.Hour backend := &fakeTable{ - rows: []store.Schedule{{ + rows: []store.Reminder{{ GrainId: store.GrainId{GrainType: "account", GrainKey: "alice"}, Name: "wake", Method: "Wake", @@ -140,7 +144,7 @@ func TestPoller_AdvancesToFirstFutureTime(t *testing.T) { } fakeClock := clock.NewFake(start) invoker := &recordingInvoker{recorder: backend.recorder} - poller := New(backend, fakeClock, time.Second, invoker) + poller := New(backend, fakeClock, time.Second, invoker, testReminderCall) synctest.Wait() fakeClock.Advance(time.Second) synctest.Wait() @@ -156,11 +160,84 @@ func TestPoller_AdvancesToFirstFutureTime(t *testing.T) { }) } +func TestPoller_PassesPeriodicTickStatus(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Unix(250, 0).UTC() + period := time.Hour + first := start.Add(-3 * period) + due := start.Add(-period) + backend := &fakeTable{ + rows: []store.Reminder{{ + GrainId: store.GrainId{GrainType: "account", GrainKey: "alice"}, + Name: "wake", + Method: "Wake", + FirstTickTime: first, + DueAt: due, + Interval: period, + ETag: 1, + }}, + claimWon: true, + recorder: &stepRecorder{}, + } + fakeClock := clock.NewFake(start) + var gotFirst, gotCurrent time.Time + var gotPeriod time.Duration + factory := func(_ store.GrainId, _ string, firstTick time.Time, tickPeriod time.Duration, current time.Time) (any, any) { + gotFirst = firstTick + gotPeriod = tickPeriod + gotCurrent = current + return &struct{}{}, &struct{}{} + } + poller := New(backend, fakeClock, time.Second, &recordingInvoker{recorder: backend.recorder}, factory) + synctest.Wait() + fakeClock.Advance(time.Second) + synctest.Wait() + poller.Close() + + if !gotFirst.Equal(first) || gotPeriod != period || !gotCurrent.Equal(due) { + t.Fatalf("TickStatus = first %s period %s current %s, want %s %s %s", gotFirst, gotPeriod, gotCurrent, first, period, due) + } + }) +} + +func TestPoller_PassesZeroPeriodForOneShot(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + start := time.Unix(275, 0).UTC() + backend := &fakeTable{ + rows: []store.Reminder{{ + GrainId: store.GrainId{GrainType: "account", GrainKey: "alice"}, + Name: "wake", + Method: "Wake", + FirstTickTime: start, + DueAt: start, + ETag: 1, + }}, + claimWon: true, + recorder: &stepRecorder{}, + } + fakeClock := clock.NewFake(start) + var gotPeriod time.Duration + factory := func(_ store.GrainId, _ string, _ time.Time, period time.Duration, _ time.Time) (any, any) { + gotPeriod = period + return &struct{}{}, &struct{}{} + } + poller := New(backend, fakeClock, time.Second, &recordingInvoker{recorder: backend.recorder}, factory) + synctest.Wait() + fakeClock.Advance(time.Second) + synctest.Wait() + poller.Close() + + if gotPeriod != 0 { + t.Fatalf("one-shot Period = %s, want 0", gotPeriod) + } + }) +} + func TestPoller_ClaimFailureDoesNotInvoke(t *testing.T) { synctest.Test(t, func(t *testing.T) { start := time.Unix(300, 0).UTC() backend := &fakeTable{ - rows: []store.Schedule{{ + rows: []store.Reminder{{ GrainId: store.GrainId{GrainType: "account", GrainKey: "alice"}, Name: "wake", Method: "Wake", @@ -170,7 +247,7 @@ func TestPoller_ClaimFailureDoesNotInvoke(t *testing.T) { } fakeClock := clock.NewFake(start) invoker := &recordingInvoker{recorder: backend.recorder} - poller := New(backend, fakeClock, time.Second, invoker) + poller := New(backend, fakeClock, time.Second, invoker, testReminderCall) synctest.Wait() fakeClock.Advance(time.Second) synctest.Wait() @@ -186,7 +263,7 @@ func TestPoller_SkipsSchedulesNotOwnedByInvoker(t *testing.T) { synctest.Test(t, func(t *testing.T) { start := time.Unix(350, 0).UTC() backend := store.NewMemory() - schedule := store.Schedule{ + schedule := store.Reminder{ GrainId: store.GrainId{GrainType: "account", GrainKey: "alice"}, Name: "wake", Method: "Wake", @@ -200,8 +277,8 @@ func TestPoller_SkipsSchedulesNotOwnedByInvoker(t *testing.T) { owner := &ownershipInvoker{owns: true} nonOwnerClock := clock.NewFake(start) ownerClock := clock.NewFake(start) - nonOwnerPoller := New(backend, nonOwnerClock, time.Second, nonOwner) - ownerPoller := New(backend, ownerClock, time.Second, owner) + nonOwnerPoller := New(backend, nonOwnerClock, time.Second, nonOwner, testReminderCall) + ownerPoller := New(backend, ownerClock, time.Second, owner, testReminderCall) synctest.Wait() nonOwnerClock.Advance(time.Second) @@ -227,7 +304,7 @@ func TestPoller_CloseStopsTheGoroutine(t *testing.T) { synctest.Test(t, func(t *testing.T) { start := time.Unix(400, 0).UTC() backend := &fakeTable{ - rows: []store.Schedule{{ + rows: []store.Reminder{{ GrainId: store.GrainId{GrainType: "account", GrainKey: "alice"}, Name: "wake", Method: "Wake", @@ -238,7 +315,7 @@ func TestPoller_CloseStopsTheGoroutine(t *testing.T) { } fakeClock := clock.NewFake(start) invoker := &blockingInvoker{started: make(chan struct{}), finished: make(chan struct{})} - poller := New(backend, fakeClock, time.Second, invoker) + poller := New(backend, fakeClock, time.Second, invoker, testReminderCall) synctest.Wait() fakeClock.Advance(time.Second) synctest.Wait() From 382fb49fb955c7332604a29e86d8948572ce0ac4 Mon Sep 17 00:00:00 2001 From: Tequila Sunset Date: Sat, 8 Aug 2026 19:39:00 +0800 Subject: [PATCH 3/5] Bound reminder due time calculation --- timer/timer.go | 35 ++++++++++++++++++++++++++++++----- timer/timer_test.go | 28 ++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 5 deletions(-) diff --git a/timer/timer.go b/timer/timer.go index cd7d414..926e321 100644 --- a/timer/timer.go +++ b/timer/timer.go @@ -104,13 +104,38 @@ func (p *Poller) poll() { } } +const maxDuration = time.Duration(1<<63 - 1) + func nextDueAt(reminder store.Reminder, now time.Time) time.Time { - if reminder.Interval <= 0 { + period := reminder.Interval + if period <= 0 { return time.Time{} } - next := reminder.DueAt.Add(reminder.Interval) - for !next.After(now) { - next = next.Add(reminder.Interval) + if reminder.DueAt.After(now) { + return reminder.DueAt + } + + elapsed := now.Sub(reminder.DueAt) + missed := elapsed / period + if missed == maxDuration { + return futureDueAt(now, period) + } + missed++ + if missed > maxDuration/period { + return futureDueAt(now, period) + } + + candidate := reminder.DueAt.Add(period * missed) + if !candidate.After(now) { + return futureDueAt(now, period) + } + return candidate +} + +func futureDueAt(now time.Time, period time.Duration) time.Time { + fallback := now.Add(period) + if fallback.After(now) { + return fallback } - return next + return now.Add(time.Nanosecond) } diff --git a/timer/timer_test.go b/timer/timer_test.go index 1f8211d..428aea2 100644 --- a/timer/timer_test.go +++ b/timer/timer_test.go @@ -160,6 +160,34 @@ func TestPoller_AdvancesToFirstFutureTime(t *testing.T) { }) } +func TestNextDueAt_LargeDowntimeReturnsPromptly(t *testing.T) { + period := time.Nanosecond + dueAt := time.Unix(0, 0).UTC() + now := dueAt.Add(time.Hour) + reminder := store.Reminder{DueAt: dueAt, Interval: period} + + result := make(chan time.Time, 1) + go func() { + result <- nextDueAt(reminder, now) + }() + + select { + case got := <-result: + want := now.Add(period) + if !got.After(now) || !got.Equal(want) { + t.Fatalf("next due time = %s, want %s strictly after now", got, want) + } + case <-time.After(time.Second): + t.Fatal("nextDueAt did not return promptly") + } + + future := now.Add(time.Hour) + reminder.DueAt = future + if got := nextDueAt(reminder, now); !got.Equal(future) { + t.Fatalf("future due time = %s, want unchanged %s", got, future) + } +} + func TestPoller_PassesPeriodicTickStatus(t *testing.T) { synctest.Test(t, func(t *testing.T) { start := time.Unix(250, 0).UTC() From bc8e7d0b5a792cb2d35ad45b03e5cd2973234d72 Mon Sep 17 00:00:00 2001 From: Tequila Sunset Date: Sat, 8 Aug 2026 19:47:26 +0800 Subject: [PATCH 4/5] Make timer arithmetic tests deterministic --- timer/timer_test.go | 39 ++++++++++++++++++++++++++------------- 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/timer/timer_test.go b/timer/timer_test.go index 428aea2..c8a5ae6 100644 --- a/timer/timer_test.go +++ b/timer/timer_test.go @@ -166,19 +166,10 @@ func TestNextDueAt_LargeDowntimeReturnsPromptly(t *testing.T) { now := dueAt.Add(time.Hour) reminder := store.Reminder{DueAt: dueAt, Interval: period} - result := make(chan time.Time, 1) - go func() { - result <- nextDueAt(reminder, now) - }() - - select { - case got := <-result: - want := now.Add(period) - if !got.After(now) || !got.Equal(want) { - t.Fatalf("next due time = %s, want %s strictly after now", got, want) - } - case <-time.After(time.Second): - t.Fatal("nextDueAt did not return promptly") + got := nextDueAt(reminder, now) + want := now.Add(period) + if !got.After(now) || !got.Equal(want) { + t.Fatalf("next due time = %s, want %s strictly after now", got, want) } future := now.Add(time.Hour) @@ -186,6 +177,28 @@ func TestNextDueAt_LargeDowntimeReturnsPromptly(t *testing.T) { if got := nextDueAt(reminder, now); !got.Equal(future) { t.Fatalf("future due time = %s, want unchanged %s", got, future) } + + reminder.DueAt = now + reminder.Interval = 2 * time.Nanosecond + if got := nextDueAt(reminder, now); !got.Equal(now.Add(reminder.Interval)) { + t.Fatalf("due-now next time = %s, want %s", got, now.Add(reminder.Interval)) + } + + for _, interval := range []time.Duration{0, -time.Nanosecond} { + reminder.Interval = interval + if got := nextDueAt(reminder, now); !got.IsZero() { + t.Fatalf("interval %s next time = %s, want zero", interval, got) + } + } + + const maxElapsed = time.Duration(1<<63 - 1) + overflowDueAt := time.Unix(0, 0).UTC() + overflowNow := overflowDueAt.Add(maxElapsed) + reminder.DueAt = overflowDueAt + reminder.Interval = 2 * time.Nanosecond + if got := nextDueAt(reminder, overflowNow); !got.After(overflowNow) { + t.Fatalf("overflow fallback = %s, want strictly after %s", got, overflowNow) + } } func TestPoller_PassesPeriodicTickStatus(t *testing.T) { From 063ba379740e7e5fafdce78fb65126597def4fc0 Mon Sep 17 00:00:00 2001 From: Tequila Sunset Date: Sat, 8 Aug 2026 19:53:22 +0800 Subject: [PATCH 5/5] Update timer design implementation gaps --- design/timers.md | 25 +++++++++++++++++-------- 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/design/timers.md b/design/timers.md index 820e678..647d895 100644 --- a/design/timers.md +++ b/design/timers.md @@ -253,7 +253,13 @@ change to receive one `BackgroundError`. Reminder deliveries read The error sink is implemented in the current code. `OnError` receives a `BackgroundError` with the Grain, original error, and source. The source set -is sealed. The public naming migration remains part of the 0.1.0 API work. +is sealed. Reminder delivery failures and deactivation hook failures use the +sources described above; scan failures, claim failures, and shutdown +cancellations remain outside the sink by design. + +Cluster ownership checks and forwarding are implemented for the current +optional cluster preview. Rolling upgrades and other operational cluster work +remain deferred and are outside this design. ## Don't claim rows that are not yours @@ -316,10 +322,13 @@ The minimum failure, restart, and claim tests must cover these cases: ## Gap -The typed Reminder method handle is implemented in the current code. The -public naming migration to `Reminder`, `ReminderTime`, `NewReminder`, and -`ReminderStore` remains part of the 0.1.0 API work. The `first_tick_time` row -field and the generated typed `newReminderCall` factory are also part of that -work. This design batch does not rename the Go implementation. The method -name is read from the expression once. The table, poller, and restart recovery -use the method-name string as an internal identifier. +The typed Reminder method handle, the public Reminder names, the +`first_tick_time` row field, and the generated typed Reminder-call factory are +implemented. The structured error sink is also implemented. The method name +is read from the expression once. The table, poller, and restart recovery use +the method-name string as an internal identifier. + +The remaining work is outside this single-node Reminder contract. The +optional cluster implementation is shipped as a preview; rolling upgrades +and operational cleanup remain deferred. The announced 0.1.0 release still +requires its conformance and failure-evidence work.