Implement Reminders and TickStatus

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,24 +66,24 @@ 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
 }
 
 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
 			}),

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 {

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 {

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
+}

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

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 {

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 {

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.