feat: background syncer for github_rpm remotes (#108)

## Why

Builds on #107 (merged), which derives `github_rpm` RPM metadata lazily on the
client request path, single-flighted per replica. Two problems remain: the
derive still happens per replica, so across a multi-replica deployment the same
releases are scanned and re-derived N times, multiplying GitHub queries; and a
cold cache blocks the first request on a full derive. GitHub's rate limits are
low (~60/hr unauthenticated, ~5000/hr authenticated), so this needs a single
coordinated syncer with a shared rate limit and conditional requests.

## How

- Add a single per-process background syncer (started at boot, cleanly stopped
  on shutdown) that owns a deduped/coalescing work queue, a worker pool, and one
  global token-bucket rate limiter (`golang.org/x/time/rate`) bound onto the
  github provider so every GitHub call (releases list + each ranged asset GET)
  acquires a token first.
- Re-check each `github_rpm` remote for new/changed releases on its existing
  `mutable_ttl` cadence; derive only new/changed assets incrementally and prune
  assets that disappear upstream. Repodata is served from primed DB rows.
- Prime metadata in the background on remote creation; the create call returns
  immediately.
- Send the stored releases-list `ETag` as `If-None-Match`; a `304` derives
  nothing and is not counted against GitHub's rate limit, so an unchanged repo
  is nearly free.
- Coordinate replicas through a `github_rpm_sync_state` row (`last_synced_at`,
  `etag`, `sync_lease_owner`, `sync_lease_expires`): a periodic scan runs only
  for the replica that atomically claims the lease, bounding total GitHub load
  to ~once per `mutable_ttl` regardless of replica count; the ETag is shared
  through the same row.
- Keep the request path fast: serve current cache, enqueue a prime on an empty
  cache, and return a bounded wait then a retryable `503` rather than blocking
  on a cold derive.
- Add `GITHUB_SYNC_RATE` / `GITHUB_SYNC_BURST` / `GITHUB_SYNC_WORKERS` /
  `GITHUB_SYNC_POLL_INTERVAL` config with conservative defaults (1 req/s, burst
  5, 3 workers, 60s tick) and document the syncer in the README.

## Tests

- Unit (httptest, Range/ETag-aware fixture): `304` releases response derives
  nothing; incremental derive fetches only the newly added asset; the shared
  limiter caps request rate; work-queue enqueues coalesce to one job; prime
  enqueues a job; a held lease stops a second replica from scanning; cold-start
  serves `503` while warm cache serves `200`.
- DB integration (testcontainers postgres): the real lease SQL — one holder at a
  time, recency gate blocks a too-soon periodic re-claim, prime (freshness 0)
  bypasses recency but respects a live lease.
- Docker e2e re-run: `dnf install dotvault` works; prime-on-create derives in the
  background at ~1 req/s (global limiter); `dnf makecache` served fast from the
  priming cache (no cold block); clean shutdown mid-scan, no panics.

## Notes

- Reuses `mutable_ttl` as the check interval (no new per-remote field), per brief.

Reviewed-on: #108
Co-authored-by: Ben Vincent <ben@unkin.net>
Co-committed-by: Ben Vincent <ben@unkin.net>

internal/provider/rpm/github.go

@@ -18,13 +18,19 @@ import (
 	"time"
 
 	rpmlib "github.com/cavaliergopher/rpm"
+	"golang.org/x/time/rate"
 
 	"git.unkin.net/unkin/artifactapi/internal/provider"
 	"git.unkin.net/unkin/artifactapi/pkg/models"
 )
 
+// gitHubProvider is the process-wide singleton. The background Syncer binds its
+// shared rate limiter and work queue onto this instance so the request path and
+// the syncer drive the same derive machinery.
+var gitHubProvider = newGitHubProvider()
+
 func init() {
-	provider.Register(newGitHubProvider())
+	provider.Register(gitHubProvider)
 }
 
 // Tuning knobs for the no-precache header fetch. Fields (not consts) so tests
@@ -39,6 +45,12 @@ const (
 	defaultScanTimeout = 10 * time.Minute
 	// defaultServeTimeout bounds a repodata DB read served on a detached context.
 	defaultServeTimeout = 30 * time.Second
+
+	// defaultColdWait bounds how long a repodata request blocks waiting for a
+	// just-enqueued prime to populate an empty cache before returning a
+	// retryable 503. Kept short so a client never hangs on a rate-limited derive
+	// of a large repo; small repos usually prime within this window.
+	defaultColdWait = 8 * time.Second
 )
 
 // GitHubProvider is a metadata-only remote: it scans a GitHub repo's releases
@@ -54,6 +66,15 @@ type GitHubProvider struct {
 	pageCap       int
 	scanTimeout   time.Duration
 	serveTimeout  time.Duration
+	coldWait      time.Duration
+
+	// limiter, when set by the Syncer, gates every GitHub HTTP call (releases
+	// list + each ranged asset fetch) through a single process-wide token bucket.
+	// nil means unlimited (direct provider use / unit tests).
+	limiter *rate.Limiter
+	// syncer, when set, routes freshness refresh and cold-start priming through
+	// the shared background work queue instead of an inline per-replica scan.
+	syncer *Syncer
 
 	mu       sync.Mutex
 	scanning map[string]bool
@@ -68,11 +89,21 @@ func newGitHubProvider() *GitHubProvider {
 		pageCap:       defaultReleasePageCap,
 		scanTimeout:   defaultScanTimeout,
 		serveTimeout:  defaultServeTimeout,
+		coldWait:      defaultColdWait,
 		scanning:      map[string]bool{},
 		lastScan:      map[string]time.Time{},
 	}
 }
 
+// limiterWait blocks until the shared rate limiter grants a token, or returns
+// the context error if it is canceled first. A nil limiter is a no-op.
+func (p *GitHubProvider) limiterWait(ctx context.Context) error {
+	if p.limiter == nil {
+		return nil
+	}
+	return p.limiter.Wait(ctx)
+}
+
 func (p *GitHubProvider) Type() models.PackageType { return models.PackageGitHubRPM }
 
 // Classify/ContentType/UpstreamURL/RewriteResponse/AuthHeaders satisfy the
@@ -114,7 +145,7 @@ func (p *GitHubProvider) AuthHeaders(_ context.Context, remote models.Remote) (h
 // 302-redirects .rpm downloads to the backend releases_remote. Returns false
 // only for paths it does not own, letting the normal proxy path take over.
 func (p *GitHubProvider) ServeRemote(w http.ResponseWriter, r *http.Request, remote models.Remote, path, proxyBaseURL string, store provider.RemoteMetadataStore) bool {
-	p.refresh(remote, store)
+	p.onRequest(remote, store)
 
 	if strings.HasPrefix(path, "repodata/") {
 		// Serve repodata on a context detached from the inbound request: a
@@ -124,6 +155,15 @@ func (p *GitHubProvider) ServeRemote(w http.ResponseWriter, r *http.Request, rem
 		defer cancel()
 		sr := r.WithContext(sctx)
 
+		// Cold start: with the syncer wired, an empty cache means the prime has
+		// not landed yet. Enqueue it and wait briefly rather than serving empty
+		// repodata; if it still has not primed, return a retryable 503.
+		if p.syncer != nil && !p.ensurePrimed(sctx, remote, store) {
+			w.Header().Set("Retry-After", "5")
+			http.Error(w, "metadata is being prepared, retry shortly", http.StatusServiceUnavailable)
+			return true
+		}
+
 		tail := strings.TrimPrefix(path, "repodata/")
 		lp := &Provider{}
 		switch {
@@ -154,6 +194,56 @@ func (p *GitHubProvider) ServeRemote(w http.ResponseWriter, r *http.Request, rem
 	return false
 }
 
+// onRequest keeps a remote's derived metadata fresh off the request path. With
+// the background syncer wired it enqueues a deduped, rate-limited, lease-gated
+// refresh and returns immediately; the request always serves the current cache.
+// Without a syncer (direct provider use / unit tests) it falls back to the
+// legacy inline single-flight scan.
+func (p *GitHubProvider) onRequest(remote models.Remote, store provider.RemoteMetadataStore) {
+	if p.syncer != nil {
+		p.syncer.enqueue(remote, false)
+		return
+	}
+	p.refresh(remote, store)
+}
+
+// ensurePrimed returns true once the remote has at least one cached metadata
+// row. On an empty cache it enqueues a prime and polls briefly for it to land,
+// so the very first client after a remote is created gets real repodata instead
+// of an empty index or a blocking multi-minute derive. Returns false if the
+// cache is still empty after the bounded wait.
+func (p *GitHubProvider) ensurePrimed(ctx context.Context, remote models.Remote, store provider.RemoteMetadataStore) bool {
+	if !p.cacheEmpty(ctx, store, remote.Name) {
+		return true
+	}
+	if p.syncer != nil {
+		p.syncer.enqueue(remote, true)
+	}
+
+	deadline := time.Now().Add(p.coldWait)
+	for time.Now().Before(deadline) {
+		select {
+		case <-ctx.Done():
+			return false
+		case <-time.After(400 * time.Millisecond):
+		}
+		if !p.cacheEmpty(ctx, store, remote.Name) {
+			return true
+		}
+	}
+	return false
+}
+
+func (p *GitHubProvider) cacheEmpty(ctx context.Context, store provider.RemoteMetadataStore, name string) bool {
+	rows, err := store.ListRPMMetadataEntries(ctx, name)
+	if err != nil {
+		// Treat a failed read as "not empty" so a transient DB error becomes a
+		// normal serve attempt (which reports its own error) rather than a 503.
+		return false
+	}
+	return len(rows) == 0
+}
+
 // refresh brings the derived metadata up to date without coupling the scan to
 // the inbound request. When the cache is stale it single-flights a scan: if the
 // cache already holds rows the scan runs in the background and the caller serves
@@ -212,15 +302,31 @@ func (p *GitHubProvider) runScan(remote models.Remote, store provider.RemoteMeta
 	p.mu.Unlock()
 }
 
+// scan runs a full unconditional derive. Retained for the legacy inline refresh
+// path and existing tests; the syncer uses scanWithState to pass and receive the
+// releases-list ETag.
 func (p *GitHubProvider) scan(ctx context.Context, remote models.Remote, store provider.RemoteMetadataStore) error {
-	releases, err := p.fetchReleases(ctx, remote)
+	_, _, err := p.scanWithState(ctx, remote, store, "")
+	return err
+}
+
+// scanWithState derives metadata incrementally. It sends the prior releases-list
+// ETag as a conditional request: a 304 means nothing changed, so it returns
+// (etag, changed=false) without a single asset fetch. On a 200 it diffs the
+// release assets against the cache, derives only new/changed assets, prunes
+// assets that disappeared, and returns the new ETag.
+func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote, store provider.RemoteMetadataStore, etag string) (newEtag string, changed bool, err error) {
+	releases, newEtag, notModified, err := p.fetchReleases(ctx, remote, etag)
 	if err != nil {
-		return err
+		return etag, false, err
+	}
+	if notModified {
+		return etag, false, nil
 	}
 
 	existing, err := store.ListRPMMetadataEntries(ctx, remote.Name)
 	if err != nil {
-		return err
+		return newEtag, false, err
 	}
 	existingByPath := make(map[string]provider.RPMMetadata, len(existing))
 	for _, m := range existing {
@@ -229,7 +335,7 @@ func (p *GitHubProvider) scan(ctx context.Context, remote models.Remote, store p
 
 	allow, err := compilePatterns(remote.Patterns)
 	if err != nil {
-		return err
+		return newEtag, false, err
 	}
 
 	seen := map[string]bool{}
@@ -277,7 +383,7 @@ func (p *GitHubProvider) scan(ctx context.Context, remote models.Remote, store p
 			_ = store.DeleteRPMMetadata(ctx, remote.Name, fp)
 		}
 	}
-	return nil
+	return newEtag, true, nil
 }
 
 type ghRelease struct {
@@ -293,32 +399,53 @@ type ghAsset struct {
 	Digest             string `json:"digest"`
 }
 
-func (p *GitHubProvider) fetchReleases(ctx context.Context, remote models.Remote) ([]ghRelease, error) {
+// fetchReleases lists a repo's releases. It sends the prior ETag as
+// If-None-Match on page 1 (the newest releases, where a new one first appears):
+// a 304 there means the repo is unchanged, so it returns notModified without
+// paging further — GitHub does not count 304 conditional responses against the
+// rate limit, making an unchanged repo nearly free. On a 200 it captures the
+// page-1 ETag and pages through the rest normally. Every call waits on the
+// shared limiter first.
+func (p *GitHubProvider) fetchReleases(ctx context.Context, remote models.Remote, etag string) (all []ghRelease, newEtag string, notModified bool, err error) {
 	base := strings.TrimRight(remote.BaseURL, "/") + "/releases"
-	var all []ghRelease
 	for page := 1; page <= p.pageCap; page++ {
 		u := fmt.Sprintf("%s?per_page=100&page=%d", base, page)
 		req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
 		if err != nil {
-			return nil, err
+			return nil, "", false, err
 		}
 		copyHeaders(req, githubHeaders(remote, true))
+		if page == 1 && etag != "" {
+			req.Header.Set("If-None-Match", etag)
+		}
 
+		if err := p.limiterWait(ctx); err != nil {
+			return nil, "", false, err
+		}
 		resp, err := p.client.Do(req)
 		if err != nil {
-			return nil, err
+			return nil, "", false, err
+		}
+		if page == 1 && resp.StatusCode == http.StatusNotModified {
+			io.Copy(io.Discard, resp.Body)
+			resp.Body.Close()
+			return nil, etag, true, nil
 		}
 		body, err := io.ReadAll(resp.Body)
+		respEtag := resp.Header.Get("ETag")
 		resp.Body.Close()
 		if err != nil {
-			return nil, err
+			return nil, "", false, err
 		}
 		if resp.StatusCode != http.StatusOK {
-			return nil, fmt.Errorf("github releases API %s: status %d", u, resp.StatusCode)
+			return nil, "", false, fmt.Errorf("github releases API %s: status %d", u, resp.StatusCode)
+		}
+		if page == 1 {
+			newEtag = respEtag
 		}
 		var releases []ghRelease
 		if err := json.Unmarshal(body, &releases); err != nil {
-			return nil, fmt.Errorf("decode releases: %w", err)
+			return nil, "", false, fmt.Errorf("decode releases: %w", err)
 		}
 		if len(releases) == 0 {
 			break
@@ -328,7 +455,7 @@ func (p *GitHubProvider) fetchReleases(ctx context.Context, remote models.Remote
 			break
 		}
 	}
-	return all, nil
+	return all, newEtag, false, nil
 }
 
 func (p *GitHubProvider) deriveAsset(ctx context.Context, remote models.Remote, asset ghAsset, fp string) (*provider.RPMMetadata, error) {
@@ -447,6 +574,9 @@ func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, dow
 	copyHeaders(req, githubHeaders(remote, false))
 	req.Header.Set("Range", fmt.Sprintf("bytes=0-%d", n-1))
 
+	if err := p.limiterWait(ctx); err != nil {
+		return nil, false, err
+	}
 	resp, err := p.client.Do(req)
 	if err != nil {
 		return nil, false, err
@@ -471,6 +601,9 @@ func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote
 	}
 	copyHeaders(req, githubHeaders(remote, false))
 
+	if err := p.limiterWait(ctx); err != nil {
+		return "", err
+	}
 	resp, err := p.client.Do(req)
 	if err != nil {
 		return "", err

internal/provider/rpm/github_test.go

@@ -66,11 +66,14 @@ func (f *fakeStore) ListRPMMetadataEntries(ctx context.Context, _ string) ([]pro
 // Range support) for a set of packages. digest controls whether the asset
 // carries a sha256 digest (no-download path) or not (compute path).
 type githubFixture struct {
-	srv      *httptest.Server
-	rpmBytes map[string][]byte // asset filename -> bytes
-	rangeHit map[string]int    // asset filename -> number of ranged GETs
-	fullHit  map[string]int    // asset filename -> number of full GETs
-	mu       sync.Mutex
+	srv         *httptest.Server
+	rpmBytes    map[string][]byte // asset filename -> bytes
+	rangeHit    map[string]int    // asset filename -> number of ranged GETs
+	fullHit     map[string]int    // asset filename -> number of full GETs
+	etag        string            // when set, served as ETag; matching If-None-Match yields 304
+	releasesHit int               // total releases-list requests (200 + 304)
+	notModHit   int               // releases-list requests answered 304
+	mu          sync.Mutex
 }
 
 func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
@@ -89,6 +92,19 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
 			w.Write([]byte("[]"))
 			return
 		}
+		f.mu.Lock()
+		f.releasesHit++
+		etag := f.etag
+		if etag != "" && r.Header.Get("If-None-Match") == etag {
+			f.notModHit++
+			f.mu.Unlock()
+			w.WriteHeader(http.StatusNotModified)
+			return
+		}
+		f.mu.Unlock()
+		if etag != "" {
+			w.Header().Set("ETag", etag)
+		}
 		var assets []map[string]any
 		for name := range f.rpmBytes {
 			a := map[string]any{

internal/provider/rpm/syncer.go

@@ -0,0 +1,256 @@
+package rpm
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"log/slog"
+	"os"
+	"sync"
+	"time"
+
+	"golang.org/x/time/rate"
+
+	"git.unkin.net/unkin/artifactapi/internal/provider"
+	"git.unkin.net/unkin/artifactapi/pkg/models"
+)
+
+const (
+	// syncLeaseDuration is how long a claimed sync lease is held before it is
+	// considered abandoned. It comfortably exceeds a scan's own timeout so a live
+	// scan never loses its lease, while a crashed replica's lease still expires.
+	syncLeaseDuration = 15 * time.Minute
+	// defaultSyncFreshness is the periodic re-check interval used when a remote's
+	// mutable_ttl is unset.
+	defaultSyncFreshness = 5 * time.Minute
+	// jobQueueDepth bounds the pending work queue; enqueues past it are dropped
+	// (a later poll re-enqueues), never blocking the caller.
+	jobQueueDepth = 256
+)
+
+// SyncStore is the persistence surface the syncer needs: the metadata cache it
+// primes plus the shared sync-state coordination (remote enumeration and the
+// per-remote lease). *database.DB satisfies it.
+type SyncStore interface {
+	provider.RemoteMetadataStore
+	ListGitHubRPMRemotes(ctx context.Context) ([]models.Remote, error)
+	ClaimGitHubSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (claimed bool, etag string, err error)
+	ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error
+}
+
+// SyncConfig tunes the shared syncer. Zero values fall back to safe defaults.
+type SyncConfig struct {
+	RatePerSec   float64       // global GitHub request rate (req/s)
+	Burst        int           // token-bucket burst
+	Workers      int           // concurrent scan workers
+	PollInterval time.Duration // base scheduler tick; per-remote cadence is mutable_ttl
+}
+
+type syncJob struct {
+	remote models.Remote
+	prime  bool
+}
+
+// Syncer is the single per-process background worker that keeps every
+// github_rpm remote's derived metadata fresh. It owns a deduped work queue, a
+// pool of workers, and a global token-bucket rate limiter shared across all
+// remotes and bound onto the github provider so every GitHub call it makes
+// passes through the same bucket. Periodic checks are gated by a shared DB lease
+// so, across replicas, only one performs each scan.
+type Syncer struct {
+	store   SyncStore
+	prov    *GitHubProvider
+	limiter *rate.Limiter
+	cfg     SyncConfig
+	owner   string
+
+	jobs   chan syncJob
+	mu     sync.Mutex
+	active map[string]bool // remotes queued or in-flight, for dedup/coalescing
+}
+
+// NewSyncer builds the syncer bound to the process-wide github provider
+// singleton. Call Run to start it.
+func NewSyncer(store SyncStore, cfg SyncConfig) *Syncer {
+	return newSyncer(store, gitHubProvider, cfg)
+}
+
+func newSyncer(store SyncStore, prov *GitHubProvider, cfg SyncConfig) *Syncer {
+	if cfg.RatePerSec <= 0 {
+		cfg.RatePerSec = 1
+	}
+	if cfg.Burst <= 0 {
+		cfg.Burst = 5
+	}
+	if cfg.Workers <= 0 {
+		cfg.Workers = 3
+	}
+	if cfg.PollInterval <= 0 {
+		cfg.PollInterval = 60 * time.Second
+	}
+
+	lim := rate.NewLimiter(rate.Limit(cfg.RatePerSec), cfg.Burst)
+	s := &Syncer{
+		store:   store,
+		prov:    prov,
+		limiter: lim,
+		cfg:     cfg,
+		owner:   leaseOwner(),
+		jobs:    make(chan syncJob, jobQueueDepth),
+		active:  map[string]bool{},
+	}
+	// Bind the shared limiter and back-reference so the request path routes
+	// through this syncer and every derive HTTP call is rate limited.
+	prov.limiter = lim
+	prov.syncer = s
+	return s
+}
+
+// Run starts the worker pool and the periodic scheduler and blocks until ctx is
+// canceled, at which point it drains in-flight scans and returns.
+func (s *Syncer) Run(ctx context.Context) {
+	slog.Info("github_rpm syncer started",
+		"rate_per_sec", s.cfg.RatePerSec, "burst", s.cfg.Burst,
+		"workers", s.cfg.Workers, "poll_interval", s.cfg.PollInterval, "owner", s.owner)
+
+	var wg sync.WaitGroup
+	for i := 0; i < s.cfg.Workers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			s.worker(ctx)
+		}()
+	}
+
+	ticker := time.NewTicker(s.cfg.PollInterval)
+	defer ticker.Stop()
+
+	s.schedule(ctx) // sweep at boot so existing remotes are checked immediately
+	for {
+		select {
+		case <-ctx.Done():
+			wg.Wait()
+			slog.Info("github_rpm syncer stopped")
+			return
+		case <-ticker.C:
+			s.schedule(ctx)
+		}
+	}
+}
+
+// schedule enqueues a periodic check for every github_rpm remote. The DB lease
+// (claimed in the worker) enforces the per-remote mutable_ttl cadence and cross
+// replica coordination, so enqueuing every tick is cheap: a not-yet-due remote
+// simply fails to claim and is skipped.
+func (s *Syncer) schedule(ctx context.Context) {
+	remotes, err := s.store.ListGitHubRPMRemotes(ctx)
+	if err != nil {
+		slog.Error("github_rpm syncer: list remotes", "error", err)
+		return
+	}
+	for _, r := range remotes {
+		s.enqueue(r, false)
+	}
+}
+
+// EnqueuePrime queues an immediate background prime for a freshly created
+// remote so its metadata is derived without blocking the create call.
+func (s *Syncer) EnqueuePrime(remote models.Remote) {
+	if s == nil {
+		return
+	}
+	s.enqueue(remote, true)
+}
+
+// enqueue adds a job unless the remote is already queued or in-flight, coalescing
+// duplicate requests down to one scan. It never blocks: a full queue drops the
+// job (a later poll re-enqueues it) after clearing the dedup slot.
+func (s *Syncer) enqueue(remote models.Remote, prime bool) {
+	s.mu.Lock()
+	if s.active[remote.Name] {
+		s.mu.Unlock()
+		return
+	}
+	s.active[remote.Name] = true
+	s.mu.Unlock()
+
+	select {
+	case s.jobs <- syncJob{remote: remote, prime: prime}:
+	default:
+		s.mu.Lock()
+		delete(s.active, remote.Name)
+		s.mu.Unlock()
+	}
+}
+
+func (s *Syncer) worker(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case job := <-s.jobs:
+			s.process(ctx, job)
+		}
+	}
+}
+
+// process claims the shared lease and, if won, runs an incremental scan. The
+// lease bounds total GitHub load to one scan per freshness window across all
+// replicas; losing the claim (another replica scanning, or not yet due) is a
+// no-op.
+func (s *Syncer) process(ctx context.Context, job syncJob) {
+	defer func() {
+		s.mu.Lock()
+		delete(s.active, job.remote.Name)
+		s.mu.Unlock()
+	}()
+
+	freshness := time.Duration(job.remote.MutableTTL) * time.Second
+	if freshness <= 0 {
+		freshness = defaultSyncFreshness
+	}
+	if job.prime {
+		freshness = 0 // prime ignores the recency gate but still respects a live lease
+	}
+
+	claimed, etag, err := s.store.ClaimGitHubSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration)
+	if err != nil {
+		slog.Error("github_rpm syncer: claim lease", "remote", job.remote.Name, "error", err)
+		return
+	}
+	if !claimed {
+		return
+	}
+
+	scanCtx, cancel := context.WithTimeout(ctx, s.prov.scanTimeout)
+	defer cancel()
+
+	newEtag, changed, scanErr := s.prov.scanWithState(scanCtx, job.remote, s.store, etag)
+	releaseEtag := etag
+	if scanErr == nil {
+		releaseEtag = newEtag
+	} else {
+		slog.Error("github_rpm syncer: scan failed", "remote", job.remote.Name, "error", scanErr)
+	}
+
+	// Release on a detached context so a clean shutdown mid-scan still frees the
+	// lease and advances last_synced_at (otherwise it simply expires).
+	relCtx, relCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
+	defer relCancel()
+	if err := s.store.ReleaseGitHubSyncLease(relCtx, job.remote.Name, s.owner, releaseEtag, time.Now()); err != nil {
+		slog.Warn("github_rpm syncer: release lease", "remote", job.remote.Name, "error", err)
+	}
+
+	if scanErr == nil && changed {
+		slog.Info("github_rpm syncer: refreshed", "remote", job.remote.Name, "prime", job.prime)
+	}
+}
+
+// leaseOwner is a per-replica identity for the lease: hostname plus a random
+// suffix so restarts and colocated replicas never collide.
+func leaseOwner() string {
+	host, _ := os.Hostname()
+	var b [6]byte
+	_, _ = rand.Read(b[:])
+	return host + "-" + hex.EncodeToString(b[:])
+}

internal/provider/rpm/syncer_test.go

@@ -0,0 +1,312 @@
+package rpm
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"golang.org/x/time/rate"
+
+	"git.unkin.net/unkin/artifactapi/internal/provider"
+	"git.unkin.net/unkin/artifactapi/internal/testsupport"
+	"git.unkin.net/unkin/artifactapi/pkg/models"
+)
+
+// fakeSyncStore is an in-memory SyncStore: the metadata cache (via the embedded
+// fakeStore) plus the shared sync-state lease, whose claim mirrors the atomic
+// semantics of the real SQL (recency gate AND no live lease).
+type fakeSyncStore struct {
+	*fakeStore
+
+	mu         sync.Mutex
+	remotes    []models.Remote
+	leaseOwner map[string]string
+	leaseExp   map[string]time.Time
+	lastSynced map[string]time.Time
+	etags      map[string]string
+}
+
+func newFakeSyncStore() *fakeSyncStore {
+	return &fakeSyncStore{
+		fakeStore:  newFakeStore(),
+		leaseOwner: map[string]string{},
+		leaseExp:   map[string]time.Time{},
+		lastSynced: map[string]time.Time{},
+		etags:      map[string]string{},
+	}
+}
+
+func (f *fakeSyncStore) ListGitHubRPMRemotes(_ context.Context) ([]models.Remote, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]models.Remote(nil), f.remotes...), nil
+}
+
+func (f *fakeSyncStore) ClaimGitHubSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration) (bool, string, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	now := time.Now()
+	ls, hasLS := f.lastSynced[name]
+	exp, hasExp := f.leaseExp[name]
+	freshOK := !hasLS || now.Sub(ls) >= freshness
+	leaseOK := !hasExp || exp.Before(now)
+	if freshOK && leaseOK {
+		f.leaseOwner[name] = owner
+		f.leaseExp[name] = now.Add(lease)
+		return true, f.etags[name], nil
+	}
+	return false, "", nil
+}
+
+func (f *fakeSyncStore) ReleaseGitHubSyncLease(_ context.Context, name, owner, etag string, syncedAt time.Time) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.leaseOwner[name] != owner {
+		return nil
+	}
+	f.lastSynced[name] = syncedAt
+	f.etags[name] = etag
+	delete(f.leaseOwner, name)
+	delete(f.leaseExp, name)
+	return nil
+}
+
+func testSyncConfig() SyncConfig {
+	return SyncConfig{RatePerSec: 1000, Burst: 100, Workers: 1, PollInterval: time.Hour}
+}
+
+// (a) A 304 conditional response must derive nothing: no asset header GETs and
+// changed=false, so an unchanged repo is nearly free.
+func TestSyncerConditionalNotModifiedSkipsDerive(t *testing.T) {
+	fx := newGitHubFixture(t, true)
+	fx.etag = `"v1"`
+	p := newTestProvider()
+	store := newFakeStore()
+	remote := fx.remote()
+
+	etag1, changed, err := p.scanWithState(context.Background(), remote, store, "")
+	if err != nil {
+		t.Fatalf("first scan: %v", err)
+	}
+	if !changed || etag1 != `"v1"` {
+		t.Fatalf("first scan changed=%v etag=%q, want true and \"v1\"", changed, etag1)
+	}
+	priorRange := fx.rangeHit["demo-1.2-3.x86_64.rpm"]
+	if priorRange == 0 {
+		t.Fatal("first scan should have fetched the asset header")
+	}
+
+	etag2, changed2, err := p.scanWithState(context.Background(), remote, store, etag1)
+	if err != nil {
+		t.Fatalf("second scan: %v", err)
+	}
+	if changed2 {
+		t.Fatal("304 scan must report changed=false")
+	}
+	if etag2 != etag1 {
+		t.Fatalf("etag changed across 304: %q -> %q", etag1, etag2)
+	}
+	if fx.notModHit != 1 {
+		t.Fatalf("want exactly one 304 releases response, got %d", fx.notModHit)
+	}
+	if got := fx.rangeHit["demo-1.2-3.x86_64.rpm"]; got != priorRange {
+		t.Fatalf("304 scan re-fetched asset header: %d -> %d", priorRange, got)
+	}
+}
+
+// (b) On a real change, only the newly added asset is derived; assets already
+// cached are never re-fetched.
+func TestSyncerIncrementalDerivesOnlyNewAsset(t *testing.T) {
+	fx := newGitHubFixture(t, true)
+	fx.etag = `"v1"`
+	p := newTestProvider()
+	store := newFakeStore()
+	remote := fx.remote()
+
+	if _, _, err := p.scanWithState(context.Background(), remote, store, ""); err != nil {
+		t.Fatalf("first scan: %v", err)
+	}
+	demoRange := fx.rangeHit["demo-1.2-3.x86_64.rpm"]
+
+	// Add a new asset and bump the ETag so the conditional request returns 200.
+	fx.rpmBytes["other-9-9.aarch64.rpm"] = testsupport.MinimalRPM("other", "9", "9", "aarch64")
+	fx.etag = `"v2"`
+
+	if _, changed, err := p.scanWithState(context.Background(), remote, store, `"v1"`); err != nil || !changed {
+		t.Fatalf("second scan changed=%v err=%v", changed, err)
+	}
+
+	rows, _ := store.ListRPMMetadataEntries(context.Background(), remote.Name)
+	if len(rows) != 2 {
+		t.Fatalf("want 2 cached rows after incremental derive, got %d", len(rows))
+	}
+	if got := fx.rangeHit["demo-1.2-3.x86_64.rpm"]; got != demoRange {
+		t.Fatalf("already-cached asset was re-fetched: %d -> %d", demoRange, got)
+	}
+	if fx.rangeHit["other-9-9.aarch64.rpm"] == 0 {
+		t.Fatal("newly added asset was not derived")
+	}
+}
+
+// (c) The shared limiter caps the request rate: three gated releases calls at
+// one token per 120ms cannot complete faster than ~2 gaps.
+func TestRateLimiterCapsRequestRate(t *testing.T) {
+	fx := newGitHubFixture(t, true)
+	p := newTestProvider()
+	p.limiter = rate.NewLimiter(rate.Every(120*time.Millisecond), 1)
+	remote := fx.remote()
+
+	start := time.Now()
+	for i := 0; i < 3; i++ {
+		if _, _, _, err := p.fetchReleases(context.Background(), remote, ""); err != nil {
+			t.Fatalf("fetchReleases %d: %v", i, err)
+		}
+	}
+	if elapsed := time.Since(start); elapsed < 200*time.Millisecond {
+		t.Fatalf("rate limiter did not throttle: 3 calls took %v, want >= 200ms", elapsed)
+	}
+}
+
+// (d) Concurrent enqueues for the same remote coalesce to a single queued job.
+func TestSyncerEnqueueDedup(t *testing.T) {
+	store := newFakeSyncStore()
+	p := newTestProvider()
+	s := newSyncer(store, p, testSyncConfig())
+	remote := models.Remote{Name: "acme-rpm", PackageType: models.PackageGitHubRPM, MutableTTL: 3600}
+
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() { defer wg.Done(); s.enqueue(remote, false) }()
+	}
+	wg.Wait()
+
+	if got := len(s.jobs); got != 1 {
+		t.Fatalf("want exactly 1 coalesced job, got %d", got)
+	}
+}
+
+// (e) Prime-on-create enqueues a prime job.
+func TestSyncerEnqueuePrime(t *testing.T) {
+	store := newFakeSyncStore()
+	p := newTestProvider()
+	s := newSyncer(store, p, testSyncConfig())
+	remote := models.Remote{Name: "acme-rpm", PackageType: models.PackageGitHubRPM, MutableTTL: 3600}
+
+	s.EnqueuePrime(remote)
+	select {
+	case job := <-s.jobs:
+		if !job.prime || job.remote.Name != "acme-rpm" {
+			t.Fatalf("bad prime job: %+v", job)
+		}
+	default:
+		t.Fatal("EnqueuePrime did not enqueue a job")
+	}
+}
+
+// (f) A held lease prevents a second replica from scanning: with the lease owned
+// by another replica, process claims nothing and makes zero GitHub calls.
+func TestSyncerLeasePreventsSecondReplica(t *testing.T) {
+	fx := newGitHubFixture(t, true)
+	fx.etag = `"v1"`
+	store := newFakeSyncStore()
+	p := newTestProvider()
+	s := newSyncer(store, p, testSyncConfig())
+	remote := fx.remote()
+
+	// Replica 1 holds the lease.
+	claimed, _, err := store.ClaimGitHubSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration)
+	if err != nil || !claimed {
+		t.Fatalf("replica-1 claim: claimed=%v err=%v", claimed, err)
+	}
+
+	// Replica 2 (this syncer) tries to process the same remote; it must skip.
+	s.process(context.Background(), syncJob{remote: remote})
+
+	if fx.releasesHit != 0 {
+		t.Fatalf("second replica scanned while lease held: %d releases calls", fx.releasesHit)
+	}
+	if rows, _ := store.ListRPMMetadataEntries(context.Background(), remote.Name); len(rows) != 0 {
+		t.Fatalf("second replica derived metadata while lease held: %d rows", len(rows))
+	}
+}
+
+// With the syncer wired and the cache empty, a repodata request enqueues a
+// prime and, when it has not landed within the bounded cold wait, returns a
+// retryable 503 rather than serving empty repodata (and without regressing the
+// detached-context serve).
+func TestServeRemoteColdStartReturns503(t *testing.T) {
+	fx := newGitHubFixture(t, true)
+	store := newFakeSyncStore()
+	p := newTestProvider()
+	p.coldWait = 300 * time.Millisecond
+	_ = newSyncer(store, p, testSyncConfig()) // binds p.syncer, but no workers running
+	remote := fx.remote()
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, "/api/v1/remote/acme-rpm/repodata/repomd.xml", nil)
+	if !p.ServeRemote(rec, req, remote, "repodata/repomd.xml", "https://x", store) {
+		t.Fatal("ServeRemote did not handle repomd.xml")
+	}
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Fatalf("cold empty cache must return 503, got %d", rec.Code)
+	}
+	if rec.Header().Get("Retry-After") == "" {
+		t.Fatal("503 should carry Retry-After")
+	}
+	// The prime was enqueued.
+	if got := len(p.syncer.jobs); got != 1 {
+		t.Fatalf("cold start did not enqueue a prime, jobs=%d", got)
+	}
+}
+
+// With the cache warm, the same request serves repodata immediately (no 503).
+func TestServeRemoteWarmCacheServesImmediately(t *testing.T) {
+	fx := newGitHubFixture(t, true)
+	store := newFakeSyncStore()
+	p := newTestProvider()
+	_ = newSyncer(store, p, testSyncConfig())
+	remote := fx.remote()
+
+	if err := p.scan(context.Background(), remote, store); err != nil {
+		t.Fatalf("warm scan: %v", err)
+	}
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, "/api/v1/remote/acme-rpm/repodata/repomd.xml", nil)
+	if !p.ServeRemote(rec, req, remote, "repodata/repomd.xml", "https://x", store) {
+		t.Fatal("ServeRemote did not handle repomd.xml")
+	}
+	if rec.Code != http.StatusOK {
+		t.Fatalf("warm cache must serve 200, got %d body=%s", rec.Code, rec.Body.String())
+	}
+}
+
+// A prime job (freshness 0) runs even right after a sync, deriving metadata,
+// while a periodic job at the same moment is gated by the recency window.
+func TestSyncerPrimeBypassesRecencyPeriodicDoesNot(t *testing.T) {
+	fx := newGitHubFixture(t, true)
+	fx.etag = `"v1"`
+	store := newFakeSyncStore()
+	p := newTestProvider()
+	s := newSyncer(store, p, testSyncConfig())
+	remote := fx.remote()
+
+	var _ provider.RemoteMetadataStore = store
+
+	// Prime derives despite no prior sync.
+	s.process(context.Background(), syncJob{remote: remote, prime: true})
+	if rows, _ := store.ListRPMMetadataEntries(context.Background(), remote.Name); len(rows) != 1 {
+		t.Fatalf("prime did not derive: %d rows", len(rows))
+	}
+	releasesAfterPrime := fx.releasesHit
+
+	// A periodic job immediately after is gated by mutable_ttl recency: no new
+	// releases call.
+	s.process(context.Background(), syncJob{remote: remote, prime: false})
+	if fx.releasesHit != releasesAfterPrime {
+		t.Fatalf("periodic scan ran inside recency window: %d -> %d releases calls", releasesAfterPrime, fx.releasesHit)
+	}
+}