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 }