From b78ac597589b6bcd8c2ad4f6beb7f4d97e97a7f8 Mon Sep 17 00:00:00 2001 From: miyuko Date: Sun, 31 May 2026 16:34:16 +0100 Subject: [PATCH] Make existence cache refresh checks more robust. For If-Modified-Since checks on the last change marker, only ever send timestamps that we previously received from the server. This sidesteps issues like races and anything dealing with clock precision. Also make the code simpler to follow: handleFilterUpdates() now owns lastRefresh, and refresh() now owns lastChanged. Also retry with an exponential backoff if we fail to refresh the cache. All sends to accessCh are non-blocking and are not affected. V12-Ref: F-77198 --- src/backend.go | 2 +- src/backend_fs.go | 4 ++-- src/backend_s3.go | 27 +++++++++++++++++-------- src/existence.go | 51 +++++++++++++++++++++++++++++++++++------------ src/observe.go | 4 ++-- 5 files changed, 62 insertions(+), 26 deletions(-) diff --git a/src/backend.go b/src/backend.go index 8c1ca61..5b80714 100644 --- a/src/backend.go +++ b/src/backend.go @@ -136,7 +136,7 @@ type Backend interface { GetAllManifests(ctx context.Context) iter.Seq2[tuple[*ManifestMetadata, *Manifest], error] // Check whether the set of sites we serve has changed since the time passed to this method. - HasSiteListChanged(ctx context.Context, since time.Time) (changed bool, err error) + HasSiteListChanged(ctx context.Context, since time.Time) (changed bool, lastChanged time.Time, err error) // Check whether a domain has any deployments. CheckDomain(ctx context.Context, domain string) (found bool, err error) diff --git a/src/backend_fs.go b/src/backend_fs.go index 378cb4f..2619449 100644 --- a/src/backend_fs.go +++ b/src/backend_fs.go @@ -488,8 +488,8 @@ func (fs *FSBackend) UnfreezeDomain(ctx context.Context, domain string) error { } } -func (fs *FSBackend) HasSiteListChanged(ctx context.Context, since time.Time) (bool, error) { - return true, nil // not implemented +func (fs *FSBackend) HasSiteListChanged(ctx context.Context, since time.Time) (bool, time.Time, error) { + return true, time.Time{}, nil // not implemented } func auditDetachedName(id AuditID) string { diff --git a/src/backend_s3.go b/src/backend_s3.go index 3f3bae5..867f7e2 100644 --- a/src/backend_s3.go +++ b/src/backend_s3.go @@ -818,17 +818,28 @@ func (s3 *S3Backend) UnfreezeDomain(ctx context.Context, domain string) error { const lastSiteUpdateObjectName = "meta/last-site-update" -func (s3 *S3Backend) HasSiteListChanged(ctx context.Context, since time.Time) (bool, error) { - info, err := s3.client.StatObject(ctx, s3.bucket, lastSiteUpdateObjectName, - minio.GetObjectOptions{}) - if err != nil { - if errResp := minio.ToErrorResponse(err); errResp.Code == "NoSuchKey" { - return true, nil +func (s3 *S3Backend) HasSiteListChanged(ctx context.Context, since time.Time) (bool, time.Time, error) { + opts := minio.GetObjectOptions{} + if !since.IsZero() { + if err := opts.SetModified(since); err != nil { + return false, time.Time{}, err } - return false, err } - return info.LastModified.After(since), nil + info, err := s3.client.StatObject(ctx, s3.bucket, lastSiteUpdateObjectName, opts) + if err != nil { + if errResp := minio.ToErrorResponse(err); errResp.Code == "NoSuchKey" { + // The marker has not been written yet, meaning no mutations have occurred. + return false, time.Time{}, nil + } else if errResp.StatusCode == http.StatusNotModified { + // The marker has not changed. + return false, time.Time{}, nil + } + + return false, time.Time{}, err + } + + return true, info.LastModified, nil } func (s3 *S3Backend) bumpLastSiteUpdateTimestamp(ctx context.Context) error { diff --git a/src/existence.go b/src/existence.go index 14c1951..e38033a 100644 --- a/src/existence.go +++ b/src/existence.go @@ -14,6 +14,7 @@ import ( "time" "github.com/bits-and-blooms/bloom/v3" + exponential "github.com/jpillora/backoff" ) type Existence int @@ -72,7 +73,7 @@ type bloomExistenceCache struct { accessCh chan struct{} refreshMu sync.Mutex - lastRefresh time.Time + lastChanged time.Time maxAge time.Duration } @@ -91,33 +92,58 @@ func createBloomExistenceCache(ctx context.Context, maxAge time.Duration) (Exist } func (c *bloomExistenceCache) handleFilterUpdates(ctx context.Context) { + lastRefresh := time.Now() + for range c.accessCh { - if time.Since(c.lastRefresh) > c.maxAge { - logc.Println(ctx, "existence: refreshing") - if err := c.refresh(ctx); err != nil { - logc.Printf(ctx, "existence: refresh error: %v", err) + if time.Since(lastRefresh) > c.maxAge { + backoff := exponential.Backoff{ + Jitter: true, + Min: time.Second * 1, + Max: time.Second * 60, + } + for { + logc.Println(ctx, "existence: refreshing") + err := c.refreshIfChanged(ctx) + if err == nil { + lastRefresh = time.Now() + break + } + sleepFor := backoff.Duration() + logc.Printf(ctx, "existence: refresh error: %v (retry in %s)", err, sleepFor) + time.Sleep(sleepFor) } } } } -func (c *bloomExistenceCache) refresh(ctx context.Context) error { - c.refreshMu.Lock() - defer c.refreshMu.Unlock() - - if changed, err := backend.HasSiteListChanged(ctx, c.lastRefresh); err != nil { +func (c *bloomExistenceCache) refreshIfChanged(ctx context.Context) error { + changed, lastChanged, err := backend.HasSiteListChanged(ctx, c.lastChanged) + if err != nil { return err } else if !changed { logc.Println(ctx, "existence: unchanged") - c.lastRefresh = time.Now() return nil } + err = c.refresh(ctx) + if err != nil { + return err + } + + c.lastChanged = lastChanged + return nil +} + +func (c *bloomExistenceCache) refresh(ctx context.Context) error { // Create two 256 KiB Bloom filters that will fit ~150K entries each with 0.1% false positive rate. sites := bloom.New(256*1024*8, 10) domains := bloom.New(256*1024*8, 10) - logc.Printf(ctx, "existence: refreshing") + // To prevent sites created during enumeration from getting dropped, acquire an exclusive lock + // so that filter additions wait until we finish enumerating sites and update the filters. + c.refreshMu.Lock() + defer c.refreshMu.Unlock() + siteCount := 0 for metadata, err := range backend.EnumerateManifests(ctx) { if err != nil { @@ -136,7 +162,6 @@ func (c *bloomExistenceCache) refresh(ctx context.Context) error { c.filterMu.Unlock() logc.Printf(ctx, "existence: refreshed with %d sites", siteCount) - c.lastRefresh = time.Now() return nil } diff --git a/src/observe.go b/src/observe.go index 2fbfba4..97ff495 100644 --- a/src/observe.go +++ b/src/observe.go @@ -358,9 +358,9 @@ func (backend *observedBackend) UnfreezeDomain(ctx context.Context, domain strin return } -func (backend *observedBackend) HasSiteListChanged(ctx context.Context, since time.Time) (changed bool, err error) { +func (backend *observedBackend) HasSiteListChanged(ctx context.Context, since time.Time) (changed bool, lastChanged time.Time, err error) { span, ctx := ObserveFunction(ctx, "HasSiteListChanged", "since", since) - changed, err = backend.inner.HasSiteListChanged(ctx, since) + changed, lastChanged, err = backend.inner.HasSiteListChanged(ctx, since) span.Finish() return }