mirror of
https://github.com/git-pkgs/proxy.git
synced 2026-11-09 15:49:16 -05:00
c4615ff561
The feed readers move to registries' changes package. The proxy keeps the cursor and moves it past the changes only once their metadata is expired, so a failed expiry reads them again.
108 lines
3.8 KiB
Go
108 lines
3.8 KiB
Go
package server
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"net/http"
|
|
"time"
|
|
|
|
"github.com/git-pkgs/proxy/internal/database"
|
|
"github.com/git-pkgs/proxy/internal/metrics"
|
|
"github.com/git-pkgs/registries/changes"
|
|
"github.com/git-pkgs/registries/client"
|
|
)
|
|
|
|
// changeFeedInterval is how often an upstream change feed is read.
|
|
const changeFeedInterval = time.Minute
|
|
|
|
// changeFeedMaxBackoff caps the wait between reads of a failing feed.
|
|
const changeFeedMaxBackoff = 15 * time.Minute
|
|
|
|
// startChangeFeeds follows the change feeds the configuration turns on. Each
|
|
// instance sharing a database follows them itself: the reads are small, and
|
|
// expiring a row twice does no harm.
|
|
func (s *Server) startChangeFeeds(ctx context.Context, httpClient *http.Client, metadataTTL time.Duration) {
|
|
c := changes.WithClient(client.NewClient(client.WithHTTPClient(httpClient)))
|
|
var feeds []changes.Feed
|
|
if s.cfg.Upstream.NPMChangeFeed {
|
|
feeds = append(feeds, changes.NewNPM(c))
|
|
}
|
|
if s.cfg.Upstream.ComposerChangeFeed {
|
|
feeds = append(feeds, changes.NewPackagist(c, changes.WithURL(changes.PackagistFeedURL(s.cfg.Upstream.Composer))))
|
|
}
|
|
if len(feeds) == 0 {
|
|
return
|
|
}
|
|
if !s.cfg.CacheMetadata || metadataTTL <= 0 {
|
|
// Every request already revalidates, so there is nothing to expire.
|
|
s.logger.Info("change feeds not followed: metadata is not served from cache without revalidating")
|
|
return
|
|
}
|
|
for _, feed := range feeds {
|
|
s.background.Go(func() { followChangeFeed(ctx, s.db, s.logger, feed, changeFeedInterval) })
|
|
}
|
|
}
|
|
|
|
// followChangeFeed reads feed every interval until ctx ends and expires the
|
|
// cached metadata it lists. After a failed read it waits twice as long as
|
|
// before, up to changeFeedMaxBackoff. The position in the feed is kept in
|
|
// memory: after a restart the feed is followed from its current position.
|
|
func followChangeFeed(ctx context.Context, db *database.DB, logger *slog.Logger, feed changes.Feed, interval time.Duration) {
|
|
wait := interval
|
|
var cursor changes.Cursor
|
|
for {
|
|
next, err := readChangeFeed(ctx, db, logger, feed, cursor)
|
|
cursor = next
|
|
if err != nil {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
wait = min(wait+wait, changeFeedMaxBackoff)
|
|
logger.Warn("change feed: read failed", "ecosystem", feed.Ecosystem(), "error", err, "retry_in", wait)
|
|
} else {
|
|
wait = interval
|
|
}
|
|
timer := time.NewTimer(wait)
|
|
select {
|
|
case <-ctx.Done():
|
|
timer.Stop()
|
|
return
|
|
case <-timer.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
// readChangeFeed reads feed once after cursor, or finds its current position
|
|
// when cursor is empty, and expires the cached metadata of what it lists, so
|
|
// the next request for it revalidates upstream. Only rows fetched before the
|
|
// read finished are expired: a row fetched after it was fetched after every
|
|
// change the read returned. It returns the cursor to read on from, which
|
|
// moves past the changes only once they are expired, so a failed expiry
|
|
// reads them again.
|
|
func readChangeFeed(ctx context.Context, db *database.DB, logger *slog.Logger, feed changes.Feed, cursor changes.Cursor) (changes.Cursor, error) {
|
|
if cursor == "" {
|
|
return feed.Start(ctx)
|
|
}
|
|
found, next, err := feed.Changes(ctx, cursor)
|
|
if err != nil {
|
|
return cursor, err
|
|
}
|
|
cutoff := time.Now()
|
|
ecosystem := feed.Ecosystem()
|
|
var expired []string
|
|
switch {
|
|
case found.All:
|
|
logger.Info("change feed: upstream asked for a full resync, revalidating all cached metadata", "ecosystem", ecosystem)
|
|
expired, err = db.ExpireAllMetadataCache(ecosystem, cutoff)
|
|
case len(found.Names) > 0:
|
|
expired, err = db.ExpireMetadataCache(ecosystem, found.Names, cutoff)
|
|
}
|
|
for _, name := range expired {
|
|
logger.Debug("change feed: cached metadata changed upstream", "ecosystem", ecosystem, "key", name)
|
|
}
|
|
metrics.RecordMetadataExpiredByChangeFeed(ecosystem, len(expired))
|
|
if err != nil {
|
|
return cursor, err
|
|
}
|
|
return next, nil
|
|
}
|