diff --git a/Server/go.mod b/Server/go.mod index abc88c81..9c524e13 100644 --- a/Server/go.mod +++ b/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 diff --git a/Server/updater/text_asset_cache_test.go b/Server/updater/text_asset_cache_test.go new file mode 100644 index 00000000..94a2ba80 --- /dev/null +++ b/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) + } +} diff --git a/Server/updater/updater.go b/Server/updater/updater.go index 34f71898..2a6c74aa 100644 --- a/Server/updater/updater.go +++ b/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.