Clone mirrors aside and clean up interrupted clones

pkg/app/app.go

@@ -275,6 +275,9 @@ func (a *App) initRepos(ctx context.Context) error {
 		ReposDir:     a.cfg.Storage.ReposDir,
 		WorktreesDir: a.cfg.Storage.WorktreesDir,
 		MaxDiskBytes: a.cfg.Storage.MaxDiskBytes,
+		OnRemove: func(what, path string, size int64) {
+			a.Print(ctx, "storage removed", "what", what, "path", path, "bytes", size)
+		},
 	})
 	if a.cfg.Storage.ReposDir == "" {
 		a.Print(ctx, "code tools disabled: Storage.ReposDir is empty")

pkg/app/mirrors.go

@@ -3,11 +3,13 @@ package app
 import (
 	"context"
 	"crypto/subtle"
+	"errors"
 	"io"
 	"net/http"
 	"sort"
 	"time"
 
+	"github.com/vmkteam/ringsrv/pkg/client/git"
 	"github.com/vmkteam/ringsrv/pkg/client/gitlab"
 
 	"github.com/labstack/echo/v4"
@@ -148,15 +150,23 @@ func (a *App) fetchMirror(ctx context.Context, name string) {
 	ctx, cancel := context.WithTimeout(ctx, webhookTimeout)
 	defer cancel()
 
-	if err := a.repos.Ensure(ctx, repo.Spec(name)); err != nil {
+	err := a.repos.Ensure(ctx, repo.Spec(name))
+	switch {
+	case errors.Is(err, git.ErrCloning):
+		// Not a failure: the first clone of a large repository outlives the
+		// window, and the next tick finds the mirror in place.
+		a.Print(ctx, "mirror still cloning", "repo", name)
+		return
+	case err != nil:
 		a.Error(ctx, "mirror fetch failed", "repo", name, "err", err.Error())
 		return
 	}
 	a.Print(ctx, "mirror fetched", "repo", name)
 }
 
-// pruneStorage evicts the worktrees nobody is reading once the disk budget is
-// exceeded. Their AST indexes live inside them, so both go together.
+// pruneStorage removes what interrupted clones and fetches left in the mirrors,
+// then evicts the worktrees nobody is reading once the disk budget is exceeded.
+// Their AST indexes live inside them, so both go together.
 func (a *App) pruneStorage(ctx context.Context) error {
 	freed, err := a.repos.Prune(ctx)
 	if err != nil {

pkg/client/codegraph/client.go

@@ -18,6 +18,8 @@ import (
 	"regexp"
 	"strings"
 	"time"
+
+	"github.com/vmkteam/ringsrv/pkg/client/proc"
 )
 
 // Errors the domain distinguishes.
@@ -92,7 +94,9 @@ func (c *Client) Version(ctx context.Context) (string, error) {
 	ctx, cancel := context.WithTimeout(ctx, c.opts.Timeout)
 	defer cancel()
 
-	out, err := exec.CommandContext(ctx, c.opts.Bin, "--version").CombinedOutput() //nolint:gosec // fixed argv
+	cmd := exec.CommandContext(ctx, c.opts.Bin, "--version") //nolint:gosec // fixed argv
+	proc.KillGroupOnCancel(cmd)
+	out, err := cmd.CombinedOutput()
 	if err != nil {
 		return "", fmt.Errorf("codegraph: %s --version: %w: %s", c.opts.Bin, err, bytes.TrimSpace(out))
 	}
@@ -163,6 +167,7 @@ func (c *Client) call(ctx context.Context, dir, tool string, args map[string]any
 func (c *Client) exec(ctx context.Context, dir, tool string, req []byte, dst any) error {
 	cmd := exec.CommandContext(ctx, c.opts.Bin, "serve") //nolint:gosec // fixed argv; arguments travel as JSON on stdin
 	cmd.Dir = dir
+	proc.KillGroupOnCancel(cmd)
 	cmd.Stdin = bytes.NewReader(req)
 	var stdout, stderr bytes.Buffer
 	cmd.Stdout, cmd.Stderr = &stdout, &stderr

pkg/client/codegraph/index.go

@@ -9,6 +9,8 @@ import (
 	"path/filepath"
 	"time"
 
+	"github.com/vmkteam/ringsrv/pkg/client/proc"
+
 	"golang.org/x/sync/singleflight"
 )
 
@@ -70,6 +72,7 @@ func (c *Client) Finalize(ctx context.Context, dir string) error {
 
 	cmd := exec.CommandContext(ctx, c.opts.Bin, "incremental-index") //nolint:gosec // fixed argv
 	cmd.Dir = dir
+	proc.KillGroupOnCancel(cmd)
 	var out bytes.Buffer
 	cmd.Stdout, cmd.Stderr = &out, &out
 
@@ -88,6 +91,7 @@ func (c *Client) index(ctx context.Context, dir string) error {
 
 	cmd := exec.CommandContext(ctx, c.opts.Bin, "rebuild-index", "--confirm") //nolint:gosec // fixed argv
 	cmd.Dir = dir
+	proc.KillGroupOnCancel(cmd)
 	var out bytes.Buffer
 	cmd.Stdout, cmd.Stderr = &out, &out
 

pkg/client/git/git.go

@@ -19,6 +19,8 @@ import (
 	"regexp"
 	"strings"
 	"time"
+
+	"github.com/vmkteam/ringsrv/pkg/client/proc"
 )
 
 // allowedSubcommands is the whole surface of git this service uses. Anything
@@ -50,6 +52,10 @@ var (
 	// ErrBadRef and ErrBadPath are refusals to build a command at all.
 	ErrBadRef  = errors.New("git: bad ref")
 	ErrBadPath = errors.New("git: bad path")
+	// ErrCloning means the caller stopped waiting while the first clone goes
+	// on: for a large repository it outlives a cron tick and a question, and a
+	// bare "deadline exceeded" read as a failed mirror.
+	ErrCloning = errors.New("git: first clone still in progress")
 )
 
 // shaRe is what a ref may look like when it is not a known branch name.
@@ -106,15 +112,24 @@ func (s *Store) run(ctx context.Context, dir string, args ...string) ([]byte, er
 // runEnv is run with extra environment variables — the way a clone token
 // reaches git without touching argv or the clone's config (see authEnv).
 func (s *Store) runEnv(ctx context.Context, dir string, env []string, args ...string) ([]byte, error) {
+	return s.runEnvTimeout(ctx, s.timeout(), dir, env, args...)
+}
+
+// runEnvTimeout is runEnv with its own deadline, for the one command that
+// honestly takes longer than DefaultTimeout: the first clone.
+func (s *Store) runEnvTimeout(ctx context.Context, timeout time.Duration, dir string, env []string, args ...string) ([]byte, error) {
 	if len(args) == 0 || !allowedSubcommands[args[0]] {
 		return nil, fmt.Errorf("git: subcommand %q is not allowed", first(args))
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, s.timeout())
+	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
 	cmd := exec.CommandContext(ctx, s.opts.Git, args...) //nolint:gosec // argv is built here, never from a caller
 	cmd.Dir = dir
+	// clone and fetch are a tree of processes — remote-https runs index-pack —
+	// and a helper left alive goes on writing a pack nobody will use.
+	proc.KillGroupOnCancel(cmd)
 	cmd.Env = append(os.Environ(),
 		"GIT_TERMINAL_PROMPT=0",
 		"GIT_CONFIG_NOSYSTEM=1",
@@ -127,7 +142,7 @@ func (s *Store) runEnv(ctx context.Context, dir string, env []string, args ...st
 
 	if err := cmd.Run(); err != nil {
 		if ctx.Err() != nil {
-			return nil, fmt.Errorf("git: git %s timed out after %s", args[0], s.timeout())
+			return nil, fmt.Errorf("git: git %s timed out after %s", args[0], timeout)
 		}
 		return nil, fmt.Errorf("git: git %s: %w: %s", args[0], err, clip(stderr.String()))
 	}
@@ -141,6 +156,13 @@ func (s *Store) timeout() time.Duration {
 	return DefaultTimeout
 }
 
+func (s *Store) cloneTimeout() time.Duration {
+	if s.opts.CloneTimeout > 0 {
+		return s.opts.CloneTimeout
+	}
+	return DefaultCloneTimeout
+}
+
 func first(args []string) string {
 	if len(args) == 0 {
 		return ""

pkg/client/git/git_test.go

@@ -11,6 +11,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/vmkteam/ringsrv/pkg/ring/code/gittest"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -597,3 +599,153 @@ func (s *Store) busy2(key string) int {
 	defer s.mu.Unlock()
 	return s.inUse[key]
 }
+
+// A clone killed mid-transfer by an earlier version, which cloned in place,
+// left packs and no HEAD at the mirror's path. git refuses to clone into a
+// directory that is not empty, so the mirror stayed broken — for days on one
+// instance, with 65 GB under it. Ensure clears it and clones.
+func TestEnsure_ReplacesWhatAKilledCloneLeft(t *testing.T) {
+	t.Parallel()
+	origin, _, second := originRepo(t)
+	s := newStore(t)
+	gittest.Write(t, s.ClonePath("apisrv"), "objects/pack/tmp_pack_x", "half a pack")
+	require.False(t, s.Have("apisrv"))
+	removed := recordRemovals(s)
+
+	require.NoError(t, s.Ensure(t.Context(), Spec{Name: "apisrv", CloneURL: origin, DefaultBranch: "master"}))
+
+	assert.NoFileExists(t, filepath.Join(s.ClonePath("apisrv"), "objects", "pack", "tmp_pack_x"))
+	assert.Equal(t, []removal{{"broken mirror", s.ClonePath("apisrv"), int64(len("half a pack"))}}, *removed,
+		"a disk that frees space by itself says what went")
+	sha, err := s.ResolveSHA(t.Context(), "apisrv", "master")
+	require.NoError(t, err)
+	assert.Equal(t, second, sha)
+}
+
+// A clone that fails leaves nothing behind: neither a mirror that looks
+// present nor the directory it was being made in.
+func TestEnsure_FailedCloneLeavesNothing(t *testing.T) {
+	t.Parallel()
+	s := newStore(t)
+
+	err := s.Ensure(t.Context(), Spec{Name: "apisrv", CloneURL: filepath.Join(t.TempDir(), "missing"), DefaultBranch: "master"})
+	require.Error(t, err)
+
+	assert.False(t, s.Have("apisrv"))
+	entries, err := os.ReadDir(s.opts.ReposDir)
+	require.NoError(t, err)
+	assert.Empty(t, entries)
+}
+
+// A clone killed after HEAD but before the refspec has no fetch line in its
+// config. Fetches then write FETCH_HEAD only, refs/remotes/origin/* never
+// move, and a branch name resolves to the clone-day snapshot forever. Ensure
+// writes the refspec back.
+func TestEnsure_RestoresMissingRefspec(t *testing.T) {
+	t.Parallel()
+	origin, _, _ := originRepo(t)
+	s := newStore(t)
+	spec := Spec{Name: "apisrv", CloneURL: origin, DefaultBranch: "master"}
+	require.NoError(t, s.Ensure(t.Context(), spec))
+	gitT(t, s.ClonePath("apisrv"), "config", "--unset-all", "remote.origin.fetch")
+
+	gitT(t, origin, "commit", "--allow-empty", "-m", "ABC-3 after the clone")
+	third := gitT(t, origin, "rev-parse", "HEAD")
+
+	require.NoError(t, s.Ensure(t.Context(), spec))
+	sha, err := s.ResolveSHA(t.Context(), "apisrv", "master")
+	require.NoError(t, err)
+	assert.Equal(t, third, sha, "master follows the remote again")
+}
+
+// A caller that gives up does not stop its clone, so the cron's limit bounds
+// waiting, not cloning: on an empty volume every mirror would start cloning
+// within one tick, all in the service's memory. Clones queue for a slot.
+func TestEnsure_ClonesWaitForASlot(t *testing.T) {
+	t.Parallel()
+	origin, _, _ := originRepo(t)
+	s := newStore(t)
+	for range cap(s.cloneSlots) {
+		s.cloneSlots <- struct{}{} // every slot is taken
+	}
+	spec := Spec{Name: "apisrv", CloneURL: origin, DefaultBranch: "master"}
+
+	ctx, cancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
+	defer cancel()
+	err := s.Ensure(ctx, spec)
+	require.ErrorIs(t, err, ErrCloning, "the caller left, the clone did not")
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	assert.False(t, s.Have("apisrv"), "no clone runs without a slot")
+	entries, err := os.ReadDir(s.opts.ReposDir)
+	if !os.IsNotExist(err) {
+		require.NoError(t, err)
+		assert.Empty(t, entries, "not even its directory")
+	}
+
+	<-s.cloneSlots // a slot frees up, and the clone that waited for it runs
+	require.NoError(t, s.Ensure(t.Context(), spec))
+	assert.True(t, s.Have("apisrv"))
+}
+
+// A clone or a fetch killed together with the process — a restart, the OOM
+// killer — leaves its clone directory or tmp_* pack behind, and git removes
+// those only in gc, after two weeks. Prune does, budget or not, once they are
+// older than anything still running could be.
+func TestPrune_RemovesAbandonedLeftovers(t *testing.T) {
+	t.Parallel()
+	origin, _, _ := originRepo(t)
+	s := newStore(t) // MaxDiskBytes 0: no budget, leftovers go anyway
+	require.NoError(t, s.Ensure(t.Context(), Spec{Name: "apisrv", CloneURL: origin, DefaultBranch: "master"}))
+
+	abandoned := s.abandonedBefore().Add(-time.Minute)
+	plant := func(dir, name string, at time.Time) {
+		gittest.Write(t, dir, name, "part of a pack")
+		require.NoError(t, os.Chtimes(filepath.Join(dir, name), at, at))
+	}
+	oldClone := filepath.Join(s.opts.ReposDir, cloneTmpPrefix+"apisrv-1")
+	plant(oldClone, "objects/pack/tmp_pack_a", abandoned)
+	require.NoError(t, os.Chtimes(oldClone, abandoned, abandoned))
+	runningClone := filepath.Join(s.opts.ReposDir, cloneTmpPrefix+"apisrv-2")
+	plant(runningClone, "HEAD", time.Now())
+
+	packs := filepath.Join(s.ClonePath("apisrv"), "objects", "pack")
+	plant(packs, "tmp_pack_old", abandoned)
+	plant(packs, "tmp_pack_running", time.Now())
+	removed := recordRemovals(s)
+
+	freed, err := s.Prune(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, int64(2*len("part of a pack")), freed)
+
+	assert.NoDirExists(t, oldClone)
+	assert.NoFileExists(t, filepath.Join(packs, "tmp_pack_old"))
+	assert.DirExists(t, runningClone, "a clone in progress is left alone")
+	assert.FileExists(t, filepath.Join(packs, "tmp_pack_running"), "a fetch in progress is left alone")
+	assert.ElementsMatch(t, []removal{
+		{"abandoned clone", oldClone, int64(len("part of a pack"))},
+		{"abandoned pack", filepath.Join(packs, "tmp_pack_old"), int64(len("part of a pack"))},
+	}, *removed)
+
+	_, err = s.ResolveSHA(t.Context(), "apisrv", "master")
+	require.NoError(t, err, "the mirror itself is untouched")
+}
+
+// removal is one OnRemove report.
+type removal struct {
+	what, path string
+	size       int64
+}
+
+// recordRemovals collects what the store reports removing. The reports come
+// from the goroutine doing the work, and the test reads them after that work
+// returned.
+func recordRemovals(s *Store) *[]removal {
+	var mu sync.Mutex
+	got := []removal{}
+	s.opts.OnRemove = func(what, path string, size int64) {
+		mu.Lock()
+		defer mu.Unlock()
+		got = append(got, removal{what, path, size})
+	}
+	return &got
+}