refactor(sync/dataset): use released cachable API

sync/dataset/dataset.go

@@ -1,56 +1,57 @@
 // Package dataset manages values that are periodically re-fetched from an
 // upstream source and hot-swapped behind atomic pointers. Consumers read via
 // View.Value (lock-free); a single Load drives any number of views off a
-// shared set of Fetchers, so upstreams (one git pull, one tar.gz download)
+// shared set of Upstreams, so upstreams (one git pull, one tar.gz download)
 // don't get re-fetched per view.
 //
 // Typical lifecycle:
 //
-//	s := dataset.NewSet(repo) // *gitshallow.Repo satisfies Fetcher
+//	s := dataset.NewSet(repo) // *gitshallow.Repo satisfies Upstream
 //	inbound  := dataset.Add(s, func(ctx context.Context) (*ipcohort.Cohort, error) { ... })
 //	outbound := dataset.Add(s, func(ctx context.Context) (*ipcohort.Cohort, error) { ... })
-//	if err := s.Load(ctx); err != nil { ... }       // initial populate
+//	if err := s.Load(ctx, true); err != nil { ... }       // initial populate
-//	go s.Tick(ctx, 47*time.Minute, onError)         // background refresh
+//	inbound.Start(ctx, 47*time.Minute)              // optional refresh
-//	current := inbound.Value()                      // lock-free read
+//	current := inbound.Current()                    // lock-free read
 package dataset
 
 import (
 	"context"
-	"errors"
 	"io"
 	"os"
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/therootcompany/golib/sync/cachable"
 )
 
-// Fetcher reports whether an upstream source has changed since the last call.
+// Upstream reports whether an upstream source has changed since the last call.
 // Implementations should dedup rapid-fire calls internally (e.g. gitshallow
 // skips redundant pulls within a short window; httpcache uses ETag) and must
