package dataset_test import ( "context" "errors" "os" "sync/atomic" "testing" "time" "github.com/therootcompany/golib/sync/dataset" ) type countUpdater struct { calls atomic.Int32 updated bool err 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 := &countUpdater{} g := dataset.NewSet(f) var aCalls, bCalls int a := dataset.Add(g, func(_ context.Context) (*string, error) { aCalls++ v := "a" return &v, nil }) b := dataset.Add(g, func(_ context.Context) (*int, error) { bCalls++ v := 42 return &v, nil }) if err := g.Load(t.Context(), true); err != nil { t.Fatal(err) } if f.calls.Load() != 1 { t.Errorf("Fetch called %d times, want 1", f.calls.Load()) } if aCalls != 1 || bCalls != 1 { t.Errorf("loaders called (%d,%d), want (1,1)", aCalls, bCalls) } if got := a.Current(); got == nil || *got != "a" { t.Errorf("a.Current() = %v", got) } if got := b.Current(); got == nil || *got != 42 { t.Errorf("b.Current() = %v", got) } } func TestSet_SecondLoadSkipsUnchanged(t *testing.T) { f := &countUpdater{updated: false} g := dataset.NewSet(f) calls := 0 dataset.Add(g, func(_ context.Context) (*string, error) { calls++ v := "x" return &v, 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(), true); err != nil { t.Fatal(err) } if calls != 1 { t.Errorf("second load ran loader %d times, want 1 (no upstream change)", calls) } } func TestSet_LoadOnUpdateSwaps(t *testing.T) { 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(), true); err != nil { t.Fatal(err) } if err := g.Load(t.Context(), true); err != nil { t.Fatal(err) } if got := v.Current(); got == nil || *got != 2 { t.Errorf("v.Current() = %v, want 2", got) } } func TestSet_FetchError(t *testing.T) { 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(), true); err == nil { t.Error("expected fetch error") } } func TestSet_LoaderError(t *testing.T) { 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(), true); err == nil { t.Error("expected loader error") } } func TestPollFiles(t *testing.T) { dir := t.TempDir() a := dir + "/a.txt" b := dir + "/b.txt" if err := os.WriteFile(a, []byte("1"), 0o644); err != nil { t.Fatal(err) } if err := os.WriteFile(b, []byte("2"), 0o644); err != nil { t.Fatal(err) } p := dataset.PollFiles(a, b) 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.Update(t.Context()); err != nil || u { t.Fatalf("unchanged Fetch: updated=%v err=%v, want false/nil", u, err) } // Bump mtime + change contents on b. future := time.Now().Add(2 * time.Second) if err := os.WriteFile(b, []byte("22"), 0o644); err != nil { t.Fatal(err) } if err := os.Chtimes(b, future, future); err != nil { t.Fatal(err) } 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.Update(t.Context()); err != nil || u { t.Errorf("steady Fetch: updated=%v err=%v, want false/nil", u, err) } } 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(), true); err != nil { t.Fatal(err) } ctx, cancel := context.WithCancel(t.Context()) s.Start(ctx, 10*time.Millisecond) time.Sleep(30 * time.Millisecond) // let a few ticks fire // 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) } time.Sleep(30 * time.Millisecond) // no more ticks should fire // Current should be nil after Clear. if v.Current() != nil { t.Error("Current() should be nil after Clear") } // After cancellation, no further fetches should occur. finalCalls := f.calls.Load() time.Sleep(10 * time.Millisecond) if f.calls.Load() != finalCalls { t.Errorf("fetch count increased after cancel (%d → %d)", finalCalls, f.calls.Load()) } } 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(), true); err != nil { t.Fatal(err) } if err := s.Clear(); err != nil { t.Fatal(err) } if err := s.Clear(); err != nil { t.Error("second Clear should be a no-op") } if err := s.Load(t.Context(), true); err != nil { t.Fatalf("Load after Clear: %v", err) } } 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(), true); err != nil { t.Fatal(err) } if v.Current() == nil || *v.Current() != 42 { t.Fatal("expected 42 after initial load") } if err := s.Clear(); err != nil { t.Fatal(err) } 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") } } func TestSet_AddAfterLoadPanic(t *testing.T) { defer func() { if r := recover(); r == nil { t.Error("Add after Load should panic") } }() s := dataset.NewSet(dataset.NopUpstream{}) dataset.Add(s, func(_ context.Context) (*string, error) { s := "x" return &s, nil }) if err := s.Load(t.Context(), true); err != nil { t.Fatal(err) } dataset.Add(s, func(_ context.Context) (*string, error) { s := "y" return &s, nil }) } func TestSet_AddInitialAfterLoadPanic(t *testing.T) { defer func() { if r := recover(); r == nil { t.Error("AddInitial after Load should panic") } }() s := dataset.NewSet(dataset.NopUpstream{}) dataset.Add(s, func(_ context.Context) (*string, error) { s := "x" return &s, nil }) if err := s.Load(t.Context(), true); err != nil { t.Fatal(err) } var x string dataset.AddInitial(s, &x, func(_ context.Context) (*string, error) { s := "y" return &s, nil }) }