refactor(sync/cachable): finalize release API

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

sync/cachable/go.mod

@@ -0,0 +1,3 @@
+module github.com/therootcompany/golib/sync/cachable
+
+go 1.26.0

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

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