fix(git): prepare cold snapshot misses in background

internal/gitclone/manager.go

@@ -415,13 +415,56 @@ func (r *Repository) MarkReady() {
 	r.mu.Unlock()
 }
 
+// Clone starts one mirror clone. Other callers wait for the result.
 func (r *Repository) Clone(ctx context.Context) error {
+	return r.clone(ctx, 0)
+}
+
+// CloneWithWaitTimeout limits how long a caller waits for another clone.
+// The limit does not apply to the caller that starts the clone.
+func (r *Repository) CloneWithWaitTimeout(ctx context.Context, waitTimeout time.Duration) error {
+	return r.clone(ctx, waitTimeout)
+}
+
+func (r *Repository) clone(ctx context.Context, waitTimeout time.Duration) error {
+	if r.TryStartCloning() {
+		return r.CloneClaimed(ctx)
+	}
+	if waitTimeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, waitTimeout)
+		defer cancel()
+	}
+
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		switch r.State() {
+		case StateReady:
+			return nil
+		case StateEmpty:
+			return errors.New("repository clone did not complete")
+		case StateCloning:
+		}
+		select {
+		case <-ctx.Done():
+			return errors.Wrap(ctx.Err(), "wait for repository clone")
+		case <-ticker.C:
+		}
+	}
+}
+
+// CloneClaimed requires the caller to acquire ownership with TryStartCloning first.
+func (r *Repository) CloneClaimed(ctx context.Context) error {
 	r.mu.Lock()
 	if r.state == StateReady {
 		r.mu.Unlock()
 		return nil
 	}
-	r.state = StateCloning
+	if r.state != StateCloning {
+		r.mu.Unlock()
+		return errors.New("repository clone was not claimed")
+	}
 	r.mu.Unlock()
 
 	err := r.executeClone(ctx)

internal/gitclone/manager_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/alecthomas/assert/v2"
+	"github.com/alecthomas/errors"
 
 	"github.com/block/cachew/internal/logging"
 )
@@ -415,6 +416,41 @@ func TestRepository_Clone_StateVisibleDuringClone(t *testing.T) {
 	assert.Equal(t, StateReady, repo.State())
 }
 
