diff --git a/net/iplist/go.mod b/net/iplist/go.mod index 02a3996..2405e31 100644 --- a/net/iplist/go.mod +++ b/net/iplist/go.mod @@ -5,7 +5,8 @@ go 1.26.5 require ( github.com/therootcompany/golib/https v0.9.0 github.com/therootcompany/golib/io/transform/gsheet2csv v1.1.0 - github.com/therootcompany/golib/net/httpcache v0.5.1 + github.com/therootcompany/golib/net/httpcache v0.6.0 + github.com/therootcompany/golib/sync/cachable v0.1.0 ) require golang.org/x/sync v0.20.0 // indirect diff --git a/net/iplist/go.sum b/net/iplist/go.sum index 3304d22..601d233 100644 --- a/net/iplist/go.sum +++ b/net/iplist/go.sum @@ -2,7 +2,9 @@ github.com/therootcompany/golib/https v0.9.0 h1:D9gh0K6ljkvF805JUbylITn8SHPMo08q github.com/therootcompany/golib/https v0.9.0/go.mod h1:ilExqR/Uh7oB36oy9mwuNhANcFh+zdCfD3uJiZ11YAY= github.com/therootcompany/golib/io/transform/gsheet2csv v1.1.0 h1:R5zAPun5p3JPlNfcMD9es4LD9sqh6nWDNTflZBeeUJo= github.com/therootcompany/golib/io/transform/gsheet2csv v1.1.0/go.mod h1:CsntOAhhIMdzSAYuwuRZtfkdoiER9+TfSgYOIDWN71o= -github.com/therootcompany/golib/net/httpcache v0.5.1 h1:vQM758tBwPEqCo7ojpSeL3fHqgq29LYmvD2hvDTBfKY= -github.com/therootcompany/golib/net/httpcache v0.5.1/go.mod h1:mKSQJWpGSCDBzWkTCwT7dO1SCVCsFmdkRgKr39fNS1U= +github.com/therootcompany/golib/net/httpcache v0.6.0 h1:qFBnSs6rZrtUQ+fweP4bHF2FQd/Syvs7mRN5nO5YtkY= +github.com/therootcompany/golib/net/httpcache v0.6.0/go.mod h1:mKSQJWpGSCDBzWkTCwT7dO1SCVCsFmdkRgKr39fNS1U= +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= golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= diff --git a/net/iplist/iplist.go b/net/iplist/iplist.go index be9bed4..7a6708d 100644 --- a/net/iplist/iplist.go +++ b/net/iplist/iplist.go @@ -201,7 +201,7 @@ func readSource(ctx context.Context, source, cacheDir string, seen map[string]st cache.Header = header cache.MaxAge = defaultMaxAge cache.MaxBytes = defaultMaxBytes - if _, fetchErr := cache.Fetch(ctx); fetchErr != nil { + if _, fetchErr := cache.Update(ctx); fetchErr != nil { // No previous cache to fall back on — propagate the fetch error. if previousErr != nil { return nil, fmt.Errorf("fetch %s: %w", safeURL(source), fetchErr) diff --git a/net/iplist/source.go b/net/iplist/source.go index 6027dd7..a2ef95f 100644 --- a/net/iplist/source.go +++ b/net/iplist/source.go @@ -9,101 +9,189 @@ import ( "sync" "sync/atomic" "time" + + "github.com/therootcompany/golib/sync/cachable" ) -const DefaultRefreshInterval = time.Hour + 57*time.Minute + 13*time.Second +const ( + DefaultRefreshInterval = time.Hour + 57*time.Minute + 13*time.Second + DefaultRefreshTimeout = 15 * time.Minute +) -type SourceConfig struct { +type IPListConfig struct { Source string CacheDir string HTTPClient *http.Client RefreshInterval time.Duration Optional bool - OnRefresh func(SourceEvent) + RefreshTimeout time.Duration } -type SourceEvent struct { - Entries uint64 - Err error -} - -type sourceSnapshot struct { - entries []string -} - -type Source struct { +type IPList struct { ctx context.Context cancel context.CancelFunc - config SourceConfig - current atomic.Pointer[sourceSnapshot] - observers []func(SourceEvent) - observerMu sync.RWMutex + config IPListConfig + current atomic.Pointer[[]string] + start sync.Once + generation atomic.Uint64 + loadedAt atomic.Pointer[time.Time] + lastErr atomic.Pointer[error] + refresh cachable.Refresh } -func NewSource(ctx context.Context, config SourceConfig) (*Source, error) { +var ( + _ cachable.Cacheable[[]string] = (*IPList)(nil) + _ cachable.Mutable[[]string] = (*IPList)(nil) + _ cachable.Inspectable = (*IPList)(nil) + _ cachable.Tickable = (*IPList)(nil) +) + +func NewIPList(ctx context.Context, config IPListConfig) (*IPList, error) { if config.RefreshInterval <= 0 { config.RefreshInterval = DefaultRefreshInterval } sourceCtx, cancel := context.WithCancel(ctx) - s := &Source{ctx: sourceCtx, cancel: cancel, config: config} - entries, err := s.load() - if err != nil { + s := &IPList{ctx: sourceCtx, cancel: cancel, config: config} + if _, err := s.Load(ctx, true); err != nil { cancel() return nil, err } - s.store(entries) - go s.refreshLoop() return s, nil } -func (s *Source) Entries() []string { - current := s.current.Load() - if current == nil { - return nil - } - return append([]string(nil), current.entries...) +func (s *IPList) Current() *[]string { + return s.current.Load() } -func (s *Source) Refresh() error { - entries, err := s.load() +func (s *IPList) Due(ctx context.Context) (bool, error) { + if err := ctx.Err(); err != nil { + return false, err + } + status := s.Status() + return !status.HasValue || status.StaleAt.IsZero() || !time.Now().Before(status.StaleAt), nil +} + +func (s *IPList) 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.config.RefreshTimeout + if timeout <= 0 { + timeout = DefaultRefreshTimeout + } + refreshCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout) + defer cancel() + s.refresh.Finish(s.update(refreshCtx)) + }() + return true, nil +} + +func (s *IPList) update(ctx context.Context) error { + generation := s.generation.Load() + entries, err := s.load(ctx) if err != nil { - s.emit(SourceEvent{Entries: uint64(len(s.Entries())), Err: err}) + errCopy := err + s.lastErr.Store(&errCopy) return err } - s.store(entries) - s.emit(SourceEvent{Entries: uint64(len(entries))}) + old := s.current.Load() + changed := old == nil || !sameEntries(*old, entries) + if changed && s.generation.Load() == generation { + s.store(entries) + } + s.lastErr.Store(nil) return nil } -func (s *Source) Subscribe(observer func(SourceEvent)) { - if observer == nil { - return +func (s *IPList) Load(ctx context.Context, wait bool) (*[]string, error) { + _, startErr := s.Revalidate(ctx) + if startErr != nil { + return s.Current(), startErr } - s.observerMu.Lock() - s.observers = append(s.observers, observer) - s.observerMu.Unlock() + if s.Current() == nil || wait { + if err := s.refresh.Wait(ctx); err != nil { + return s.Current(), err + } + } + return s.Current(), nil } -func (s *Source) emit(event SourceEvent) { - if s.config.OnRefresh != nil { - s.config.OnRefresh(event) - } - s.observerMu.RLock() - observers := append([]func(SourceEvent){}, s.observers...) - s.observerMu.RUnlock() - for _, observer := range observers { - observer(event) +func (s *IPList) Set(entries *[]string) error { + if entries == nil { + return s.Clear() } + s.generation.Add(1) + s.store(*entries) + return nil } -func (s *Source) Close() error { - if s != nil && s.cancel != nil { +func (s *IPList) Clear() error { + s.generation.Add(1) + s.current.Store(nil) + s.loadedAt.Store(nil) + return nil +} + +func (s *IPList) Status() cachable.Status { + loadedAt := time.Time{} + if t := s.loadedAt.Load(); t != nil { + loadedAt = *t + } + var lastErr error + if p := s.lastErr.Load(); p != nil { + lastErr = *p + } + status := cachable.Status{LoadedAt: loadedAt, HasValue: s.Current() != nil, Refreshing: s.refresh.Running(), LastError: lastErr} + if !loadedAt.IsZero() { + status.StaleAt = loadedAt.Add(s.config.RefreshInterval) + } + return status +} + +func (s *IPList) Start(ctx context.Context, interval time.Duration) { + s.start.Do(func() { + if interval <= 0 { + interval = s.config.RefreshInterval + } + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-s.ctx.Done(): + return + case <-ticker.C: + _, _ = s.Revalidate(s.ctx) + } + } + }() + }) +} + +func (s *IPList) Stop() error { + if s == nil { + return nil + } + if s.cancel != nil { s.cancel() } - return nil + timeout := s.config.RefreshTimeout + if timeout <= 0 { + timeout = DefaultRefreshTimeout + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + return s.refresh.Stop(ctx) } -func (s *Source) load() ([]string, error) { +func (s *IPList) load(ctx context.Context) ([]string, error) { if s.config.Source == "" { if s.config.Optional { return nil, nil @@ -121,23 +209,24 @@ func (s *Source) load() ([]string, error) { } } } - return Load(s.ctx, s.config.Source, s.config.CacheDir, s.config.HTTPClient) + return Load(ctx, s.config.Source, s.config.CacheDir, s.config.HTTPClient) } -func (s *Source) store(entries []string) { - copyEntries := append([]string(nil), entries...) - s.current.Store(&sourceSnapshot{entries: copyEntries}) +func (s *IPList) store(entries []string) { + cp := append([]string(nil), entries...) + s.current.Store(&cp) + now := time.Now() + s.loadedAt.Store(&now) } -func (s *Source) refreshLoop() { - ticker := time.NewTicker(s.config.RefreshInterval) - defer ticker.Stop() - for { - select { - case <-s.ctx.Done(): - return - case <-ticker.C: - _ = s.Refresh() +func sameEntries(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false } } + return true } diff --git a/net/iplist/source_concurrency_test.go b/net/iplist/source_concurrency_test.go new file mode 100644 index 0000000..4441d93 --- /dev/null +++ b/net/iplist/source_concurrency_test.go @@ -0,0 +1,248 @@ +package iplist + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" +) + +// clearIPCache removes the on-disk httpcache entries so the next Load +// re-fetches from the source instead of serving the cached body. +func clearIPCache(t *testing.T, cacheDir string) { + t.Helper() + if err := os.RemoveAll(filepath.Join(cacheDir, "ip-sources")); err != nil { + t.Fatalf("clear cache: %v", err) + } +} + +// newBlockingServer returns an httptest server whose first response serves +// the given body immediately (for the initial Load), and whose second +// response blocks until release is closed. The request count is recorded. +func newBlockingServer(body string) (srv *httptest.Server, release chan struct{}, hits *atomic.Int32) { + release = make(chan struct{}) + hits = new(atomic.Int32) + srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + // Block only on the second and later requests. + if hits.Load() > 1 { + select { + case <-release: + case <-r.Context().Done(): + return + } + } + _, _ = w.Write([]byte(body)) + })) + return srv, release, hits +} + +// TestIPList_RevalidateCoalesces confirms concurrent Revalidate calls run the +// underlying fetch at most once while one is in flight. +func TestIPList_RevalidateCoalesces(t *testing.T) { + srv, release, hits := newBlockingServer("192.0.2.1\n") + defer srv.Close() + + // Initial load: configure a short interval so the first Load proceeds. + src, err := NewIPList(t.Context(), IPListConfig{ + Source: srv.URL, + CacheDir: t.TempDir(), + RefreshInterval: time.Hour, // avoid re-due during the test + }) + if err != nil { + t.Fatal(err) + } + defer src.Stop() + + // Stale the source so Revalidate is due, then make the next fetch block. + clearIPCache(t, src.config.CacheDir) + src.config.RefreshInterval = 0 // always due + + const n = 12 + var startedCount atomic.Int32 + var wg sync.WaitGroup + start := make(chan struct{}) + wg.Add(n) + for range n { + go func() { + defer wg.Done() + <-start + started, err := src.Revalidate(t.Context()) + if err != nil { + t.Errorf("Revalidate: %v", err) + return + } + if started { + startedCount.Add(1) + } + }() + } + close(start) + // Allow the coalesced refresh to enter the blocking server exactly once. + time.Sleep(30 * time.Millisecond) + if got := hits.Load(); got != 2 { + t.Fatalf("server hit %d times during in-flight refresh, want 2 (1 initial + 1 coalesced)", got) + } + close(release) // unblock the in-flight fetch + wg.Wait() + + // Subsequent requests after release serve immediately; drain. + if _, err := src.Load(t.Context(), true); err != nil { + t.Logf("drain: %v", err) + } + if startedCount.Load() != 1 { + t.Errorf("exactly one caller should have started, got %d", startedCount.Load()) + } +} + +// TestIPList_SetDuringRefreshNotClobbered confirms the generation guard: a +// manual Set while a refresh is in flight is not overwritten by the stale +// refresh result. +func TestIPList_SetDuringRefreshNotClobbered(t *testing.T) { + srv, release, hits := newBlockingServer("192.0.2.1\n") + defer srv.Close() + + src, err := NewIPList(t.Context(), IPListConfig{ + Source: srv.URL, + CacheDir: t.TempDir(), + RefreshInterval: time.Hour, + }) + if err != nil { + t.Fatal(err) + } + defer src.Stop() + + // Force due and start a blocking refresh. + clearIPCache(t, src.config.CacheDir) + src.config.RefreshInterval = 0 + if started, _ := src.Revalidate(t.Context()); !started { + t.Fatal("Revalidate should start") + } + time.Sleep(20 * time.Millisecond) + if hits.Load() != 2 { + t.Fatalf("expected 1 in-flight fetch (2 total), got %d", hits.Load()) + } + + // Manually set entries while the refresh is blocked. + manual := []string{"10.0.0.1"} + if err := src.Set(&manual); err != nil { + t.Fatalf("Set: %v", err) + } + if got := src.Current(); got == nil || len(*got) != 1 || (*got)[0] != "10.0.0.1" { + t.Fatalf("after Set Current = %v, want 10.0.0.1", got) + } + + // Release the stale refresh; it must not overwrite the manual value. + close(release) + _, _ = src.Load(t.Context(), true) + if got := src.Current(); got == nil || len(*got) != 1 || (*got)[0] != "10.0.0.1" { + t.Errorf("after stale refresh Current = %v, want 10.0.0.1 (generation guard)", got) + } +} + +// TestIPList_ClearDuringRefreshPreventsPublication confirms Clear bumps the +// generation so an in-flight refresh does not republish. +func TestIPList_ClearDuringRefreshPreventsPublication(t *testing.T) { + srv, release, hits := newBlockingServer("192.0.2.1\n") + defer srv.Close() + + src, err := NewIPList(t.Context(), IPListConfig{ + Source: srv.URL, + CacheDir: t.TempDir(), + RefreshInterval: time.Hour, + }) + if err != nil { + t.Fatal(err) + } + defer src.Stop() + + clearIPCache(t, src.config.CacheDir) + src.config.RefreshInterval = 0 + if started, _ := src.Revalidate(t.Context()); !started { + t.Fatal("Revalidate should start") + } + time.Sleep(20 * time.Millisecond) + if hits.Load() != 2 { + t.Fatalf("expected 1 in-flight fetch (2 total), got %d", hits.Load()) + } + + if err := src.Clear(); err != nil { + t.Fatalf("Clear: %v", err) + } + if src.Current() != nil { + t.Fatal("Current should be nil after Clear") + } + + close(release) + _, _ = src.Load(t.Context(), true) + if src.Current() != nil { + t.Errorf("Current = %v after Clear+stale refresh, want nil (generation guard)", src.Current()) + } +} + +// TestIPList_DetachedContextSurvivesCallerCancel confirms cancelling the +// caller context after Revalidate returns does not kill the background fetch. +func TestIPList_DetachedContextSurvivesCallerCancel(t *testing.T) { + srv, release, hits := newBlockingServer("192.0.2.1\n") + defer srv.Close() + + src, err := NewIPList(t.Context(), IPListConfig{ + Source: srv.URL, + CacheDir: t.TempDir(), + RefreshInterval: time.Hour, + }) + if err != nil { + t.Fatal(err) + } + defer src.Stop() + + clearIPCache(t, src.config.CacheDir) + src.config.RefreshInterval = 0 + ctx, cancel := context.WithCancel(t.Context()) + if started, err := src.Revalidate(ctx); err != nil || !started { + t.Fatalf("Revalidate started=%v err=%v", started, err) + } + cancel() + time.Sleep(30 * time.Millisecond) + if got := hits.Load(); got != 2 { + t.Fatalf("server hit %d times after caller cancel, want 2 (detached)", got) + } + close(release) + if _, err := src.Load(t.Context(), true); err != nil { + t.Fatalf("Load after cancel: %v", err) + } +} + +// TestIPList_KeepsLastGoodOnRefreshFailure is the file-source version of the +// existing http failure test, confirming last-good retention for a local +// source that disappears mid-run. +func TestIPList_KeepsLastGoodOnRefreshFailure(t *testing.T) { + path := filepath.Join(t.TempDir(), "list.tsv") + if err := os.WriteFile(path, []byte("192.0.2.1\n"), 0o600); err != nil { + t.Fatal(err) + } + src, err := NewIPList(t.Context(), IPListConfig{Source: path, CacheDir: t.TempDir()}) + if err != nil { + t.Fatal(err) + } + defer src.Stop() + + if got := src.Current(); got == nil || len(*got) != 1 { + t.Fatalf("initial = %#v", got) + } + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + src.config.RefreshInterval = 0 + if _, err := src.Load(t.Context(), true); err == nil { + t.Fatal("expected refresh error") + } + if got := src.Current(); got == nil || len(*got) != 1 || (*got)[0] != "192.0.2.1" { + t.Errorf("last-good lost after failure: %#v", got) + } +} diff --git a/net/iplist/source_test.go b/net/iplist/source_test.go index 5e41899..8ad465b 100644 --- a/net/iplist/source_test.go +++ b/net/iplist/source_test.go @@ -6,41 +6,70 @@ import ( "testing" ) +func TestCurrentZeroAlloc(t *testing.T) { + path := filepath.Join(t.TempDir(), "list.tsv") + var body []byte + for i := range 1000 { + body = append(body, []byte("10.0.0.")...) + body = append(body, '0'+byte(i%10)) + body = append(body, '\n') + } + if err := os.WriteFile(path, body, 0o600); err != nil { + t.Fatal(err) + } + s, err := NewIPList(t.Context(), IPListConfig{Source: path, CacheDir: t.TempDir()}) + if err != nil { + t.Fatal(err) + } + defer s.Stop() + + _ = s.Current() // warm up + + allocs := testing.AllocsPerRun(100, func() { + _ = s.Current() + }) + if allocs != 0 { + t.Errorf("Current() allocs per call = %.1f, want 0 (hot path must be zero-alloc)", allocs) + } +} + func TestSourceKeepsLastGoodEntriesAfterFailedRefresh(t *testing.T) { path := filepath.Join(t.TempDir(), "list.tsv") if err := os.WriteFile(path, []byte("192.0.2.1\n"), 0o600); err != nil { t.Fatal(err) } - source, err := NewSource(t.Context(), SourceConfig{Source: path}) + source, err := NewIPList(t.Context(), IPListConfig{Source: path, CacheDir: t.TempDir()}) if err != nil { t.Fatal(err) } - defer func() { _ = source.Close() }() + defer source.Stop() - if got := source.Entries(); len(got) != 1 || got[0] != "192.0.2.1" { + if got := source.Current(); got == nil || len(*got) != 1 || (*got)[0] != "192.0.2.1" { t.Fatalf("initial entries = %#v", got) } if err := os.Remove(path); err != nil { t.Fatal(err) } - if err := source.Refresh(); err == nil { + source.config.RefreshInterval = 0 + if _, err := source.Load(t.Context(), true); err == nil { t.Fatal("invalid refresh unexpectedly succeeded") } - if got := source.Entries(); len(got) != 1 || got[0] != "192.0.2.1" { + if got := source.Current(); got == nil || len(*got) != 1 || (*got)[0] != "192.0.2.1" { t.Fatalf("entries after failed refresh = %#v", got) } } func TestSourceOptionalMissingFileIsEmpty(t *testing.T) { - source, err := NewSource(t.Context(), SourceConfig{ + source, err := NewIPList(t.Context(), IPListConfig{ Source: filepath.Join(t.TempDir(), "missing.tsv"), + CacheDir: t.TempDir(), Optional: true, }) if err != nil { t.Fatal(err) } - defer func() { _ = source.Close() }() - if got := source.Entries(); len(got) != 0 { + defer source.Stop() + if got := source.Current(); got == nil || len(*got) != 0 { t.Fatalf("entries = %#v", got) } }