refactor(net/iplist): use released cache APIs

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

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.6.0 h1:qFBnSs6rZrtUQ+fweP4bHF2FQd/Syvs7mRN5nO5YtkY=
-github.com/therootcompany/golib/net/httpcache v0.5.1/go.mod h1:mKSQJWpGSCDBzWkTCwT7dO1SCVCsFmdkRgKr39fNS1U=
+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=

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)

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 {
+type IPList struct {
-	Entries uint64
-	Err     error
-}
-
-type sourceSnapshot struct {
-	entries []string
-}
-
-type Source struct {
 	ctx        context.Context
 	cancel     context.CancelFunc
-	config     SourceConfig
+	config     IPListConfig
-	current    atomic.Pointer[sourceSnapshot]
+	current    atomic.Pointer[[]string]
-	observers  []func(SourceEvent)
+	start      sync.Once
-	observerMu sync.RWMutex
+	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}
+	s := &IPList{ctx: sourceCtx, cancel: cancel, config: config}
-	entries, err := s.load()
+	if _, err := s.Load(ctx, true); err != nil {
-	if err != nil {
 		cancel()
 		return nil, err
 	}
-	s.store(entries)
-	go s.refreshLoop()
 	return s, nil
 }
 
-func (s *Source) Entries() []string {
+func (s *IPList) Current() *[]string {
-	current := s.current.Load()
+	return s.current.Load()
-	if current == nil {
-		return nil
-	}
-	return append([]string(nil), current.entries...)
 }
 
-func (s *Source) Refresh() error {
+func (s *IPList) Due(ctx context.Context) (bool, error) {
-	entries, err := s.load()
+	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
 	}
+	old := s.current.Load()
+	changed := old == nil || !sameEntries(*old, entries)
+	if changed && s.generation.Load() == generation {
 		s.store(entries)
-	s.emit(SourceEvent{Entries: uint64(len(entries))})
+	}
+	s.lastErr.Store(nil)
 	return nil
 }
 
-func (s *Source) Subscribe(observer func(SourceEvent)) {
+func (s *IPList) Load(ctx context.Context, wait bool) (*[]string, error) {
-	if observer == nil {
+	_, startErr := s.Revalidate(ctx)
+	if startErr != nil {
+		return s.Current(), startErr
+	}
+	if s.Current() == nil || wait {
+		if err := s.refresh.Wait(ctx); err != nil {
+			return s.Current(), err
+		}
+	}
+	return s.Current(), nil
+}
+
+func (s *IPList) Set(entries *[]string) error {
+	if entries == nil {
+		return s.Clear()
+	}
+	s.generation.Add(1)
+	s.store(*entries)
+	return 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)
 				}
-	s.observerMu.Lock()
+			}
-	s.observers = append(s.observers, observer)
+		}()
-	s.observerMu.Unlock()
+	})
 }
 
-func (s *Source) emit(event SourceEvent) {
+func (s *IPList) Stop() error {
-	if s.config.OnRefresh != nil {
+	if s == nil {
-		s.config.OnRefresh(event)
+		return nil
 	}
-	s.observerMu.RLock()
+	if s.cancel != nil {
-	observers := append([]func(SourceEvent){}, s.observers...)
-	s.observerMu.RUnlock()
-	for _, observer := range observers {
-		observer(event)
-	}
-}
-
-func (s *Source) Close() error {
-	if s != nil && 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) {
+func (s *IPList) store(entries []string) {
-	copyEntries := append([]string(nil), entries...)
+	cp := append([]string(nil), entries...)
-	s.current.Store(&sourceSnapshot{entries: copyEntries})
+	s.current.Store(&cp)
+	now := time.Now()
+	s.loadedAt.Store(&now)
 }
 
-func (s *Source) refreshLoop() {
+func sameEntries(a, b []string) bool {
-	ticker := time.NewTicker(s.config.RefreshInterval)
+	if len(a) != len(b) {
-	defer ticker.Stop()
+		return false
-	for {
+	}
-		select {
+	for i := range a {
-		case <-s.ctx.Done():
+		if a[i] != b[i] {
-			return
+			return false
-		case <-ticker.C:
-			_ = s.Refresh()
 		}
 	}
+	return true
 }

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

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() }()
+	defer source.Stop()
-	if got := source.Entries(); len(got) != 0 {
+	if got := source.Current(); got == nil || len(*got) != 0 {
 		t.Fatalf("entries = %#v", got)
 	}
 }