Merge pull request #1203 from J3vb/fix/updater-singleflight

fix(updater): coalesce and negative-cache text-asset fetches (W3-1)

Server/go.mod

@@ -32,6 +32,7 @@ require (
 	go.yaml.in/yaml/v3 v3.0.4
 	golang.org/x/crypto v0.51.0
 	golang.org/x/mod v0.35.0
+	golang.org/x/sync v0.20.0
 	modernc.org/sqlite v1.48.0
 )
 
@@ -132,7 +133,6 @@ require (
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	golang.org/x/exp v0.0.0-20260212183809-81e46e3db34a // indirect
 	golang.org/x/net v0.55.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.45.0 // indirect
 	golang.org/x/text v0.37.0 // indirect
 	golang.org/x/time v0.14.0 // indirect

Server/updater/text_asset_cache_test.go

@@ -0,0 +1,163 @@
+package updater
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// textAssetServer serves a fixed body at /asset and counts inbound requests.
+// handler may be swapped to simulate an upstream outage.
+func textAssetServer(t *testing.T, fail *atomic.Bool, hits *atomic.Int64) *httptest.Server {
+	t.Helper()
+	mux := http.NewServeMux()
+	mux.HandleFunc("/asset", func(w http.ResponseWriter, _ *http.Request) {
+		hits.Add(1)
+		if fail.Load() {
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		fmt.Fprint(w, "asset-body")
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+// A burst of concurrent misses must collapse into a single outbound fetch.
+// Without singleflight each caller fetches independently (W3-1).
+func TestFetchTextAssetCachedCoalescesConcurrentMisses(t *testing.T) {
+	var fail atomic.Bool
+	var hits atomic.Int64
+	srv := textAssetServer(t, &fail, &hits)
+	u := newTestUpdater(srv.URL, "1.0.0")
+
+	const callers = 25
+	var wg sync.WaitGroup
+	results := make([]string, callers)
+	errs := make([]error, callers)
+	start := make(chan struct{})
+
+	for i := range callers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start // release all goroutines together to force a real burst
+			results[i], errs[i] = u.FetchTextAssetCached(context.Background(), srv.URL+"/asset")
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	if got := hits.Load(); got != 1 {
+		t.Fatalf("outbound fetches = %d, want exactly 1 (singleflight should coalesce)", got)
+	}
+	for i := range callers {
+		if errs[i] != nil {
+			t.Fatalf("caller %d: unexpected error: %v", i, errs[i])
+		}
+		if results[i] != "asset-body" {
+			t.Fatalf("caller %d: content = %q, want %q", i, results[i], "asset-body")
+		}
+	}
+}
+
+// A failed fetch must be cached for errorCacheTTL so an upstream outage does
+// not produce one outbound request per caller.
+func TestFetchTextAssetCachedCachesFailures(t *testing.T) {
+	var fail atomic.Bool
+	var hits atomic.Int64
+	srv := textAssetServer(t, &fail, &hits)
+	fail.Store(true)
+	u := newTestUpdater(srv.URL, "1.0.0")
+	url := srv.URL + "/asset"
+
+	if _, err := u.FetchTextAssetCached(context.Background(), url); err == nil {
+		t.Fatal("first call: want error from failing upstream, got nil")
+	}
+	if _, err := u.FetchTextAssetCached(context.Background(), url); err == nil {
+		t.Fatal("second call: want cached error, got nil")
+	}
+	if got := hits.Load(); got != 1 {
+		t.Fatalf("outbound fetches = %d, want 1 (the failure should be cached)", got)
+	}
+
+	// Once the negative entry expires, the upstream is retried — a cached error
+	// must not be permanent.
+	u.mu.Lock()
+	entry := u.textAssetCache[url]
+	entry.expiry = time.Now().Add(-time.Second)
+	u.textAssetCache[url] = entry
+	u.mu.Unlock()
+
+	fail.Store(false)
+	content, err := u.FetchTextAssetCached(context.Background(), url)
+	if err != nil {
+		t.Fatalf("after negative-cache expiry: unexpected error: %v", err)
+	}
+	if content != "asset-body" {
+		t.Fatalf("content = %q, want %q", content, "asset-body")
+	}
+	if got := hits.Load(); got != 2 {
+		t.Fatalf("outbound fetches = %d, want 2 (retry after expiry)", got)
+	}
+}
+
+// Asset URLs carry a version, so stale keys must be evicted or the map grows by
+// one entry per release for the process lifetime.
+func TestFetchTextAssetCachedEvictsExpiredKeys(t *testing.T) {
+	var fail atomic.Bool
+	var hits atomic.Int64
+	srv := textAssetServer(t, &fail, &hits)
+	u := newTestUpdater(srv.URL, "1.0.0")
+
+	// A superseded entry from an earlier release, already past its expiry.
+	u.mu.Lock()
+	u.textAssetCache = map[string]textAssetCacheEntry{
+		"https://example.invalid/v0.9.0/chatserver.exe.sig": {
+			content: "stale",
+			expiry:  time.Now().Add(-time.Hour),
+		},
+	}
+	u.mu.Unlock()
+
+	if _, err := u.FetchTextAssetCached(context.Background(), srv.URL+"/asset"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	u.mu.Lock()
+	defer u.mu.Unlock()
+	if _, ok := u.textAssetCache["https://example.invalid/v0.9.0/chatserver.exe.sig"]; ok {
+		t.Fatal("expired key survived a cache write; it should have been evicted")
+	}
+	if len(u.textAssetCache) != 1 {
+		t.Fatalf("cache size = %d, want 1 (only the fresh entry)", len(u.textAssetCache))
+	}
+}
+
+// A live cached entry must be served without any outbound request.
+func TestFetchTextAssetCachedServesFromCache(t *testing.T) {
+	var fail atomic.Bool
+	var hits atomic.Int64
+	srv := textAssetServer(t, &fail, &hits)
+	u := newTestUpdater(srv.URL, "1.0.0")
+	url := srv.URL + "/asset"
+
+	for range 3 {
+		content, err := u.FetchTextAssetCached(context.Background(), url)
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if content != "asset-body" {
+			t.Fatalf("content = %q, want %q", content, "asset-body")
+		}
+	}
+	if got := hits.Load(); got != 1 {
+		t.Fatalf("outbound fetches = %d, want 1 (subsequent calls should hit cache)", got)
+	}
+}

Server/updater/updater.go

@@ -29,6 +29,7 @@ import (
 	"github.com/owncord/server/syncutil"
 
 	"golang.org/x/mod/semver"
+	"golang.org/x/sync/singleflight"
 )
 
 const (
@@ -118,6 +119,7 @@ type Updater struct {
 	cachedErr      error
 	errCacheExpiry time.Time
 	textAssetCache map[string]textAssetCacheEntry
+	textAssetSF    singleflight.Group
 	mu             syncutil.Mutex
 	httpClient     *http.Client
 	signingKeyText string
@@ -128,7 +130,11 @@ type Updater struct {
 // memory instead of re-fetching from GitHub on every call.
 type textAssetCacheEntry struct {
 	content string
-	expiry  time.Time
+	// err caches a failed fetch so an upstream outage is not re-dialled on
+	// every request. Cached errors expire after errorCacheTTL, successes
+	// after cacheTTL.
+	err    error
+	expiry time.Time
 }
 
 // NewUpdater creates an Updater for the given repository.
@@ -751,29 +757,65 @@ func (u *Updater) FetchTextAsset(ctx context.Context, url string) (string, error
 // unrate-limited callers (e.g. the client-update endpoint) be served from
 // memory instead of triggering an outbound fetch on every request.
 func (u *Updater) FetchTextAssetCached(ctx context.Context, url string) (string, error) {
-	now := time.Now()
-
-	u.mu.Lock()
-	if entry, ok := u.textAssetCache[url]; ok && now.Before(entry.expiry) {
-		content := entry.content
-		u.mu.Unlock()
-		return content, nil
+	if entry, ok := u.lookupTextAsset(url, time.Now()); ok {
+		return entry.content, entry.err
 	}
-	u.mu.Unlock()
 
-	content, err := u.FetchTextAsset(ctx, url)
+	// Coalesce concurrent misses: when the TTL expires under load, every caller
+	// would otherwise issue its own outbound fetch. One flight per URL runs and
+	// the rest wait on its result.
+	//
+	// ponytail: the leader's ctx drives the fetch, so a cancelled leader fails
+	// its followers too. Acceptable here — callers are the unauthenticated
+	// client-update endpoint, and the outcome is cached either way.
+	v, err, _ := u.textAssetSF.Do(url, func() (any, error) {
+		now := time.Now()
+		// Re-check: another flight may have filled the cache while we queued.
+		if entry, ok := u.lookupTextAsset(url, now); ok {
+			return entry.content, entry.err
+		}
+		content, fetchErr := u.FetchTextAsset(ctx, url)
+		u.storeTextAsset(url, content, fetchErr, now)
+		return content, fetchErr
+	})
 	if err != nil {
 		return "", err
 	}
+	return v.(string), nil
+}
 
+// lookupTextAsset returns a live cache entry for url, if one exists. A cached
+// entry may hold either content or an error; both are honoured until expiry.
+func (u *Updater) lookupTextAsset(url string, now time.Time) (textAssetCacheEntry, bool) {
 	u.mu.Lock()
+	defer u.mu.Unlock()
+	entry, ok := u.textAssetCache[url]
+	if !ok || !now.Before(entry.expiry) {
+		return textAssetCacheEntry{}, false
+	}
+	return entry, true
+}
+
+// storeTextAsset records the outcome of a fetch, caching failures briefly so an
+// upstream outage does not trigger an outbound request per caller.
+func (u *Updater) storeTextAsset(url, content string, err error, now time.Time) {
+	u.mu.Lock()
+	defer u.mu.Unlock()
 	if u.textAssetCache == nil {
 		u.textAssetCache = make(map[string]textAssetCacheEntry)
 	}
-	u.textAssetCache[url] = textAssetCacheEntry{content: content, expiry: now.Add(cacheTTL)}
-	u.mu.Unlock()
-
-	return content, nil
+	// Drop superseded keys: asset URLs carry a version, so without this the map
+	// grows by one entry per release for the lifetime of the process.
+	for k, e := range u.textAssetCache {
+		if !now.Before(e.expiry) {
+			delete(u.textAssetCache, k)
+		}
+	}
+	ttl := cacheTTL
+	if err != nil {
+		ttl = errorCacheTTL
+	}
+	u.textAssetCache[url] = textAssetCacheEntry{content: content, err: err, expiry: now.Add(ttl)}
 }
 
 // downloadFile downloads the content at url and writes it to destPath.