From 6416cc0494129a09ade1bc3b98564000dc8edeea Mon Sep 17 00:00:00 2001 From: Elizabeth Worstell Date: Mon, 20 Jul 2026 19:03:26 -0700 Subject: [PATCH] feat(git): coordinate snapshot generation across replicas MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every replica independently runs the periodic snapshot, mirror-snapshot, and LFS-snapshot jobs for each repository, so N replicas regenerate and re-upload the same artifact every interval. For large repositories an LFS snapshot generation takes ~20 minutes, multiplied across every replica, every interval — and because a new pod's schedule store is empty, every deploy triggers an immediate full regeneration on every replica at once. Share per-artifact generation state through the metadata store: before generating, a replica checks whether a peer completed a generation within the interval or holds an unexpired in-progress claim, and skips if so. Claims expire after 30 minutes so a crashed generator can't block peers indefinitely. Steady-state ticks get per-replica jitter and first runs after registration are spread over a few minutes, so claims (which sync asynchronously) propagate before peers decide. Coordination is advisory — the store is last-write-wins — so replicas deciding within the same sync window can still generate concurrently; that's rare with the spread, and a duplicate generation is wasteful but harmless. Without a metadata store, behavior is unchanged. Co-authored-by: Amp Amp-Thread-ID: https://ampcode.com/threads/T-019f85b8-d346-7243-961f-fbf2236a7f94 --- internal/strategy/git/git.go | 53 ++++-- internal/strategy/git/git_test.go | 8 +- internal/strategy/git/snapshot.go | 79 ++++++++- internal/strategy/git/snapshotcoord.go | 115 ++++++++++++ internal/strategy/git/snapshotcoord_test.go | 184 ++++++++++++++++++++ 5 files changed, 420 insertions(+), 19 deletions(-) create mode 100644 internal/strategy/git/snapshotcoord.go create mode 100644 internal/strategy/git/snapshotcoord_test.go diff --git a/internal/strategy/git/git.go b/internal/strategy/git/git.go index e6e941cc..78be557a 100644 --- a/internal/strategy/git/git.go +++ b/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) diff --git a/internal/strategy/git/git_test.go b/internal/strategy/git/git_test.go index e2a4ff2c..b0651e1f 100644 --- a/internal/strategy/git/git_test.go +++ b/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") } diff --git a/internal/strategy/git/snapshot.go b/internal/strategy/git/snapshot.go index 6781e9fd..d54dd106 100644 --- a/internal/strategy/git/snapshot.go +++ b/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) diff --git a/internal/strategy/git/snapshotcoord.go b/internal/strategy/git/snapshotcoord.go new file mode 100644 index 00000000..f207c074 --- /dev/null +++ b/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 +} diff --git a/internal/strategy/git/snapshotcoord_test.go b/internal/strategy/git/snapshotcoord_test.go new file mode 100644 index 00000000..444aad24 --- /dev/null +++ b/internal/strategy/git/snapshotcoord_test.go @@ -0,0 +1,184 @@ +package git //nolint:testpackage // white-box testing required for clock injection + +import ( + "context" + "log/slog" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + + "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/metadatadb" +) + +func newTestSnapshotCoordinators(t *testing.T, now func() time.Time, replicas int) []*SnapshotCoordinator { + t.Helper() + ctx := logging.ContextWithLogger(context.Background(), slog.Default()) + backend := metadatadb.NewMemoryBackend() + coords := make([]*SnapshotCoordinator, replicas) + for i := range coords { + store := metadatadb.New(ctx, backend) + coords[i] = NewSnapshotCoordinator(store.Namespace("git")) + coords[i].now = now + } + return coords +} + +func TestSnapshotCoordinatorNilSafe(t *testing.T) { + var c *SnapshotCoordinator + claimed, err := c.Claim("snapshot", "https://github.com/foo/bar", time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) + assert.NoError(t, c.Complete("snapshot", "https://github.com/foo/bar")) + assert.NoError(t, c.Prime(context.Background())) + assert.Zero(t, NewSnapshotCoordinator(nil)) +} + +func TestSnapshotCoordinatorFreshArtifactSkips(t *testing.T) { + clock := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC) + coords := newTestSnapshotCoordinators(t, func() time.Time { return clock }, 2) + const upstream = "https://github.com/foo/bar" + + claimed, err := coords[0].Claim("snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) + clock = clock.Add(2 * time.Minute) + assert.NoError(t, coords[0].Complete("snapshot", upstream)) + + clock = clock.Add(10 * time.Minute) + claimed, err = coords[1].Claim("snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.False(t, claimed) + + clock = clock.Add(time.Hour) + claimed, err = coords[1].Claim("snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) +} + +func TestSnapshotCoordinatorFreshUntilFullInterval(t *testing.T) { + clock := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC) + coords := newTestSnapshotCoordinators(t, func() time.Time { return clock }, 2) + const upstream = "https://github.com/foo/bar" + + claimed, err := coords[0].Claim("snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) + assert.NoError(t, coords[0].Complete("snapshot", upstream)) + + // A completion just shy of the full interval still suppresses peers. + clock = clock.Add(time.Hour - time.Minute) + claimed, err = coords[1].Claim("snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.False(t, claimed) + + clock = clock.Add(time.Minute) + claimed, err = coords[1].Claim("snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) +} + +func TestSnapshotCoordinatorShortIntervalStillSuppresses(t *testing.T) { + clock := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC) + coords := newTestSnapshotCoordinators(t, func() time.Time { return clock }, 2) + const upstream = "https://github.com/foo/bar" + + claimed, err := coords[0].Claim("snapshot", upstream, 5*time.Minute) + assert.NoError(t, err) + assert.True(t, claimed) + assert.NoError(t, coords[0].Complete("snapshot", upstream)) + + clock = clock.Add(2 * time.Minute) + claimed, err = coords[1].Claim("snapshot", upstream, 5*time.Minute) + assert.NoError(t, err) + assert.False(t, claimed) +} + +func TestSnapshotCoordinatorInProgressClaimSuppressesPeers(t *testing.T) { + clock := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC) + coords := newTestSnapshotCoordinators(t, func() time.Time { return clock }, 2) + const upstream = "https://github.com/foo/bar" + + claimed, err := coords[0].Claim("lfs-snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) + + clock = clock.Add(5 * time.Minute) + claimed, err = coords[1].Claim("lfs-snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.False(t, claimed) + + // An expired claim (crashed generator) no longer suppresses peers. + clock = clock.Add(snapshotClaimTTL) + claimed, err = coords[1].Claim("lfs-snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) +} + +func TestSnapshotCoordinatorFailedGenerationDoesNotMarkFresh(t *testing.T) { + clock := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC) + coords := newTestSnapshotCoordinators(t, func() time.Time { return clock }, 2) + const upstream = "https://github.com/foo/bar" + + claimed, err := coords[0].Claim("snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) + // Generation fails: Complete is never called. After the claim expires the + // peer generates instead of waiting a full interval. + clock = clock.Add(snapshotClaimTTL) + claimed, err = coords[1].Claim("snapshot", upstream, time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) +} + +func TestSnapshotCoordinatorKeysAreIndependent(t *testing.T) { + clock := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC) + coords := newTestSnapshotCoordinators(t, func() time.Time { return clock }, 1) + c := coords[0] + + claimed, err := c.Claim("snapshot", "https://github.com/foo/bar", time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) + assert.NoError(t, c.Complete("snapshot", "https://github.com/foo/bar")) + + for _, job := range []string{"lfs-snapshot", "mirror-snapshot"} { + claimed, err = c.Claim(job, "https://github.com/foo/bar", time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) + } + claimed, err = c.Claim("snapshot", "https://github.com/foo/other", time.Hour) + assert.NoError(t, err) + assert.True(t, claimed) +} + +func TestJitterInterval(t *testing.T) { + assert.Equal(t, time.Duration(0), jitterInterval(0)) + for range 100 { + j := jitterInterval(time.Hour) + assert.True(t, j >= time.Hour) + assert.True(t, j < time.Hour+time.Hour/8) + } +} + +func TestStartupSpreadDelay(t *testing.T) { + for range 100 { + d := startupSpreadDelay() + assert.True(t, d >= 0) + assert.True(t, d < snapshotStartupSpread) + } +} + +func TestSnapshotSchedule(t *testing.T) { + uncoordinated := &Strategy{} + delay, interval := uncoordinated.snapshotSchedule(time.Hour) + assert.Equal(t, time.Duration(0), delay) + assert.Equal(t, time.Hour, interval) + + coordinated := &Strategy{snapshotCoord: newTestSnapshotCoordinators(t, time.Now, 1)[0]} + for range 100 { + delay, interval = coordinated.snapshotSchedule(time.Hour) + assert.True(t, delay >= 0 && delay < snapshotStartupSpread) + assert.True(t, interval >= time.Hour && interval < time.Hour+time.Hour/8) + } +}