-// honor ctx so Set.Tick can cancel mid-fetch on shutdown.
+// honor ctx so asynchronous updates can cancel on shutdown.
-type Fetcher interface {
+type Upstream interface {
-	Fetch(ctx context.Context) (updated bool, err error)
+	Update(ctx context.Context) (updated bool, err error)
 }
 
-// FetcherFunc adapts a plain function to Fetcher.
+// UpstreamFunc adapts a plain function to Upstream.
-type FetcherFunc func(ctx context.Context) (bool, error)
+type UpstreamFunc func(ctx context.Context) (bool, error)
 
-func (f FetcherFunc) Fetch(ctx context.Context) (bool, error) { return f(ctx) }
+func (f UpstreamFunc) Update(ctx context.Context) (bool, error) { return f(ctx) }
 
-// NopFetcher always reports no update. Use for sets whose source never
+// NopUpstream always reports no update. Use for sets whose source never
 // changes (test fixtures, embedded data).
-type NopFetcher struct{}
+type NopUpstream struct{}
 
-func (NopFetcher) Fetch(ctx context.Context) (bool, error) { return false, nil }
+func (NopUpstream) Update(ctx context.Context) (bool, error) { return false, nil }
 
-// PollFiles returns a Fetcher that stat's the given paths and reports
+// PollFiles returns a Upstream that stat's the given paths and reports
 // "updated" whenever any file's size or modtime has changed since the last
 // call. The first call always reports updated=true.
 //
 // Use for Sets whose source is local files that may be edited out of band
-// (e.g. a user-provided --inbound list) — pair with Set.Tick to pick up
+// (e.g. a user-provided --inbound list) — pair with Set.Start to pick up
 // changes automatically.
-func PollFiles(paths ...string) Fetcher {
+func PollFiles(paths ...string) Upstream {
 	return &filePoller{paths: paths, stats: make(map[string]fileStat, len(paths))}
 }
 
@@ -65,7 +66,7 @@ type filePoller struct {
 	stats map[string]fileStat
 }
 
-func (p *filePoller) Fetch(ctx context.Context) (bool, error) {
+func (p *filePoller) Update(ctx context.Context) (bool, error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	changed := false
@@ -86,33 +87,43 @@ func (p *filePoller) Fetch(ctx context.Context) (bool, error) {
 	return changed, nil
 }
 
-// Set ties one or more Fetchers to one or more views. A Load call fetches
+// Set ties one or more Upstreams to one or more views. A Load call fetches
 // each source and, on the first call or when any source reports a change,
 // reloads every view and atomically swaps its current value. Use multiple
 // fetchers when a single logical dataset is spread across several archives
 // (e.g. GeoLite2 City + ASN); a single fetcher is the common case (one git
 // repo, one tar.gz).
-// Set serializes Load and Close with a mutex so concurrent callers (e.g. an
+// Set serializes snapshot publication so concurrent callers do not race views.
-// /admin/reload handler firing while Tick also fires) don't race the views.
 // Add and AddInitial mutate s.views without locking — they MUST be called
 // before the first Load.
 type Set struct {
-	mu       sync.Mutex
+	publishMu      sync.Mutex
-	closed   atomic.Bool // set by Close; Load returns early when true
+	fetchers       []Upstream
-	fetchers []Fetcher
 	views          []reloader
 	loaded         atomic.Bool
+	loadedAt       atomic.Pointer[time.Time]
+	start          sync.Once
+	cancel         context.CancelFunc
+	refresh        cachable.Refresh
+	generation     atomic.Uint64
+	RefreshTimeout time.Duration
+	// RefreshInterval gates how often Revalidate will start a refresh. Zero
+	// means always due (freshness is delegated to the upstreams' own dedup
+	// windows). Set to a positive duration to match the other cachable
+	// implementations' time-based Due.
+	RefreshInterval time.Duration
 }
 
 // reloader is a type-erased handle to a View's reload function.
 type reloader interface {
-	reload(ctx context.Context) error
+	prepare(ctx context.Context) (any, error)
+	publish(value any)
 	clear()
 }
 
 // NewSet creates a Set backed by fetchers. All fetchers are called on every
 // Load; the set reloads its views whenever any one of them reports a change.
-func NewSet(fetchers ...Fetcher) *Set {
+func NewSet(fetchers ...Upstream) *Set {
 	return &Set{fetchers: fetchers}
 }
 
@@ -121,21 +132,61 @@ func (s *Set) Loaded() bool {
 	return s.loaded.Load()
 }
 
-// Load fetches upstream and, on the first call or whenever any fetcher
+// Due reports whether the set should be revalidated. With RefreshInterval > 0
-// reports a change, reloads every view and atomically installs the new values.
+// it returns false until StaleAt (loadedAt + RefreshInterval) has passed, so a
-// Returns context.Canceled if the set has been closed.
+// hot path calling Load(false) does not start a refresh on every request.
-func (s *Set) Load(ctx context.Context) error {
+// With RefreshInterval == 0 it always returns true: the upstreams own their
-	if s.closed.Load() {
+// own dedup windows (httpcache ETag, gitshallow pull throttling) and a refresh
-		return context.Canceled
+// is a cheap "nothing changed" check.
+func (s *Set) Due(ctx context.Context) (bool, error) {
+	if err := ctx.Err(); err != nil {
+		return false, err
 	}
-	s.mu.Lock()
+	if s.RefreshInterval <= 0 {
-	defer s.mu.Unlock()
+		return true, nil
+	}
+	if !s.loaded.Load() {
+		return true, nil
+	}
+	loadedAt := s.loadedAt.Load()
+	if loadedAt == nil {
+		return true, nil
+	}
+	return !time.Now().Before(loadedAt.Add(s.RefreshInterval)), nil
+}
+
+// Revalidate checks whether work is due and starts one asynchronous update.
+// Concurrent calls coalesce onto the same update. The update runs with a
+// detached context bounded by RefreshTimeout (default 5 minutes) so a
+// cancelled caller context does not kill background work.
+func (s *Set) Revalidate(ctx context.Context) (started bool, err error) {
+	due, err := s.Due(ctx)
+	if err != nil || !due {
+		return false, err
+	}
+	if !s.refresh.Begin() {
+		return false, nil
+	}
+	go func() {
+		timeout := s.RefreshTimeout
+		if timeout <= 0 {
+			timeout = 5 * time.Minute
+		}
+		refreshCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout)
+		defer cancel()
+		s.refresh.Finish(s.update(refreshCtx))
+	}()
+	return true, nil
+}
+
+func (s *Set) update(ctx context.Context) error {
+	generation := s.generation.Load()
 	updated := false
 	for _, f := range s.fetchers {
 		if err := ctx.Err(); err != nil {
 			return err
 		}
-		u, err := f.Fetch(ctx)
+		u, err := f.Update(ctx)
 		if err != nil {
 			return err
 		}
@@ -146,102 +197,192 @@ func (s *Set) Load(ctx context.Context) error {
 	if s.loaded.Load() && !updated {
 		return nil
 	}
-	for _, v := range s.views {
+	prepared := make([]any, len(s.views))
+	for i, v := range s.views {
 		if err := ctx.Err(); err != nil {
 			return err
 		}
-		if err := v.reload(ctx); err != nil {
+		value, err := v.prepare(ctx)
+		if err != nil {
 			return err
 		}
+		prepared[i] = value
 	}
+	s.publishMu.Lock()
+	defer s.publishMu.Unlock()
+	if s.generation.Load() != generation {
+		return nil
+	}
+	for i, v := range s.views {
+		v.publish(prepared[i])
+	}
+	now := time.Now()
+	s.loadedAt.Store(&now)
 	s.loaded.Store(true)
 	return nil
 }
 
-// Close closes every view's currently-held value and clears all view
+// Load performs the initial load, or revalidates an already-loaded set. It
-// pointers. Call on shutdown to release any OS resources held by the
+// keeps the last good snapshot when a later revalidation fails.
-// final snapshots (file handles, network connections). Safe to call on a
+func (s *Set) Load(ctx context.Context, wait bool) error {
-// Set that hasn't been loaded or whose views hold pure in-memory values —
+	_, err := s.Revalidate(ctx)
-// non-Closer values are skipped. Idempotent.
+	if err != nil {
-//
+		return err
-// It is safe to call Close while Tick is still running; subsequent Load
+	}
-// calls will return context.Canceled.
+	if !s.loaded.Load() || wait {
-func (s *Set) Close() error {
+		return s.refresh.Wait(ctx)
-	if s.closed.Swap(true) {
+	}
 	return nil
 }
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	var errs []error
-	for _, v := range s.views {
-		if c, ok := v.(io.Closer); ok {
-			if err := c.Close(); err != nil {
-				errs = append(errs, err)
-			}
-		}
-		v.clear()
-	}
-	return errors.Join(errs...)
-}
 
-// Tick calls Load every interval until ctx is done. Load errors are passed to
+// Start starts at most one optional background revalidation ticker.
-// onError (if non-nil) and do not stop the loop; callers choose whether to log,
+func (s *Set) Start(ctx context.Context, interval time.Duration) {
-// count, page, or ignore. Run in a goroutine: `go s.Tick(ctx, d, onError)`.
+	s.start.Do(func() {
-func (s *Set) Tick(ctx context.Context, interval time.Duration, onError func(error)) {
+		if interval <= 0 {
-	t := time.NewTicker(interval)
+			interval = time.Hour
-	defer t.Stop()
+		}
+		tickCtx, cancel := context.WithCancel(ctx)
+		s.cancel = cancel
+		go func() {
+			ticker := time.NewTicker(interval)
+			defer ticker.Stop()
 			for {
 				select {
-		case <-ctx.Done():
+				case <-tickCtx.Done():
 					return
-		case <-t.C:
+				case <-ticker.C:
-			if err := s.Load(ctx); err != nil && onError != nil {
+					_, _ = s.Revalidate(tickCtx)
-				onError(err)
 				}
 			}
+		}()
+	})
 }
+
+// Stop stops the optional background ticker and waits for any in-flight
+// refresh to finish or reach RefreshTimeout. It does not clear snapshots.
+func (s *Set) Stop() error {
+	if s.cancel != nil {
+		s.cancel()
+	}
+	timeout := s.RefreshTimeout
+	if timeout <= 0 {
+		timeout = 5 * time.Minute
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	return s.refresh.Stop(ctx)
+}
+
+// Clear removes every published view and closes values that own resources.
+// The set remains usable and may be loaded again.
+func (s *Set) Clear() error {
+	s.generation.Add(1)
+	s.loaded.Store(false)
+	s.loadedAt.Store(nil)
+	s.publishMu.Lock()
+	defer s.publishMu.Unlock()
+	for _, v := range s.views {
+		v.clear()
+	}
+	return nil
 }
 
 // View is a read-only handle to one dataset inside a Set.
 type View[T any] struct {
+	set      *Set
 	loader   func(ctx context.Context) (*T, error)
 	ptr      atomic.Pointer[T]
 	loadedAt atomic.Pointer[time.Time] // nil until first successful reload
 }
 
-// Value returns the current snapshot. Nil before the Set is first loaded
+var (
-// unless the view was registered via AddInitial.
+	_ cachable.Cacheable[any] = (*View[any])(nil)
-func (v *View[T]) Value() *T {
+	_ cachable.Mutable[any]   = (*View[any])(nil)
+	_ cachable.Inspectable    = (*View[any])(nil)
+	_ cachable.Tickable       = (*View[any])(nil)
+)
+
+// Current returns the current snapshot without blocking. It is nil before
+// the Set is first loaded unless the view was registered via AddInitial.
+func (v *View[T]) Current() *T {
 	return v.ptr.Load()
 }
 
-// LoadedAt returns the time of the most recent successful reload, or the
+// Due reports whether the owning set should be revalidated. It delegates to
-// zero time if the view has never loaded.
+// Set.Due, which honors RefreshInterval when set and otherwise treats the set
-func (v *View[T]) LoadedAt() time.Time {
+// as always due (upstreams own their own dedup windows).
-	if t := v.loadedAt.Load(); t != nil {
+func (v *View[T]) Due(ctx context.Context) (bool, error) {
-		return *t
+	return v.set.Due(ctx)
-	}
-	return time.Time{}
 }
 
-// Close clears the view and closes the currently-held value if it
+// Revalidate asks the owning set to check its upstreams and returns whether
-// implements io.Closer. Idempotent — subsequent calls are no-ops.
+// this caller started the refresh.
-func (v *View[T]) Close() error {
+func (v *View[T]) Revalidate(ctx context.Context) (bool, error) {
-	prev := v.ptr.Swap(nil)
+	return v.set.Revalidate(ctx)
-	if prev == nil {
-		return nil
 }
-	if closer, ok := any(prev).(io.Closer); ok {
+
-		return closer.Close()
+// Load revalidates the owning set and returns this view's newest snapshot.
+// On refresh failure, the last good snapshot is returned with the error.
+func (v *View[T]) Load(ctx context.Context, wait bool) (*T, error) {
+	if err := v.set.Load(ctx, wait); err != nil {
+		return v.Current(), err
+	}
+	return v.Current(), nil
+}
+
+func (v *View[T]) Set(value *T) error {
+	v.set.publishMu.Lock()
+	defer v.set.publishMu.Unlock()
+	v.set.generation.Add(1)
+	previous := v.ptr.Swap(value)
+	if previous != nil {
+		if closer, ok := any(previous).(io.Closer); ok {
+			_ = closer.Close()
+		}
+	}
+	if value != nil {
+		now := time.Now()
+		v.loadedAt.Store(&now)
+		v.set.loadedAt.Store(&now)
 	}
 	return nil
 }
 
-func (v *View[T]) reload(ctx context.Context) error {
+func (v *View[T]) Clear() error {
-	t, err := v.loader(ctx)
+	return v.Set(nil)
-	if err != nil {
-		return err
 }
+
+// Status reports the view's published snapshot. StaleAt is populated from the
+// set's RefreshInterval (when set); with a zero interval StaleAt stays zero
+// because freshness is delegated to the upstreams.
+func (v *View[T]) Status() cachable.Status {
+	status := cachable.Status{HasValue: v.Current() != nil, Refreshing: v.set.refresh.Running()}
+	if loadedAt := v.loadedAt.Load(); loadedAt != nil {
+		status.LoadedAt = *loadedAt
+		if v.set.RefreshInterval > 0 {
+			status.StaleAt = status.LoadedAt.Add(v.set.RefreshInterval)
+		}
+	}
+	return status
+}
+
+// Start starts the owning set's optional background ticker.
+func (v *View[T]) Start(ctx context.Context, interval time.Duration) {
+	v.set.Start(ctx, interval)
+}
+
+// Stop stops the owning set's optional background ticker and waits for any
+// in-flight refresh to finish or reach RefreshTimeout.
+func (v *View[T]) Stop() error {
+	return v.set.Stop()
+}
+
+func (v *View[T]) prepare(ctx context.Context) (any, error) {
+	return v.loader(ctx)
+}
+
+func (v *View[T]) publish(value any) {
+	t := value.(*T)
 	prev := v.ptr.Swap(t)
 	// Close the replaced value if it holds OS resources (open file handles,
 	// network connections). Geoip readers and similar wrappers implement
@@ -257,13 +398,19 @@ func (v *View[T]) reload(ctx context.Context) error {
 	}
 	now := time.Now()
 	v.loadedAt.Store(&now)
-	return nil
 }
 
-func (v *View[T]) clear() { v.ptr.Swap(nil) }
+func (v *View[T]) clear() {
+	prev := v.ptr.Swap(nil)
+	if prev != nil {
+		if closer, ok := any(prev).(io.Closer); ok {
+			_ = closer.Close()
+		}
+	}
+}
 
 // Add registers a new view in s and returns it. Call after NewSet and before
-// the first Load. View.Value() returns nil until Set.Load succeeds.
+// the first Load. View.Current() returns nil until Set.Load succeeds.
 // The loader receives the ctx passed to Set.Load, so long-running parses
 // should honor ctx.Err() to support graceful shutdown.
 //
@@ -272,13 +419,13 @@ func Add[T any](s *Set, loader func(ctx context.Context) (*T, error)) *View[T] {
 	if s.loaded.Load() {
 		panic("dataset: Add called after Load")
 	}
-	v := &View[T]{loader: loader}
+	v := &View[T]{set: s, loader: loader}
 	s.views = append(s.views, v)
 	return v
 }
 
 // AddInitial is like Add but pre-populates the view with initial, so
-// View.Value() returns a usable (possibly empty) value before the first
+// View.Current() returns a usable (possibly empty) value before the first
 // Load completes. Use when the initial state is benign (e.g. an empty
 // cohort matches nothing) and you want to start serving before the
 // first load finishes.
@@ -288,7 +435,7 @@ func AddInitial[T any](s *Set, initial *T, loader func(ctx context.Context) (*T,
 	if s.loaded.Load() {
 		panic("dataset: AddInitial called after Load")
 	}
-	v := &View[T]{loader: loader}
+	v := &View[T]{set: s, loader: loader}
 	v.ptr.Store(initial)
 	s.views = append(s.views, v)
 	return v

sync/dataset/dataset_concurrency_test.go

@@ -0,0 +1,378 @@
+package dataset_test
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/therootcompany/golib/sync/dataset"
+)
+
+// blockingUpstream blocks its Update call until release is closed, then
+// reports updated and err. It records how many times Update was entered.
+type blockingUpstream struct {
+	entered atomic.Int32
+	release chan struct{}
+	updated bool
+	err     error
+}
+
+func newBlockingUpstream() *blockingUpstream {
+	return &blockingUpstream{release: make(chan struct{})}
+}
+
+func (b *blockingUpstream) Update(ctx context.Context) (bool, error) {
+	b.entered.Add(1)
+	select {
+	case <-ctx.Done():
+		return false, ctx.Err()
+	case <-b.release:
+	}
+	return b.updated, b.err
+}
+
+func (b *blockingUpstream) unblock() { close(b.release) }
+
+// TestSet_RevalidateCoalescesConcurrentCallers confirms that many concurrent
+// Revalidate calls result in exactly one upstream Update while the first is
+// in flight, and that the rest report started=false.
+func TestSet_RevalidateCoalescesConcurrentCallers(t *testing.T) {
+	up := newBlockingUpstream()
+	s := dataset.NewSet(up)
+	dataset.Add(s, func(_ context.Context) (*int, error) {
+		v := 1
+		return &v, nil
+	})
+
+	ctx := t.Context()
+	startedCount := atomic.Int32{}
+	const n = 16
+	var wg sync.WaitGroup
+	wg.Add(n)
+	start := make(chan struct{})
+	for range n {
+		go func() {
+			defer wg.Done()
+			<-start
+			started, err := s.Revalidate(ctx)
+			if err != nil {
+				t.Errorf("Revalidate: %v", err)
+				return
+			}
+			if started {
+				startedCount.Add(1)
+			}
+		}()
+	}
+	close(start)
+
+	// While the upstream is blocked, exactly one Update should have been
+	// entered; the other callers are waiting on the coalesced refresh.
+	time.Sleep(30 * time.Millisecond)
+	if got := up.entered.Load(); got != 1 {
+		t.Fatalf("upstream entered %d times during in-flight refresh, want 1", got)
+	}
+
+	up.unblock()
+	wg.Wait()
+
+	if startedCount.Load() != 1 {
+		t.Errorf("exactly one caller should have started the refresh, got %d", startedCount.Load())
+	}
+}
+
+// TestSet_LoadFalseDoesNotBlock confirms that after the first load, a
+// Load(wait=false) returns immediately even while a refresh is running.
+func TestSet_LoadFalseDoesNotBlock(t *testing.T) {
+	up := newBlockingUpstream()
+	s := dataset.NewSet(up)
+	v := dataset.Add(s, func(_ context.Context) (*int, error) {
+		n := 1
+		return &n, nil
+	})
+
+	// Initial load (blocks until we release).
+	ctx := t.Context()
+	loadErr := make(chan error, 1)
+	go func() {
+		loadErr <- s.Load(ctx, true)
+	}()
+	time.Sleep(20 * time.Millisecond)
+	up.unblock()
+	if err := <-loadErr; err != nil {
+		t.Fatalf("initial load: %v", err)
+	}
+	if got := v.Current(); got == nil || *got != 1 {
+		t.Fatalf("after initial load Current = %v", got)
+	}
+
+	// Start a second refresh that blocks.
+	up.release = make(chan struct{})
+	if started, _ := s.Revalidate(ctx); !started {
+		t.Fatal("Revalidate should start")
+	}
+	time.Sleep(20 * time.Millisecond)
+
+	// Load(false) must return immediately with the current value without
+	// blocking, even though the refresh is still running.
+	done := make(chan error, 1)
+	go func() {
+		_, err := v.Load(ctx, false)
+		done <- err
+	}()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Errorf("Load(false) returned err %v; want nil", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Load(false) blocked while refresh was running")
+	}
+	up.unblock()
+}
+
+// TestSet_LoadTrueWaitsForRunningRefresh confirms Load(wait=true) waits for an
+// already-running refresh to complete.
+func TestSet_LoadTrueWaitsForRunningRefresh(t *testing.T) {
+	up := newBlockingUpstream()
+	up.updated = true
+	s := dataset.NewSet(up)
+	v := dataset.Add(s, func(_ context.Context) (*int, error) {
+		n := 7
+		return &n, nil
+	})
+	ctx := t.Context()
+
+	// Initial load.
+	go func() { _ = s.Load(ctx, true) }()
+	time.Sleep(20 * time.Millisecond)
+	up.unblock()
+	if err := s.Load(ctx, true); err != nil {
+		t.Fatalf("drain initial: %v", err)
+	}
+
+	// Start a blocking refresh.
+	up.release = make(chan struct{})
+	if started, _ := s.Revalidate(ctx); !started {
+		t.Fatal("Revalidate should start")
+	}
+	time.Sleep(20 * time.Millisecond)
+
+	// A concurrent Load(true) must block until the refresh finishes.
+	done := make(chan struct{})
+	go func() {
+		_, _ = v.Load(ctx, true)
+		close(done)
+	}()
+	select {
+	case <-done:
+		t.Fatal("Load(true) returned before refresh finished")
+	case <-time.After(30 * time.Millisecond):
+	}
+	up.unblock()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("Load(true) did not return after refresh finished")
+	}
+}
+
+// TestSet_SetDuringRefreshDoesNotGetClobbered confirms the generation guard:
+// a manual Set while a refresh is in flight prevents the stale prepared value
+// from publishing over the manually-set value.
+func TestSet_SetDuringRefreshDoesNotGetClobbered(t *testing.T) {
+	up := newBlockingUpstream()
+	up.updated = true
+	s := dataset.NewSet(up)
+	v := dataset.Add(s, func(_ context.Context) (*int, error) {
+		n := 999 // stale value the in-flight refresh would publish
+		return &n, nil
+	})
+	ctx := t.Context()
+
+	// Start a refresh and let it enter the upstream (blocking).
+	if started, _ := s.Revalidate(ctx); !started {
+		t.Fatal("Revalidate should start")
+	}
+	time.Sleep(20 * time.Millisecond)
+
+	// Manually set a value while the refresh is in flight.
+	manual := 42
+	if err := v.Set(&manual); err != nil {
+		t.Fatalf("Set: %v", err)
+	}
+	if got := v.Current(); got == nil || *got != 42 {
+		t.Fatalf("after manual Set Current = %v, want 42", got)
+	}
+
+	// Release the stale refresh; it must not overwrite the manual value.
+	up.unblock()
+	if err := s.Load(ctx, true); err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	if got := v.Current(); got == nil || *got != 42 {
+		t.Errorf("after stale refresh Current = %v, want 42 (generation guard)", got)
+	}
+}
+
+// TestSet_ClearDuringRefreshPreventsPublication confirms Clear bumps the
+// generation so an in-flight refresh does not republish a cleared value.
+func TestSet_ClearDuringRefreshPreventsPublication(t *testing.T) {
+	up := newBlockingUpstream()
+	up.updated = true
+	s := dataset.NewSet(up)
+	v := dataset.Add(s, func(_ context.Context) (*int, error) {
+		n := 5
+		return &n, nil
+	})
+	ctx := t.Context()
+
+	// Initial load.
+	go func() { _ = s.Load(ctx, true) }()
+	time.Sleep(20 * time.Millisecond)
+	up.unblock()
+	if err := s.Load(ctx, true); err != nil {
+		t.Fatalf("drain initial: %v", err)
+	}
+
+	// Start a blocking refresh, then Clear while it's in flight.
+	up.release = make(chan struct{})
+	if started, _ := s.Revalidate(ctx); !started {
+		t.Fatal("Revalidate should start")
+	}
+	time.Sleep(20 * time.Millisecond)
+
+	if err := s.Clear(); err != nil {
+		t.Fatalf("Clear: %v", err)
+	}
+	if v.Current() != nil {
+		t.Fatal("Current should be nil after Clear")
+	}
+
+	up.unblock()
+	// Give the in-flight refresh time to (not) publish.
+	if err := s.Load(ctx, true); err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	if v.Current() != nil {
+		t.Errorf("Current = %v after Clear+stale refresh, want nil (generation guard)", v.Current())
+	}
+}
+
+// TestSet_TransactionalMultiViewPublication confirms that when one view's
+// loader fails, no view publishes a partial update.
+func TestSet_TransactionalMultiViewPublication(t *testing.T) {
+	s := dataset.NewSet(dataset.NopUpstream{})
+	a := dataset.Add(s, func(_ context.Context) (*int, error) {
+		v := 1
+		return &v, nil
+	})
+	b := dataset.Add(s, func(_ context.Context) (*string, error) {
+		return nil, errors.New("b broke")
+	})
+
+	ctx := t.Context()
+	// Initial load fails because b errors.
+	if err := s.Load(ctx, true); err == nil {
+		t.Fatal("expected loader error on initial load")
+	}
+	// Neither view should have published.
+	if a.Current() != nil {
+		t.Errorf("view a published = %v, want nil (transactional)", a.Current())
+	}
+	if b.Current() != nil {
+		t.Errorf("view b published = %v, want nil (transactional)", b.Current())
+	}
+}
+
+// TestSet_TransactionalMultiViewPublishesTogether confirms that on success,
+// all views publish atomically: a concurrent reader sees either all-old or
+// all-new, never a mix.
+func TestSet_TransactionalMultiViewPublishesTogether(t *testing.T) {
+	up := newBlockingUpstream()
+	up.updated = true
+	s := dataset.NewSet(up)
+	a := dataset.Add(s, func(_ context.Context) (*int, error) {
+		v := 1
+		return &v, nil
+	})
+	b := dataset.Add(s, func(_ context.Context) (*string, error) {
+		v := "one"
+		return &v, nil
+	})
+	ctx := t.Context()
+
+	// Initial load.
+	go func() { _ = s.Load(ctx, true) }()
+	time.Sleep(20 * time.Millisecond)
+	up.unblock()
+	if err := s.Load(ctx, true); err != nil {
+		t.Fatalf("drain initial: %v", err)
+	}
+
+	// Second refresh publishes new values; both should swap together.
+	up.release = make(chan struct{})
+	// Swap loaders is not possible after Add, so we rely on updated=true to
+	// re-run the same loaders (they return the same values). The transactional
+	// guarantee is that publish happens under publishMu for all views at once.
+	// We assert the invariant by reading both atomically from a racer and
+	// checking consistency.
+	if started, _ := s.Revalidate(ctx); !started {
+		t.Fatal("Revalidate should start")
+	}
+
+	var inconsistent atomic.Bool
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		for range 1000 {
+			av := a.Current()
+			bv := b.Current()
+			// Both nil or both non-nil is the only consistent state once
+			// loaded. (Values are identical across refreshes here, so we only
+			// check presence consistency.)
+			if (av == nil) != (bv == nil) {
+				inconsistent.Store(true)
+				return
+			}
+		}
+	})
+	up.unblock()
+	wg.Wait()
+	if inconsistent.Load() {
+		t.Error("observed mixed nil/non-nil across views during publication")
+	}
+}
+
+// TestSet_DetachedContextSurvivesCallerCancel confirms that cancelling the
+// caller's context after Revalidate returns does not cancel the background
+// refresh (it uses context.WithoutCancel).
+func TestSet_DetachedContextSurvivesCallerCancel(t *testing.T) {
+	up := newBlockingUpstream()
+	s := dataset.NewSet(up)
+	dataset.Add(s, func(_ context.Context) (*int, error) {
+		v := 1
+		return &v, nil
+	})
+
+	ctx, cancel := context.WithCancel(t.Context())
+	if started, err := s.Revalidate(ctx); err != nil || !started {
+		t.Fatalf("Revalidate started=%v err=%v", started, err)
+	}
+	// Cancel the caller context immediately.
+	cancel()
+	// The upstream should still be blocked (not cancelled). Give it a moment
+	// to ensure the detached context didn't propagate cancellation.
+	time.Sleep(30 * time.Millisecond)
+	if got := up.entered.Load(); got != 1 {
+		t.Fatalf("upstream entered %d times, want 1", got)
+	}
+
+	// Releasing the upstream should let the refresh complete normally.
+	up.unblock()
+	if err := s.Load(t.Context(), true); err != nil {
+		t.Fatalf("Load after cancel: %v", err)
+	}
+}

sync/dataset/dataset_test.go

@@ -11,19 +11,19 @@ import (
 	"github.com/therootcompany/golib/sync/dataset"
 )
 
-type countFetcher struct {
+type countUpdater struct {
 	calls   atomic.Int32
 	updated bool
 	err     error
 }
 
-func (f *countFetcher) Fetch(_ context.Context) (bool, error) {
+func (f *countUpdater) Update(_ context.Context) (bool, error) {
 	f.calls.Add(1)
 	return f.updated, f.err
 }
 
 func TestSet_LoadPopulatesAllViews(t *testing.T) {
-	f := &countFetcher{}
+	f := &countUpdater{}
 	g := dataset.NewSet(f)
 
 	var aCalls, bCalls int
@@ -38,7 +38,7 @@ func TestSet_LoadPopulatesAllViews(t *testing.T) {
 		return &v, nil
 	})
 
-	if err := g.Load(t.Context()); err != nil {
+	if err := g.Load(t.Context(), true); err != nil {
 		t.Fatal(err)
 	}
 	if f.calls.Load() != 1 {
@@ -47,16 +47,16 @@ func TestSet_LoadPopulatesAllViews(t *testing.T) {
 	if aCalls != 1 || bCalls != 1 {
 		t.Errorf("loaders called (%d,%d), want (1,1)", aCalls, bCalls)
 	}
-	if got := a.Value(); got == nil || *got != "a" {
+	if got := a.Current(); got == nil || *got != "a" {
-		t.Errorf("a.Value() = %v", got)
+		t.Errorf("a.Current() = %v", got)
 	}
-	if got := b.Value(); got == nil || *got != 42 {
+	if got := b.Current(); got == nil || *got != 42 {
-		t.Errorf("b.Value() = %v", got)
+		t.Errorf("b.Current() = %v", got)
 	}
 }
 
 func TestSet_SecondLoadSkipsUnchanged(t *testing.T) {
-	f := &countFetcher{updated: false}
+	f := &countUpdater{updated: false}
 	g := dataset.NewSet(f)
 	calls := 0
 	dataset.Add(g, func(_ context.Context) (*string, error) {
@@ -64,13 +64,13 @@ func TestSet_SecondLoadSkipsUnchanged(t *testing.T) {
 		v := "x"
 		return &v, nil
 	})
-	if err := g.Load(t.Context()); err != nil {
+	if err := g.Load(t.Context(), true); err != nil {
 		t.Fatal(err)
 	}
 	if calls != 1 {
 		t.Fatalf("initial load ran loader %d times, want 1", calls)
 	}
-	if err := g.Load(t.Context()); err != nil {
+	if err := g.Load(t.Context(), true); err != nil {
 		t.Fatal(err)
 	}
 	if calls != 1 {
@@ -79,42 +79,42 @@ func TestSet_SecondLoadSkipsUnchanged(t *testing.T) {
 }
 
 func TestSet_LoadOnUpdateSwaps(t *testing.T) {
-	f := &countFetcher{updated: true}
+	f := &countUpdater{updated: true}
 	g := dataset.NewSet(f)
 	n := 0
 	v := dataset.Add(g, func(_ context.Context) (*int, error) {
 		n++
 		return &n, nil
 	})
-	if err := g.Load(t.Context()); err != nil {
+	if err := g.Load(t.Context(), true); err != nil {
 		t.Fatal(err)
 	}
-	if err := g.Load(t.Context()); err != nil {
+	if err := g.Load(t.Context(), true); err != nil {
 		t.Fatal(err)
 	}
-	if got := v.Value(); got == nil || *got != 2 {
+	if got := v.Current(); got == nil || *got != 2 {
-		t.Errorf("v.Value() = %v, want 2", got)
+		t.Errorf("v.Current() = %v, want 2", got)
 	}
 }
 
 func TestSet_FetchError(t *testing.T) {
-	f := &countFetcher{err: errors.New("offline")}
+	f := &countUpdater{err: errors.New("offline")}
 	g := dataset.NewSet(f)
 	dataset.Add(g, func(_ context.Context) (*string, error) {
 		s := "x"
 		return &s, nil
 	})
-	if err := g.Load(t.Context()); err == nil {
+	if err := g.Load(t.Context(), true); err == nil {
 		t.Error("expected fetch error")
 	}
 }
 
 func TestSet_LoaderError(t *testing.T) {
-	g := dataset.NewSet(dataset.NopFetcher{})
+	g := dataset.NewSet(dataset.NopUpstream{})
 	dataset.Add(g, func(_ context.Context) (*string, error) {
 		return nil, errors.New("parse fail")
 	})
-	if err := g.Load(t.Context()); err == nil {
+	if err := g.Load(t.Context(), true); err == nil {
 		t.Error("expected loader error")
 	}
 }
@@ -132,10 +132,10 @@ func TestPollFiles(t *testing.T) {
 
 	p := dataset.PollFiles(a, b)
 
-	if u, err := p.Fetch(t.Context()); err != nil || !u {
+	if u, err := p.Update(t.Context()); err != nil || !u {
 		t.Fatalf("first Fetch: updated=%v err=%v, want true/nil", u, err)
 	}
-	if u, err := p.Fetch(t.Context()); err != nil || u {
+	if u, err := p.Update(t.Context()); err != nil || u {
 		t.Fatalf("unchanged Fetch: updated=%v err=%v, want false/nil", u, err)
 	}
 
@@ -147,87 +147,185 @@ func TestPollFiles(t *testing.T) {
 	if err := os.Chtimes(b, future, future); err != nil {
 		t.Fatal(err)
 	}
-	if u, err := p.Fetch(t.Context()); err != nil || !u {
+	if u, err := p.Update(t.Context()); err != nil || !u {
 		t.Errorf("after change: updated=%v err=%v, want true/nil", u, err)
 	}
-	if u, err := p.Fetch(t.Context()); err != nil || u {
+	if u, err := p.Update(t.Context()); err != nil || u {
 		t.Errorf("steady Fetch: updated=%v err=%v, want false/nil", u, err)
 	}
 }
 
-func TestSet_ClosePreventsTick(t *testing.T) {
+func TestSet_DueGatesRevalidateByInterval(t *testing.T) {
-	f := &countFetcher{updated: true}
+	f := &countUpdater{updated: false}
+	s := dataset.NewSet(f)
+	s.RefreshInterval = 30 * time.Millisecond
+	dataset.Add(s, func(_ context.Context) (*int, error) {
+		n := 1
+		return &n, nil
+	})
+
+	if err := s.Load(t.Context(), true); err != nil {
+		t.Fatal(err)
+	}
+	first := f.calls.Load()
+	if first != 1 {
+		t.Fatalf("initial load calls = %d, want 1", first)
+	}
+
+	// Within the interval, Revalidate must not start a refresh.
+	for range 3 {
+		if started, err := s.Revalidate(t.Context()); err != nil {
+			t.Fatalf("Revalidate: %v", err)
+		} else if started {
+			t.Error("Revalidate started work before StaleAt")
+		}
+	}
+	if f.calls.Load() != first {
+		t.Errorf("upstream checked %d times within interval, want %d", f.calls.Load(), first)
+	}
+
+	// After the interval elapses, Revalidate starts a refresh.
+	time.Sleep(35 * time.Millisecond)
+	if started, err := s.Revalidate(t.Context()); err != nil {
+		t.Fatalf("Revalidate after stale: %v", err)
+	} else if !started {
+		t.Error("Revalidate should start work after StaleAt")
+	}
+	if err := s.Load(t.Context(), true); err != nil {
+		t.Fatal(err)
+	}
+	if f.calls.Load() != first+1 {
+		t.Errorf("upstream checked %d times after stale, want %d", f.calls.Load(), first+1)
+	}
+}
+
+func TestSet_DueAlwaysTrueWithoutInterval(t *testing.T) {
+	f := &countUpdater{updated: false}
+	s := dataset.NewSet(f) // RefreshInterval == 0
+	dataset.Add(s, func(_ context.Context) (*int, error) {
+		n := 1
+		return &n, nil
+	})
+	if err := s.Load(t.Context(), true); err != nil {
+		t.Fatal(err)
+	}
+	// With no interval, every Revalidate is due and starts a refresh (coalesced
+	// if one is already running). Upstreams own their own dedup.
+	if started, err := s.Revalidate(t.Context()); err != nil {
+		t.Fatalf("Revalidate: %v", err)
+	} else if !started {
+		t.Error("Revalidate should start with zero interval")
+	}
+	if err := s.Load(t.Context(), true); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSet_ClearStopsTicker(t *testing.T) {
+	f := &countUpdater{updated: true}
 	s := dataset.NewSet(f)
 	v := dataset.Add(s, func(_ context.Context) (*int, error) {
 		n := 1
 		return &n, nil
 	})
-	if err := s.Load(t.Context()); err != nil {
+	if err := s.Load(t.Context(), true); err != nil {
 		t.Fatal(err)
 	}
 
 	ctx, cancel := context.WithCancel(t.Context())
-	go s.Tick(ctx, 10*time.Millisecond, nil)
+	s.Start(ctx, 10*time.Millisecond)
 	time.Sleep(30 * time.Millisecond) // let a few ticks fire
 
-	// Close while Tick is still running.
+	// Cancel the ticker before clearing so no new refresh starts after Clear.
-	if err := s.Close(); err != nil {
+	cancel()
+	time.Sleep(20 * time.Millisecond) // let in-flight refresh drain
+
+	if err := s.Clear(); err != nil {
 		t.Fatal(err)
 	}
-	cancel()
+	time.Sleep(30 * time.Millisecond) // no more ticks should fire
-	time.Sleep(30 * time.Millisecond) // let more ticks fire
 
-	// Value should be nil after Close.
+	// Current should be nil after Clear.
-	if v.Value() != nil {
+	if v.Current() != nil {
-		t.Error("Value() should be nil after Close")
+		t.Error("Current() should be nil after Clear")
 	}
-	// After Close, no further fetches should occur. The count should not
+	// After cancellation, no further fetches should occur.
-	// have increased between the Close and the final sleep.
 	finalCalls := f.calls.Load()
-	// Give the scheduler a moment to drain any in-flight Load calls.
 	time.Sleep(10 * time.Millisecond)
 	if f.calls.Load() != finalCalls {
-		t.Errorf("fetch count increased after Close (%d → %d)", finalCalls, f.calls.Load())
+		t.Errorf("fetch count increased after cancel (%d → %d)", finalCalls, f.calls.Load())
 	}
 }
 
-func TestSet_CloseIsIdempotent(t *testing.T) {
+func TestSet_ClearIsIdempotent(t *testing.T) {
-	s := dataset.NewSet(dataset.NopFetcher{})
+	s := dataset.NewSet(dataset.NopUpstream{})
 	dataset.Add(s, func(_ context.Context) (*string, error) {
 		s := "x"
 		return &s, nil
 	})
-	if err := s.Load(t.Context()); err != nil {
+	if err := s.Load(t.Context(), true); err != nil {
 		t.Fatal(err)
 	}
-	if err := s.Close(); err != nil {
+	if err := s.Clear(); err != nil {
 		t.Fatal(err)
 	}
-	if err := s.Close(); err != nil {
+	if err := s.Clear(); err != nil {
-		t.Error("second Close should be a no-op")
+		t.Error("second Clear should be a no-op")
 	}
-	if err := s.Load(t.Context()); err == nil {
+	if err := s.Load(t.Context(), true); err != nil {
-		t.Error("Load after Close should error")
+		t.Fatalf("Load after Clear: %v", err)
 	}
 }
 
-func TestSet_CloseNilsViews(t *testing.T) {
+func TestSet_ClearThenLoadRepublishes(t *testing.T) {
-	s := dataset.NewSet(dataset.NopFetcher{})
+	s := dataset.NewSet(dataset.NopUpstream{})
 	v := dataset.Add(s, func(_ context.Context) (*int, error) {
 		n := 42
 		return &n, nil
 	})
-	if err := s.Load(t.Context()); err != nil {
+	if err := s.Load(t.Context(), true); err != nil {
 		t.Fatal(err)
 	}
-	if v.Value() == nil || *v.Value() != 42 {
+	if v.Current() == nil || *v.Current() != 42 {
-		t.Fatal("Value() should be 42 after Load")
+		t.Fatal("expected 42 after initial load")
 	}
-	if err := s.Close(); err != nil {
+	if err := s.Clear(); err != nil {
 		t.Fatal(err)
 	}
-	if v.Value() != nil {
+	if v.Current() != nil {
-		t.Error("Value() should be nil after Close")
+		t.Fatal("expected nil after Clear")
+	}
+	// After Clear, Load must repopulate the view even when the upstream
+	// reports no change (NopUpstream returns false). Previously, loaded
+	// stayed true after Clear, so update skipped reloading.
+	if err := s.Load(t.Context(), true); err != nil {
+		t.Fatalf("Load after Clear: %v", err)
+	}
+	if v.Current() == nil {
+		t.Fatal("view is nil after Load following Clear")
+	}
+	if *v.Current() != 42 {
+		t.Errorf("got %d, want 42", *v.Current())
+	}
+}
+
+func TestSet_ClearNilsViews(t *testing.T) {
+	s := dataset.NewSet(dataset.NopUpstream{})
+	v := dataset.Add(s, func(_ context.Context) (*int, error) {
+		n := 42
+		return &n, nil
+	})
+	if err := s.Load(t.Context(), true); err != nil {
+		t.Fatal(err)
+	}
+	if v.Current() == nil || *v.Current() != 42 {
+		t.Fatal("Current() should be 42 after Load")
+	}
+	if err := s.Clear(); err != nil {
+		t.Fatal(err)
+	}
+	if v.Current() != nil {
+		t.Error("Current() should be nil after Clear")
 	}
 }
 
@@ -237,12 +335,12 @@ func TestSet_AddAfterLoadPanic(t *testing.T) {
 			t.Error("Add after Load should panic")
 		}
 	}()
-	s := dataset.NewSet(dataset.NopFetcher{})
+	s := dataset.NewSet(dataset.NopUpstream{})
 	dataset.Add(s, func(_ context.Context) (*string, error) {
 		s := "x"
 		return &s, nil
 	})
-	if err := s.Load(t.Context()); err != nil {
+	if err := s.Load(t.Context(), true); err != nil {
 		t.Fatal(err)
 	}
 	dataset.Add(s, func(_ context.Context) (*string, error) {
@@ -257,12 +355,12 @@ func TestSet_AddInitialAfterLoadPanic(t *testing.T) {
 			t.Error("AddInitial after Load should panic")
 		}
 	}()
-	s := dataset.NewSet(dataset.NopFetcher{})
+	s := dataset.NewSet(dataset.NopUpstream{})
 	dataset.Add(s, func(_ context.Context) (*string, error) {
 		s := "x"
 		return &s, nil
 	})
-	if err := s.Load(t.Context()); err != nil {
+	if err := s.Load(t.Context(), true); err != nil {
 		t.Fatal(err)
 	}
 	var x string

sync/dataset/go.mod

@@ -1,3 +1,5 @@
 module github.com/therootcompany/golib/sync/dataset
 
 go 1.26.0
+
+require github.com/therootcompany/golib/sync/cachable v0.1.0

sync/dataset/go.sum

@@ -0,0 +1,2 @@
+github.com/therootcompany/golib/sync/cachable v0.1.0 h1:swCXtXB116jhSmla81HaXi/9cPsqFI+Oq/6OcSfdU5Q=
+github.com/therootcompany/golib/sync/cachable v0.1.0/go.mod h1:l/NZdGiSbW5CkLlkkRa0yObkqak2sj/jQTXXqXwvCl0=