From c43fd48b340315d436af266ef5d3a1af9a573d3f Mon Sep 17 00:00:00 2001 From: AJ ONeal Date: Wed, 9 Sep 2026 22:32:31 -0600 Subject: [PATCH] refactor(sync/dataset): use released cachable API --- sync/dataset/dataset.go | 357 ++++++++++++++------- sync/dataset/dataset_concurrency_test.go | 378 +++++++++++++++++++++++ sync/dataset/dataset_test.go | 216 +++++++++---- sync/dataset/go.mod | 2 + sync/dataset/go.sum | 2 + 5 files changed, 791 insertions(+), 164 deletions(-) create mode 100644 sync/dataset/dataset_concurrency_test.go create mode 100644 sync/dataset/go.sum diff --git a/sync/dataset/dataset.go b/sync/dataset/dataset.go index 6911efa..966b3b4 100644 --- a/sync/dataset/dataset.go +++ b/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 -// go s.Tick(ctx, 47*time.Minute, onError) // background refresh -// current := inbound.Value() // lock-free read +// if err := s.Load(ctx, true); err != nil { ... } // initial populate +// inbound.Start(ctx, 47*time.Minute) // optional refresh +// 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. -type Fetcher interface { - Fetch(ctx context.Context) (updated bool, err error) +// honor ctx so asynchronous updates can cancel on shutdown. +type Upstream interface { + Update(ctx context.Context) (updated bool, err error) } -// FetcherFunc adapts a plain function to Fetcher. -type FetcherFunc func(ctx context.Context) (bool, error) +// UpstreamFunc adapts a plain function to Upstream. +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 -// /admin/reload handler firing while Tick also fires) don't race the views. +// Set serializes snapshot publication so concurrent callers do not race views. // Add and AddInitial mutate s.views without locking — they MUST be called // before the first Load. type Set struct { - mu sync.Mutex - closed atomic.Bool // set by Close; Load returns early when true - fetchers []Fetcher - views []reloader - loaded atomic.Bool + publishMu sync.Mutex + fetchers []Upstream + 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 -// reports a change, reloads every view and atomically installs the new values. -// Returns context.Canceled if the set has been closed. -func (s *Set) Load(ctx context.Context) error { - if s.closed.Load() { - return context.Canceled +// Due reports whether the set should be revalidated. With RefreshInterval > 0 +// it returns false until StaleAt (loadedAt + RefreshInterval) has passed, so a +// hot path calling Load(false) does not start a refresh on every request. +// With RefreshInterval == 0 it always returns true: the upstreams own their +// own dedup windows (httpcache ETag, gitshallow pull throttling) and a refresh +// 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() - defer s.mu.Unlock() + if s.RefreshInterval <= 0 { + 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 -// pointers. Call on shutdown to release any OS resources held by the -// final snapshots (file handles, network connections). Safe to call on a -// Set that hasn't been loaded or whose views hold pure in-memory values — -// non-Closer values are skipped. Idempotent. -// -// It is safe to call Close while Tick is still running; subsequent Load -// calls will return context.Canceled. -func (s *Set) Close() error { - if s.closed.Swap(true) { - return nil +// Load performs the initial load, or revalidates an already-loaded set. It +// keeps the last good snapshot when a later revalidation fails. +func (s *Set) Load(ctx context.Context, wait bool) error { + _, err := s.Revalidate(ctx) + if err != nil { + return err } - 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() + if !s.loaded.Load() || wait { + return s.refresh.Wait(ctx) } - return errors.Join(errs...) + return nil } -// Tick calls Load every interval until ctx is done. Load errors are passed to -// onError (if non-nil) and do not stop the loop; callers choose whether to log, -// count, page, or ignore. Run in a goroutine: `go s.Tick(ctx, d, onError)`. -func (s *Set) Tick(ctx context.Context, interval time.Duration, onError func(error)) { - t := time.NewTicker(interval) - defer t.Stop() - for { - select { - case <-ctx.Done(): - return - case <-t.C: - if err := s.Load(ctx); err != nil && onError != nil { - onError(err) - } +// Start starts at most one optional background revalidation ticker. +func (s *Set) Start(ctx context.Context, interval time.Duration) { + s.start.Do(func() { + if interval <= 0 { + interval = time.Hour } + tickCtx, cancel := context.WithCancel(ctx) + s.cancel = cancel + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-tickCtx.Done(): + return + case <-ticker.C: + _, _ = s.Revalidate(tickCtx) + } + } + }() + }) +} + +// 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 -// unless the view was registered via AddInitial. -func (v *View[T]) Value() *T { +var ( + _ cachable.Cacheable[any] = (*View[any])(nil) + _ 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 -// zero time if the view has never loaded. -func (v *View[T]) LoadedAt() time.Time { - if t := v.loadedAt.Load(); t != nil { - return *t - } - return time.Time{} +// Due reports whether the owning set should be revalidated. It delegates to +// Set.Due, which honors RefreshInterval when set and otherwise treats the set +// as always due (upstreams own their own dedup windows). +func (v *View[T]) Due(ctx context.Context) (bool, error) { + return v.set.Due(ctx) } -// Close clears the view and closes the currently-held value if it -// implements io.Closer. Idempotent — subsequent calls are no-ops. -func (v *View[T]) Close() error { - prev := v.ptr.Swap(nil) - if prev == nil { - return nil +// Revalidate asks the owning set to check its upstreams and returns whether +// this caller started the refresh. +func (v *View[T]) Revalidate(ctx context.Context) (bool, error) { + return v.set.Revalidate(ctx) +} + +// 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 } - if closer, ok := any(prev).(io.Closer); ok { - return closer.Close() + 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 { - t, err := v.loader(ctx) - if err != nil { - return err +func (v *View[T]) Clear() error { + return v.Set(nil) +} + +// 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 diff --git a/sync/dataset/dataset_concurrency_test.go b/sync/dataset/dataset_concurrency_test.go new file mode 100644 index 0000000..16baac2 --- /dev/null +++ b/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) + } +} diff --git a/sync/dataset/dataset_test.go b/sync/dataset/dataset_test.go index 88fa0d3..97dcef3 100644 --- a/sync/dataset/dataset_test.go +++ b/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" { - t.Errorf("a.Value() = %v", got) + if got := a.Current(); got == nil || *got != "a" { + t.Errorf("a.Current() = %v", got) } - if got := b.Value(); got == nil || *got != 42 { - t.Errorf("b.Value() = %v", got) + if got := b.Current(); got == nil || *got != 42 { + 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 { - t.Errorf("v.Value() = %v, want 2", got) + if got := v.Current(); got == nil || *got != 2 { + 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) { - f := &countFetcher{updated: true} +func TestSet_DueGatesRevalidateByInterval(t *testing.T) { + 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. - if err := s.Close(); err != nil { + // Cancel the ticker before clearing so no new refresh starts after Clear. + 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) // let more ticks fire + time.Sleep(30 * time.Millisecond) // no more ticks should fire - // Value should be nil after Close. - if v.Value() != nil { - t.Error("Value() should be nil after Close") + // Current should be nil after Clear. + if v.Current() != nil { + t.Error("Current() should be nil after Clear") } - // After Close, no further fetches should occur. The count should not - // have increased between the Close and the final sleep. + // After cancellation, no further fetches should occur. 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) { - s := dataset.NewSet(dataset.NopFetcher{}) +func TestSet_ClearIsIdempotent(t *testing.T) { + 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 { - t.Error("second Close should be a no-op") + if err := s.Clear(); err != nil { + t.Error("second Clear should be a no-op") } - if err := s.Load(t.Context()); err == nil { - t.Error("Load after Close should error") + if err := s.Load(t.Context(), true); err != nil { + t.Fatalf("Load after Clear: %v", err) } } -func TestSet_CloseNilsViews(t *testing.T) { - s := dataset.NewSet(dataset.NopFetcher{}) +func TestSet_ClearThenLoadRepublishes(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()); err != nil { + if err := s.Load(t.Context(), true); err != nil { t.Fatal(err) } - if v.Value() == nil || *v.Value() != 42 { - t.Fatal("Value() should be 42 after Load") + if v.Current() == nil || *v.Current() != 42 { + 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 { - t.Error("Value() should be nil after Close") + if v.Current() != nil { + 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 diff --git a/sync/dataset/go.mod b/sync/dataset/go.mod index b22cfc7..886ff23 100644 --- a/sync/dataset/go.mod +++ b/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 diff --git a/sync/dataset/go.sum b/sync/dataset/go.sum new file mode 100644 index 0000000..1f7128a --- /dev/null +++ b/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=