feat(git): coordinate snapshot generation across replicas

internal/strategy/git/git.go

@@ -68,6 +68,9 @@ type Strategy struct {
 	deferredRestoreOnce sync.Map // keyed by upstream URL, ensures at most one deferred restore per repo
 	metrics             *gitMetrics
 	repoCounts          *RepoCounts
+	snapshotCoord       *SnapshotCoordinator
+	metadataWired       chan struct{} // closed by SetMetadataStore; gates warm-up
+	wiredOnce           sync.Once
 	ready               atomic.Bool
 }
 
@@ -129,21 +132,44 @@ func New(
 	m := newGitMetrics()
 
 	s := &Strategy{
-		config:       config,
-		cache:        cache,
-		cloneManager: cloneManager,
-		httpClient:   http.DefaultClient,
-		ctx:          ctx,
-		scheduler:    scheduler.WithQueuePrefix("git"),
-		spools:       make(map[string]*RepoSpools),
-		tokenManager: tokenManager,
-		metrics:      m,
+		config:        config,
+		cache:         cache,
+		cloneManager:  cloneManager,
+		httpClient:    http.DefaultClient,
+		ctx:           ctx,
+		scheduler:     scheduler.WithQueuePrefix("git"),
+		spools:        make(map[string]*RepoSpools),
+		tokenManager:  tokenManager,
+		metrics:       m,
+		metadataWired: make(chan struct{}),
 	}
 	// Run startup fetches in the background so the HTTP listener (and
 	// /_liveness) come up immediately. /_readiness gates on Ready() so the
 	// Service load balancer holds traffic until warming completes.
 	go func() {
 		warmCtx := context.WithoutCancel(ctx)
+		// Coordination only matters when warm-up will schedule snapshot
+		// jobs; with snapshots disabled, warm immediately.
+		if s.config.SnapshotInterval > 0 {
+			// Wait for SetMetadataStore so warm-up never schedules snapshot
+			// jobs before cross-replica coordination is installed.
+			// config.Load wires every MetadataConsumer immediately after
+			// construction.
+			select {
+			case <-s.metadataWired:
+			case <-ctx.Done():
+				return
+			}
+			// Sync shared coordination state before scheduling so the first
+			// claim decisions don't run against an empty local view. Bounded
+			// and fail-open (coordination is advisory) so a slow metadata
+			// backend can't wedge readiness.
+			primeCtx, cancel := context.WithTimeout(warmCtx, time.Minute)
+			if err := s.snapshotCoord.Prime(primeCtx); err != nil {
+				logger.WarnContext(warmCtx, "Failed to prime snapshot coordination state", "error", err)
+			}
+			cancel()
+		}
 		if err := s.warmExistingRepos(warmCtx); err != nil {
 			logger.WarnContext(warmCtx, "Failed to warm existing repos", "error", err)
 		}
@@ -197,12 +223,17 @@ func (s *Strategy) Ready() bool {
 	return s.ready.Load()
 }
 
-// SetMetadataStore enables the per-repo clone histogram and schedules its
-// daily reaper. Called by config.Load after the metadata backend is built.
+// SetMetadataStore enables the per-repo clone histogram (and schedules its
+// daily reaper) and cross-replica snapshot generation coordination. Called by
+// config.Load after the metadata backend is built. It also releases the
+// warm-up goroutine, which waits for wiring so warmed repos are scheduled
+// with coordination in place.
 func (s *Strategy) SetMetadataStore(store *metadatadb.Store) {
+	defer s.wiredOnce.Do(func() { close(s.metadataWired) })
 	if store == nil {
 		return
 	}
+	s.snapshotCoord = NewSnapshotCoordinator(store.Namespace("git"))
 	s.repoCounts = NewRepoCounts(store.Namespace("git"))
 	logging.FromContext(s.ctx).InfoContext(s.ctx, "Per-repo clone histogram enabled",
 		"retention_days", s.repoCounts.retentionDays)

internal/strategy/git/git_test.go

@@ -55,6 +55,9 @@ func newTestScheduler(ctx context.Context, t *testing.T) jobscheduler.Provider {
 // warm-up goroutine started in git.New.
 func waitForReady(t *testing.T, s *git.Strategy) {
 	t.Helper()
+	// Simulate config.Load's wiring step, which releases the warm-up
+	// goroutine. Idempotent, so repeated calls are safe.
+	s.SetMetadataStore(nil)
 	deadline := time.Now().Add(5 * time.Second)
 	for !s.Ready() && time.Now().Before(deadline) {
 		time.Sleep(10 * time.Millisecond)
@@ -224,10 +227,7 @@ func TestNewIsReadyAfterWarm(t *testing.T) {
 	s, err := git.New(ctx, git.Config{}, newTestScheduler(ctx, t), nil, mux, cm, func() (*githubapp.TokenManager, error) { return nil, nil }) //nolint:nilnil
 	assert.NoError(t, err)
 
-	deadline := time.Now().Add(5 * time.Second)
-	for !s.Ready() && time.Now().Before(deadline) {
-		time.Sleep(10 * time.Millisecond)
-	}
+	waitForReady(t, s)
 	assert.True(t, s.Ready(), "strategy should be ready after warm-up completes")
 }
 

internal/strategy/git/snapshot.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math/rand/v2"
 	"net/http"
 	"os"
 	"os/exec"
@@ -202,25 +203,95 @@ func (s *Strategy) generateAndUploadMirrorSnapshot(ctx context.Context, repo *gi
 	return nil
 }
 
+// snapshotStartupSpread staggers each replica's first coordinated snapshot
+// run after registration. Periodic jobs run immediately when this pod has no
+// recorded last run, so on a deploy every replica would otherwise decide
+// within the same metadata sync window, before peers' claims propagate.
+const snapshotStartupSpread = 5 * time.Minute
+
 func (s *Strategy) scheduleSnapshotJobs(repo *gitclone.Repository) {
-	s.scheduler.SubmitPeriodicJob(repo.UpstreamURL(), "snapshot-periodic", s.config.SnapshotInterval, func(ctx context.Context) error {
+	upstream := repo.UpstreamURL()
+	submit := func(job string, interval time.Duration, generate func(ctx context.Context) error) {
+		run := s.coordinatedSnapshotJob(job, repo, interval, generate)
+		delay, interval := s.snapshotSchedule(interval)
+		if delay == 0 {
+			s.scheduler.SubmitPeriodicJob(upstream, job+"-periodic", interval, run)
+			return
+		}
+		// SubmitPeriodicJob is a no-op once the scheduler is draining or torn
+		// down, so a timer that fires during shutdown is harmless.
+		time.AfterFunc(delay, func() {
+			s.scheduler.SubmitPeriodicJob(upstream, job+"-periodic", interval, run)
+		})
+	}
+	submit("snapshot", s.config.SnapshotInterval, func(ctx context.Context) error {
 		if err := s.doFetch(ctx, repo); err != nil {
-			logging.FromContext(ctx).WarnContext(ctx, "Pre-snapshot fetch failed", "upstream", repo.UpstreamURL(), "error", err)
+			logging.FromContext(ctx).WarnContext(ctx, "Pre-snapshot fetch failed", "upstream", upstream, "error", err)
 		}
 		return s.generateAndUploadSnapshot(ctx, repo)
 	})
-	s.scheduler.SubmitPeriodicJob(repo.UpstreamURL(), "lfs-snapshot-periodic", s.config.SnapshotInterval, func(ctx context.Context) error {
+	submit("lfs-snapshot", s.config.SnapshotInterval, func(ctx context.Context) error {
 		return s.generateAndUploadLFSSnapshot(ctx, repo)
 	})
 	mirrorInterval := s.config.MirrorSnapshotInterval
 	if mirrorInterval == 0 {
 		mirrorInterval = s.config.SnapshotInterval
 	}
-	s.scheduler.SubmitPeriodicJob(repo.UpstreamURL(), "mirror-snapshot-periodic", mirrorInterval, func(ctx context.Context) error {
+	submit("mirror-snapshot", mirrorInterval, func(ctx context.Context) error {
 		return s.generateAndUploadMirrorSnapshot(ctx, repo)
 	})
 }
 
+func startupSpreadDelay() time.Duration {
+	return rand.N(snapshotStartupSpread) //nolint:gosec // scheduling jitter needs no cryptographic randomness
+}
+
+// Spread and jitter only help when replicas coordinate through shared
+// metadata; without a coordinator, preserve immediate registration at the
+// exact configured interval.
+func (s *Strategy) snapshotSchedule(interval time.Duration) (delay, jittered time.Duration) {
+	if s.snapshotCoord == nil {
+		return 0, interval
+	}
+	return startupSpreadDelay(), jitterInterval(interval)
+}
+
+// coordinatedSnapshotJob wraps a snapshot generation job with cross-replica
+// coordination: the job is skipped when another replica generated the
+// artifact recently or is generating it now. Coordination failures fail open
+// so a broken metadata store never stops snapshot generation.
+func (s *Strategy) coordinatedSnapshotJob(job string, repo *gitclone.Repository, interval time.Duration, generate func(ctx context.Context) error) func(ctx context.Context) error {
+	upstream := repo.UpstreamURL()
+	return func(ctx context.Context) error {
+		logger := logging.FromContext(ctx)
+		claimed, err := s.snapshotCoord.Claim(job, upstream, interval)
+		if err != nil {
+			logger.WarnContext(ctx, "Snapshot coordination claim failed, generating anyway", "job", job, "upstream", upstream, "error", err)
+		} else if !claimed {
+			logger.DebugContext(ctx, "Skipping snapshot generation, fresh or in progress on another replica", "job", job, "upstream", upstream)
+			return nil
+		}
+		if err := generate(ctx); err != nil {
+			return err
+		}
+		if err := s.snapshotCoord.Complete(job, upstream); err != nil {
+			logger.WarnContext(ctx, "Failed to record snapshot completion", "job", job, "upstream", upstream, "error", err)
+		}
+		return nil
+	}
+}
+
+// jitterInterval spreads replicas' periodic snapshot schedules apart so that
+// coordination claims (synced asynchronously between replicas) propagate
+// before a peer decides whether to generate. Without it, replicas deployed
+// together tick in lockstep and race inside the sync window.
+func jitterInterval(interval time.Duration) time.Duration {
+	if interval <= 0 {
+		return interval
+	}
+	return interval + rand.N(interval/8) //nolint:gosec // scheduling jitter needs no cryptographic randomness
+}
+
 func (s *Strategy) snapshotMutexFor(upstreamURL string) *sync.Mutex {
 	mu, _ := s.snapshotMu.LoadOrStore(upstreamURL, &sync.Mutex{})
 	return mu.(*sync.Mutex)

internal/strategy/git/snapshotcoord.go

@@ -0,0 +1,115 @@
+package git
+
+import (
+	"context"
+	"time"
+
+	"github.com/alecthomas/errors"
+
+	"github.com/block/cachew/internal/metadatadb"
+)
+
+const (
+	snapshotGenMapName = "snapshot_generations"
+	// '|' cannot appear in a Git upstream URL, so the split is unambiguous.
+	snapshotGenKeySeparator = "|"
+	// snapshotClaimTTL bounds how long an in-progress claim suppresses other
+	// replicas. It exceeds lfsFetchTimeout so a live LFS generation is never
+	// preempted. Peers that skipped re-arm for a full interval rather than
+	// re-checking at expiry (which would herd them into the same sync
+	// window), so a crashed generator delays regeneration by up to this TTL
+	// plus one interval in the worst case.
+	snapshotClaimTTL = 30 * time.Minute
+)
+
+// snapshotGenRecord is the shared per-artifact generation state. StartedAt is
+// the most recent claim; CompletedAt is the most recent successful generation.
+type snapshotGenRecord struct {
+	StartedAt   time.Time `json:"started_at"`
+	CompletedAt time.Time `json:"completed_at,omitzero"`
+}
+
+// SnapshotCoordinator shares per-artifact generation state across replicas so
+// that each interval one replica regenerates a given snapshot instead of all
+// of them. Coordination is advisory: the metadata store is last-write-wins
+// and syncs asynchronously, so replicas that decide within the same sync
+// window can still generate concurrently — jittered schedules make that rare,
+// and a duplicate generation is wasteful but harmless.
+//
+// All methods are nil-safe; without a metadata store every replica generates.
+type SnapshotCoordinator struct {
+	ns   *metadatadb.Namespace
+	gens *metadatadb.Map[string, snapshotGenRecord]
+	now  func() time.Time
+}
+
+// NewSnapshotCoordinator returns nil if ns is nil so callers don't need a
+// separate "no metadata configured" code path.
+func NewSnapshotCoordinator(ns *metadatadb.Namespace) *SnapshotCoordinator {
+	if ns == nil {
+		return nil
+	}
+	return &SnapshotCoordinator{
+		ns:   ns,
+		gens: metadatadb.NewMap[string, snapshotGenRecord](ns, snapshotGenMapName),
+		now:  time.Now,
+	}
+}
+
+// Prime forces a synchronous refresh of shared state. Backends that sync
+// asynchronously (S3) populate a fresh replica's local view lazily, so
+// without a prime the first claim after startup could run against an empty
+// view and regenerate an artifact a peer completed recently.
+func (c *SnapshotCoordinator) Prime(ctx context.Context) error {
+	if c == nil {
+		return nil
+	}
+	return errors.Wrap(c.ns.Flush(ctx), "prime snapshot coordination state")
+}
+
+// Claim reports whether this replica should generate the artifact now, and
+// records the claim when it should. It declines when another replica
+// completed a generation within the interval, or holds an unexpired
+// in-progress claim.
+func (c *SnapshotCoordinator) Claim(job, upstreamURL string, interval time.Duration) (bool, error) {
+	if c == nil {
+		return true, nil
+	}
+	key := snapshotGenKey(job, upstreamURL)
+	now := c.now()
+	rec, ok := c.gens.Get(key)
+	if ok {
+		// A completion within the interval means the artifact is still fresh.
+		// This also absorbs schedule drift between replicas: a tick that lands
+		// just before the generator's next one sees an almost-interval-old
+		// completion and skips rather than duplicating the imminent generation.
+		if !rec.CompletedAt.IsZero() && now.Sub(rec.CompletedAt) < interval {
+			return false, nil
+		}
+		inProgress := rec.CompletedAt.Before(rec.StartedAt)
+		if inProgress && now.Sub(rec.StartedAt) < snapshotClaimTTL {
+			return false, nil
+		}
+	}
+	rec.StartedAt = now
+	if err := c.gens.Set(key, rec); err != nil {
+		return true, errors.Wrap(err, "record snapshot claim")
+	}
+	return true, nil
+}
+
+// Complete records a successful generation so other replicas skip the
+// artifact until it goes stale again.
+func (c *SnapshotCoordinator) Complete(job, upstreamURL string) error {
+	if c == nil {
+		return nil
+	}
+	key := snapshotGenKey(job, upstreamURL)
+	rec, _ := c.gens.Get(key)
+	rec.CompletedAt = c.now()
+	return errors.Wrap(c.gens.Set(key, rec), "record snapshot completion")
+}
+
+func snapshotGenKey(job, upstreamURL string) string {
+	return job + snapshotGenKeySeparator + upstreamURL
+}