+func TestRepositoryCloneWaitsForCurrentOwner(t *testing.T) {
+	repo := &Repository{
+		state:       StateCloning,
+		config:      testRepoConfig(),
+		path:        filepath.Join(t.TempDir(), "clone"),
+		upstreamURL: "https://example.test/example/repo",
+		fetchSem:    make(chan struct{}, 1),
+	}
+	repo.fetchSem <- struct{}{}
+	ready := make(chan struct{})
+	go func() {
+		time.Sleep(20 * time.Millisecond)
+		repo.MarkReady()
+		close(ready)
+	}()
+
+	assert.NoError(t, repo.Clone(t.Context()))
+	<-ready
+	assert.Equal(t, StateReady, repo.State())
+}
+
+func TestRepositoryCloneWaitTimeout(t *testing.T) {
+	repo := &Repository{
+		state:       StateCloning,
+		config:      testRepoConfig(),
+		path:        filepath.Join(t.TempDir(), "clone"),
+		upstreamURL: "https://example.test/example/repo",
+		fetchSem:    make(chan struct{}, 1),
+	}
+	repo.fetchSem <- struct{}{}
+
+	err := repo.CloneWithWaitTimeout(t.Context(), 20*time.Millisecond)
+	assert.True(t, errors.Is(err, context.DeadlineExceeded))
+}
+
 func TestRepository_CloneSetsMirrorConfig(t *testing.T) {
 	ctx := context.Background()
 	tmpDir := t.TempDir()

internal/jobscheduler/jobs.go

@@ -48,6 +48,8 @@ type Scheduler interface {
 	//
 	// Jobs run concurrently across queues, but never within a queue.
 	Submit(queue, id string, run func(ctx context.Context) error)
+	// TrySubmit adds a job only if fewer than maxQueued jobs wait across all queues.
+	TrySubmit(queue, id string, maxQueued int, run func(ctx context.Context) error) bool
 	// SubmitPeriodicJob submits a job to the queue that runs immediately, and then periodically after the interval.
 	//
 	// Jobs run concurrently across queues, but never within a queue.
@@ -63,6 +65,10 @@ func (p *prefixedScheduler) Submit(queue, id string, run func(ctx context.Contex
 	p.scheduler.Submit(queue, p.prefix+id, run)
 }
 
+func (p *prefixedScheduler) TrySubmit(queue, id string, maxQueued int, run func(ctx context.Context) error) bool {
+	return p.scheduler.TrySubmit(queue, p.prefix+id, maxQueued, run)
+}
+
 func (p *prefixedScheduler) SubmitPeriodicJob(queue, id string, interval time.Duration, run func(ctx context.Context) error) {
 	p.scheduler.SubmitPeriodicJob(queue, p.prefix+id, interval, run)
 }
@@ -190,6 +196,20 @@ func (q *RootScheduler) Submit(queue, id string, run func(ctx context.Context) e
 	q.cond.Signal()
 }
 
+// TrySubmit limits the queue size without a wait. It rejects new jobs during shutdown.
+func (q *RootScheduler) TrySubmit(queue, id string, maxQueued int, run func(ctx context.Context) error) bool {
+	q.lock.Lock()
+	if q.done || q.draining || maxQueued <= 0 || len(q.queue) >= maxQueued {
+		q.lock.Unlock()
+		return false
+	}
+	q.queue = append(q.queue, queueJob{queue: queue, id: id, run: run})
+	q.metrics.queueDepth.Record(context.Background(), int64(len(q.queue)))
+	q.lock.Unlock()
+	q.cond.Signal()
+	return true
+}
+
 func (q *RootScheduler) SubmitPeriodicJob(queue, id string, interval time.Duration, run func(ctx context.Context) error) {
 	if q.ctx.Err() != nil || q.isDraining() {
 		return

internal/s3client/s3clienttest/s3clienttest.go

@@ -134,7 +134,7 @@ func startContainer(t *testing.T) {
 			"-p", Port+":9000",
 			"-e", "MINIO_ROOT_USER="+Username,
 			"-e", "MINIO_ROOT_PASSWORD="+Password,
-			"minio/minio", "server", "/data",
+			"quay.io/minio/minio", "server", "/data",
 		)
 		output, err := cmd.CombinedOutput()
 		if err == nil {

internal/strategy/git/clone_ownership_test.go

@@ -0,0 +1,58 @@
+package git //nolint:testpackage // This test needs access to the spool mutex.
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/alecthomas/assert/v2"
+
+	"github.com/block/cachew/internal/cache"
+	"github.com/block/cachew/internal/gitclone"
+	"github.com/block/cachew/internal/logging"
+)
+
+func TestCloneFailurePreservesSuccessorOwnership(t *testing.T) {
+	root := t.TempDir()
+	started := filepath.Join(root, "started")
+	assert.NoError(t, os.WriteFile(filepath.Join(root, "git"), []byte("#!/bin/sh\ntouch \"$CLONE_STARTED\"\nexit 42\n"), 0o750))
+	t.Setenv("PATH", root+string(os.PathListSeparator)+os.Getenv("PATH"))
+	t.Setenv("CLONE_STARTED", started)
+	ctx := logging.ContextWithLogger(context.Background(), slog.Default())
+	manager, err := gitclone.NewManager(ctx, gitclone.Config{MirrorRoot: filepath.Join(root, "mirrors")}, nil)
+	assert.NoError(t, err)
+	memCache, err := cache.NewMemory(ctx, cache.MemoryConfig{MaxTTL: time.Hour})
+	assert.NoError(t, err)
+	repo, err := manager.GetOrCreate(ctx, "https://example.test/example/repo")
+	assert.NoError(t, err)
+	s := &Strategy{cache: memCache, cloneManager: manager, metrics: newGitMetrics()}
+
+	s.spoolsMu.Lock()
+	locked := true
+	t.Cleanup(func() {
+		if locked {
+			s.spoolsMu.Unlock()
+		}
+	})
+	done := make(chan error, 1)
+	go func() { done <- s.startClone(ctx, repo) }()
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		_, err := os.Stat(started)
+		if err == nil && repo.State() == gitclone.StateEmpty {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("clone did not fail before spool cleanup")
+		}
+		time.Sleep(time.Millisecond)
+	}
+	assert.True(t, repo.TryStartCloning())
+	s.spoolsMu.Unlock()
+	locked = false
+	assert.Error(t, <-done)
+	assert.Equal(t, gitclone.StateCloning, repo.State())
+}