diff --git a/sync/cachable/cacheable.go b/sync/cachable/cacheable.go new file mode 100644 index 0000000..ffc648e --- /dev/null +++ b/sync/cachable/cacheable.go @@ -0,0 +1,46 @@ +// Package cachable defines contracts for lazily refreshed cached values. +package cachable + +import ( + "context" + "time" +) + +// Status describes the currently published snapshot. StaleAt is the next +// time the implementation wants a freshness check; it is not necessarily +// the time at which the value becomes unusable. +type Status struct { + LoadedAt time.Time + StaleAt time.Time + Refreshing bool + HasValue bool + LastError error +} + +// Cacheable is a refreshable, atomically published snapshot. Load always +// waits for the first snapshot. For later loads, wait controls whether the +// caller waits for a due or in-flight refresh. +type Cacheable[T any] interface { + Load(ctx context.Context, wait bool) (*T, error) + Current() *T + Due(ctx context.Context) (due bool, err error) + Revalidate(ctx context.Context) (started bool, err error) + Clear() error +} + +// Mutable is the optional manual publication capability. +type Mutable[T any] interface { + Set(*T) error + Clear() error +} + +// Inspectable is the optional status capability. +type Inspectable interface { + Status() Status +} + +// Tickable is the optional background revalidation lifecycle. +type Tickable interface { + Start(ctx context.Context, interval time.Duration) + Stop() error +} diff --git a/sync/cachable/go.mod b/sync/cachable/go.mod new file mode 100644 index 0000000..e374a21 --- /dev/null +++ b/sync/cachable/go.mod @@ -0,0 +1,3 @@ +module github.com/therootcompany/golib/sync/cachable + +go 1.26.0 diff --git a/sync/cachable/refresh.go b/sync/cachable/refresh.go new file mode 100644 index 0000000..b7847c3 --- /dev/null +++ b/sync/cachable/refresh.go @@ -0,0 +1,101 @@ +package cachable + +import ( + "context" + "sync" + "sync/atomic" +) + +// Refresh coordinates one in-flight refresh. It is for cache implementations; +// callers normally use Cacheable rather than this type directly. +// +// Running is lock-free (atomic.Bool) so hot-path callers that only need to +// check whether a refresh is in flight never contend on the mutex. Begin, +// Finish, and Wait use a mutex to publish and coordinate each run's result. +type Refresh struct { + mu sync.Mutex + run atomic.Bool + stopped bool + state *refreshState +} + +type refreshState struct { + done chan struct{} + err error +} + +// Begin claims the refresh slot. It returns false when work is already running. +func (r *Refresh) Begin() bool { + r.mu.Lock() + defer r.mu.Unlock() + if r.stopped || r.run.Load() { + return false + } + r.state = &refreshState{done: make(chan struct{})} + r.run.Store(true) + return true +} + +func (r *Refresh) Finish(err error) { + r.mu.Lock() + state := r.state + state.err = err + close(state.done) + r.run.Store(false) + r.mu.Unlock() +} + +// Running reports whether a refresh is in flight. It is lock-free. +func (r *Refresh) Running() bool { + return r.run.Load() +} + +// Stop prevents future refreshes and waits for the current refresh, if any. +// The caller supplies the shutdown deadline. If the refresh finishes before +// that deadline, Stop returns its update/persistence error; otherwise it +// returns ctx.Err(). +func (r *Refresh) Stop(ctx context.Context) error { + r.mu.Lock() + r.stopped = true + state := r.state + running := r.run.Load() + r.mu.Unlock() + if !running || state == nil { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-state.done: + return state.err + } +} + +// Wait blocks until the in-flight refresh completes and returns its error. +// If no refresh is running, it returns the error of the most recently completed +// refresh (nil if none has run). The stored error is cleared by the next +// Begin, so repeat callers see only the latest outcome. For first-load this +// always surfaces the refresh error: Load calls Revalidate (which Begins) +// before Wait, so the refresh this caller started is still running when Wait +// observes it via the done channel. Callers wanting the authoritative last +// error rather than a per-call wait result should read Status.LastError. +func (r *Refresh) Wait(ctx context.Context) error { + r.mu.Lock() + state := r.state + if !r.run.Load() { + if state == nil { + r.mu.Unlock() + return nil + } + err := state.err + r.mu.Unlock() + return err + } + r.mu.Unlock() + select { + case <-ctx.Done(): + return ctx.Err() + case <-state.done: + return state.err + } +} diff --git a/sync/cachable/refresh_test.go b/sync/cachable/refresh_test.go new file mode 100644 index 0000000..ffdbaa6 --- /dev/null +++ b/sync/cachable/refresh_test.go @@ -0,0 +1,237 @@ +package cachable + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" +) + +// TestRefresh_BeginCoalesces confirms only one caller wins Begin while work is +// running; all others get false. This is the stampede-protection contract. +func TestRefresh_BeginCoalesces(t *testing.T) { + var r Refresh + if !r.Begin() { + t.Fatal("first Begin must succeed") + } + for range 10 { + if r.Begin() { + t.Fatal("concurrent Begin must not succeed while running") + } + } + if !r.Running() { + t.Fatal("Running must be true after Begin") + } + r.Finish(nil) + if r.Running() { + t.Fatal("Running must be false after Finish") + } +} + +// TestRefresh_BeginAgainAfterFinish confirms the slot is reusable. +func TestRefresh_BeginAgainAfterFinish(t *testing.T) { + var r Refresh + r.Begin() + r.Finish(nil) + if !r.Begin() { + t.Fatal("Begin must succeed after a finished refresh") + } + r.Finish(nil) +} + +// TestRefresh_WaitBlocksUntilFinish confirms Wait blocks and then returns the +// refresh error. +func TestRefresh_WaitBlocksUntilFinish(t *testing.T) { + var r Refresh + r.Begin() + + want := errors.New("boom") + go func() { + time.Sleep(20 * time.Millisecond) + r.Finish(want) + }() + + if err := r.Wait(context.Background()); !errors.Is(err, want) { + t.Errorf("Wait err = %v, want %v", err, want) + } +} + +// TestRefresh_WaitManyCallers confirms many concurrent waiters all observe the +// single Finish. +func TestRefresh_WaitManyCallers(t *testing.T) { + var r Refresh + r.Begin() + + const n = 20 + var wg sync.WaitGroup + errs := make([]error, n) + start := make(chan struct{}) + wg.Add(n) + for i := range n { + go func(i int) { + defer wg.Done() + <-start + errs[i] = r.Wait(context.Background()) + }(i) + } + close(start) + + time.Sleep(20 * time.Millisecond) + want := errors.New("fail") + r.Finish(want) + wg.Wait() + + for i, err := range errs { + if !errors.Is(err, want) { + t.Errorf("waiter %d err = %v, want %v", i, err, want) + } + } +} + +// TestRefresh_WaitNoRunningReturnsStoredErr confirms that when no refresh is +// running, Wait returns the most recently completed refresh's error (the +// last-good/last-bad surface for callers that arrive after completion). +func TestRefresh_WaitNoRunningReturnsStoredErr(t *testing.T) { + var r Refresh + r.Begin() + want := errors.New("persisted") + r.Finish(want) + + if err := r.Wait(context.Background()); !errors.Is(err, want) { + t.Errorf("Wait after finish err = %v, want %v", err, want) + } +} + +// TestRefresh_WaitKeepsItsRefreshResult confirms waiters retain the result +// belonging to the refresh they observed, even when the next refresh starts +// immediately after Finish. +func TestRefresh_WaitKeepsItsRefreshResult(t *testing.T) { + var r Refresh + if !r.Begin() { + t.Fatal("initial Begin must succeed") + } + + const n = 64 + results := make(chan error, n) + for range n { + go func() { + results <- r.Wait(context.Background()) + }() + } + time.Sleep(10 * time.Millisecond) + + first := errors.New("first refresh failed") + r.Finish(first) + if !r.Begin() { + t.Fatal("second Begin must succeed") + } + second := errors.New("second refresh failed") + r.Finish(second) + + for range n { + if err := <-results; !errors.Is(err, first) { + t.Errorf("old waiter got %v, want %v", err, first) + } + } +} + +// TestRefresh_BeginClearsErr confirms a new refresh clears the previous error +// so a stale failure is not surfaced to waiters of the new refresh. +func TestRefresh_BeginClearsErr(t *testing.T) { + var r Refresh + r.Begin() + r.Finish(errors.New("old")) + + r.Begin() + r.Finish(nil) + + if err := r.Wait(context.Background()); err != nil { + t.Errorf("Wait after cleared refresh err = %v, want nil", err) + } +} + +// TestRefresh_WaitContextCancel confirms a caller can bail out of Wait. +func TestRefresh_WaitContextCancel(t *testing.T) { + var r Refresh + r.Begin() + defer r.Finish(nil) // don't leak the goroutine slot + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + if err := r.Wait(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("Wait err = %v, want DeadlineExceeded", err) + } +} + +// TestRefresh_StopWaitsAndPreventsNewRuns confirms Stop is a lifecycle barrier. +func TestRefresh_StopWaitsAndPreventsNewRuns(t *testing.T) { + var r Refresh + if !r.Begin() { + t.Fatal("Begin must succeed") + } + want := errors.New("persist failure") + done := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + done <- r.Stop(ctx) + }() + time.Sleep(10 * time.Millisecond) + if r.Begin() { + t.Fatal("Begin must fail after Stop claims the coordinator") + } + select { + case err := <-done: + t.Fatalf("Stop returned before refresh finished: %v", err) + default: + } + r.Finish(want) + if err := <-done; !errors.Is(err, want) { + t.Errorf("Stop err = %v, want %v", err, want) + } + if r.Begin() { + t.Fatal("Begin must remain disabled after Stop") + } +} + +// TestRefresh_StopHonorsTimeout confirms Stop returns its timeout error when +// the in-flight refresh does not finish before the shutdown deadline. +func TestRefresh_StopHonorsTimeout(t *testing.T) { + var r Refresh + r.Begin() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + if err := r.Stop(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("Stop err = %v, want DeadlineExceeded", err) + } + r.Finish(nil) +} + +// TestRefresh_RunningIsLockFree is a smoke test that Running does not need the +// mutex (no deadlock under contention). It races Running against Begin/Finish. +func TestRefresh_RunningUnderContention(t *testing.T) { + var r Refresh + var stop atomic.Bool + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + for !stop.Load() { + if r.Begin() { + time.Sleep(time.Microsecond) + r.Finish(nil) + } + } + }() + go func() { + defer wg.Done() + for !stop.Load() { + _ = r.Running() + } + }() + time.Sleep(50 * time.Millisecond) + stop.Store(true) + wg.Wait() +}