fix(gitcli): bound the one-shot git processes and the open blob streams

One-shot commands (run, pipe) take a slot of a semaphore sized
max(4, 2*GOMAXPROCS) (Config.MaxExec) and wait for it until their context
ends. Blob streams are counted separately (Config.MaxBlobStreams, 32) from
process start until the process is collected, and a full table is refused
with resource_exhausted instead of queued, since a slow reader may hold a
slot for long. The pool idle test drives the idle timers through an
export_test hook instead of sleeping.

commit: 147824db7e3dd7f45a07369e0df75f8f7a1af88d
author: brice · 12d ago · 2026-10-10 11:14 UTC
parent: 1dc333e
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +189 -13

M internal/adapters/gitcli/blob.go +19 -3


@@ -147,13 +147,25 @@ type blobStream struct {
147	147		read int64
148	148		stop func() bool // detaches the ctx watcher
149	149
150		-	done chan struct{} // closed when the process has been collected
151		-	once sync.Once
	150	+	done chan struct{} // closed when the process has been collected
	151	+	once sync.Once
	152	+	release func() // gives the stream slot back
152	153		}
153	154
	155	+	// startBlobStream starts the process of a blob stream. The process is counted
	156	+	// against Config.MaxBlobStreams from its start until the stream is closed (or
	157	+	// the process is collected); there is no queue: a full table is refused with
	158	+	// domain.ErrResourceExhausted.
154	159		func (g Git) startBlobStream(ctx context.Context, dir string, id domain.ObjectID, size int64) (blobStream, error) {
	160	+	select {
	161	+	case g.blobs <- struct{}{}:
	162	+	default:
	163	+	return nil, fmt.Errorf("too many blob streams open: %w", domain.ErrResourceExhausted)
	164	+	}
	165	+	release := func() { <-g.blobs }
155	166		pr, pw, err := os.Pipe()
156	167		if err != nil {
	168	+	release()
157	169		return nil, fmt.Errorf("start git cat-file: %w", err)
158	170		}
159	171		cmd := exec.Command(g.bin, readArgs(dir, "cat-file", "blob", id.String())...)

@@ -163,12 +175,13 @@ func (g *Git) startBlobStream(ctx context.Context, dir string, id domain.ObjectI
163	175		if err := cmd.Start(); err != nil {
164	176		_ = pr.Close()
165	177		_ = pw.Close()
	178	+	release()
166	179		return nil, fmt.Errorf("start git cat-file: %w", err)
167	180		}
168	181		_ = pw.Close()
169	182		s := &blobStream{
170	183		pool: g.cat, cmd: cmd, pipe: pr, r: bufio.NewReaderSize(pr, 64*1024),
171		-	size: size, done: make(chan struct{}),
	184	+	size: size, done: make(chan struct{}), release: release,
172	185		}
173	186		go s.reap()
174	187		if err := g.cat.addStream(s); err != nil {

@@ -179,8 +192,11 @@ func (g *Git) startBlobStream(ctx context.Context, dir string, id domain.ObjectI
179	192		return s, nil
180	193		}
181	194
	195	+	// reap collects the process and, with it, the stream slot: a process that is
	196	+	// gone no longer counts, whether or not the reader was closed.
182	197		func (s *blobStream) reap() {
183	198		_ = s.cmd.Wait()
	199	+	s.release()
184	200		close(s.done)
185	201		}
186	202

 	read int64
 	stop func() bool // detaches the ctx watcher
-	done chan struct{} // closed when the process has been collected
+	done    chan struct{} // closed when the process has been collected
-	once sync.Once
+	once    sync.Once
+	release func() // gives the stream slot back
 }
+// startBlobStream starts the process of a blob stream. The process is counted
+// against Config.MaxBlobStreams from its start until the stream is closed (or
+// the process is collected); there is no queue: a full table is refused with
+// domain.ErrResourceExhausted.
 func (g *Git) startBlobStream(ctx context.Context, dir string, id domain.ObjectID, size int64) (*blobStream, error) {
+	select {
+	case g.blobs <- struct{}{}:
+	default:
+		return nil, fmt.Errorf("too many blob streams open: %w", domain.ErrResourceExhausted)
+	}
+	release := func() { <-g.blobs }
 	pr, pw, err := os.Pipe()
 	if err != nil {
+		release()
 		return nil, fmt.Errorf("start git cat-file: %w", err)
 	}
 	cmd := exec.Command(g.bin, readArgs(dir, "cat-file", "blob", id.String())...)
 	if err := cmd.Start(); err != nil {
 		_ = pr.Close()
 		_ = pw.Close()
+		release()
 		return nil, fmt.Errorf("start git cat-file: %w", err)
 	}
 	_ = pw.Close()
 	s := &blobStream{
 		pool: g.cat, cmd: cmd, pipe: pr, r: bufio.NewReaderSize(pr, 64*1024),
-		size: size, done: make(chan struct{}),
+		size: size, done: make(chan struct{}), release: release,
 	}
 	go s.reap()
 	if err := g.cat.addStream(s); err != nil {
 	return s, nil
 }
+// reap collects the process and, with it, the stream slot: a process that is
+// gone no longer counts, whether or not the reader was closed.
 func (s *blobStream) reap() {
 	_ = s.cmd.Wait()
+	s.release()
 	close(s.done)
 }

M internal/adapters/gitcli/export_test.go +40 -0


@@ -34,3 +34,43 @@ func (g *Git) BlobStreamPIDs() []int {
34	34		}
35	35		return pids
36	36		}
	37	+
	38	+	// FireIdleTimers runs, for every idle cat-file process, what its idle timer
	39	+	// would do now, so that tests do not wait for real time.
	40	+	func (g *Git) FireIdleTimers() {
	41	+	g.cat.mu.Lock()
	42	+	type armed struct {
	43	+	cp *catProc
	44	+	gen uint64
	45	+	}
	46	+	var all []armed
	47	+	for _, cp := range g.cat.procs {
	48	+	all = append(all, armed{cp, cp.gen})
	49	+	}
	50	+	g.cat.mu.Unlock()
	51	+	for _, a := range all {
	52	+	g.cat.expire(a.cp, a.gen)
	53	+	}
	54	+	}
	55	+
	56	+	// FireStaleIdleTimers runs what the idle timers armed before the last use of
	57	+	// each process would do: nothing, since the process was used since.
	58	+	func (g *Git) FireStaleIdleTimers() {
	59	+	g.cat.mu.Lock()
	60	+	type armed struct {
	61	+	cp *catProc
	62	+	gen uint64
	63	+	}
	64	+	var all []armed
	65	+	for _, cp := range g.cat.procs {
	66	+	all = append(all, armed{cp, cp.gen - 1})
	67	+	}
	68	+	g.cat.mu.Unlock()
	69	+	for _, a := range all {
	70	+	g.cat.expire(a.cp, a.gen)
	71	+	}
	72	+	}
	73	+
	74	+	// ExecSlots returns the size of the one-shot process semaphore and how many
	75	+	// slots are taken.
	76	+	func (g *Git) ExecSlots() (size, taken int) { return cap(g.exec), len(g.exec) }


@@ -34,3 +34,43 @@ func (g *Git) BlobStreamPIDs() []int {
34	}	34		}
35	return pids	35		return pids
36	}	36		}
		37	+
		38	+	// FireIdleTimers runs, for every idle cat-file process, what its idle timer
		39	+	// would do now, so that tests do not wait for real time.
		40	+	func (g *Git) FireIdleTimers() {
		41	+	g.cat.mu.Lock()
		42	+	type armed struct {
		43	+	cp *catProc
		44	+	gen uint64
		45	+	}
		46	+	var all []armed
		47	+	for _, cp := range g.cat.procs {
		48	+	all = append(all, armed{cp, cp.gen})
		49	+	}
		50	+	g.cat.mu.Unlock()
		51	+	for _, a := range all {
		52	+	g.cat.expire(a.cp, a.gen)
		53	+	}
		54	+	}
		55	+
		56	+	// FireStaleIdleTimers runs what the idle timers armed before the last use of
		57	+	// each process would do: nothing, since the process was used since.
		58	+	func (g *Git) FireStaleIdleTimers() {
		59	+	g.cat.mu.Lock()
		60	+	type armed struct {
		61	+	cp *catProc
		62	+	gen uint64
		63	+	}
		64	+	var all []armed
		65	+	for _, cp := range g.cat.procs {
		66	+	all = append(all, armed{cp, cp.gen - 1})
		67	+	}
		68	+	g.cat.mu.Unlock()
		69	+	for _, a := range all {
		70	+	g.cat.expire(a.cp, a.gen)
		71	+	}
		72	+	}
		73	+
		74	+	// ExecSlots returns the size of the one-shot process semaphore and how many
		75	+	// slots are taken.
		76	+	func (g *Git) ExecSlots() (size, taken int) { return cap(g.exec), len(g.exec) }

M internal/adapters/gitcli/git.go +46 -1


@@ -12,6 +12,7 @@ import (
12	12		"fmt"
13	13		"os/exec"
14	14		"path/filepath"
	15	+	"runtime"
15	16		"strings"
16	17		"sync"
17	18		"time"

@@ -36,6 +37,16 @@ type Config struct {
36	37		CatFileIdle time.Duration
37	38		// MaxCatFiles bounds the long-lived cat-file processes. Zero means 16.
38	39		MaxCatFiles int
	40	+	// MaxExec bounds the one-shot git processes that run at once (every
	41	+	// command but the long-lived cat-file processes and the blob streams);
	42	+	// callers beyond it wait for a slot until their context ends. Zero means
	43	+	// max(4, 2*GOMAXPROCS).
	44	+	MaxExec int
	45	+	// MaxBlobStreams bounds the open blob streams, each of which holds a git
	46	+	// process until its reader is closed. Streams beyond it are refused with
	47	+	// domain.ErrResourceExhausted rather than queued: a slow reader could hold
	48	+	// a slot for long. Zero means 32.
	49	+	MaxBlobStreams int
39	50		}
40	51
41	52		// Git is the git CLI adapter for ports.GitRepos.

@@ -49,14 +60,31 @@ type Git struct {
49	60		backend string // git-http-backend, resolved on first use
50	61
51	62		cat *catPool // long-lived cat-file processes of the read side
	63	+
	64	+	exec chan struct{} // slots of the one-shot processes, see Config.MaxExec
	65	+	blobs chan struct{} // slots of the blob streams, see Config.MaxBlobStreams
52	66		}
53	67
	68	+	// Defaults of the process bounds.
	69	+	const defaultMaxBlobStreams = 32
	70	+
	71	+	func defaultMaxExec() int { return max(4, 2*runtime.GOMAXPROCS(0)) }
	72	+
54	73		// New builds an adapter. It does not run git: call CheckVersion.
55	74		func New(cfg Config) *Git {
56	75		env := buildEnv(cfg.DataDir, cfg.Path)
	76	+	maxExec, maxBlobs := cfg.MaxExec, cfg.MaxBlobStreams
	77	+	if maxExec <= 0 {
	78	+	maxExec = defaultMaxExec()
	79	+	}
	80	+	if maxBlobs <= 0 {
	81	+	maxBlobs = defaultMaxBlobStreams
	82	+	}
57	83		return &Git{
58	84		bin: cfg.Bin, dataDir: cfg.DataDir, env: env, maxRecv: cfg.MaxReceiveBytes,
59		-	cat: newCatPool(cfg.Bin, env, cfg.CatFileIdle, cfg.MaxCatFiles),
	85	+	cat: newCatPool(cfg.Bin, env, cfg.CatFileIdle, cfg.MaxCatFiles),
	86	+	exec: make(chan struct{}, maxExec),
	87	+	blobs: make(chan struct{}, maxBlobs),
60	88		}
61	89		}
62	90

@@ -80,6 +108,18 @@ func buildEnv(dataDir, path string) []string {
80	108		// commandTimeout bounds one git invocation when ctx has no deadline.
81	109		const commandTimeout = 2 * time.Minute
82	110
	111	+	// acquireExec takes a slot of the one-shot processes, waiting until ctx ends.
	112	+	// The wait counts against the deadline of ctx, so a saturated server answers a
	113	+	// request with its deadline rather than hanging it.
	114	+	func (g *Git) acquireExec(ctx context.Context) (release func(), err error) {
	115	+	select {
	116	+	case g.exec <- struct{}{}:
	117	+	return func() { <-g.exec }, nil
	118	+	case <-ctx.Done():
	119	+	return nil, ctx.Err()
	120	+	}
	121	+	}
	122	+
83	123		// run executes git with args in the isolated environment and returns stdout.
84	124		func (g *Git) run(ctx context.Context, args ...string) (string, error) {
85	125		if _, ok := ctx.Deadline(); !ok {

@@ -87,6 +127,11 @@ func (g *Git) run(ctx context.Context, args ...string) (string, error) {
87	127		ctx, cancel = context.WithTimeout(ctx, commandTimeout)
88	128		defer cancel()
89	129		}
	130	+	release, err := g.acquireExec(ctx)
	131	+	if err != nil {
	132	+	return "", fmt.Errorf("git %s: %w", subcommand(args), err)
	133	+	}
	134	+	defer release()
90	135		cmd := exec.CommandContext(ctx, g.bin, args...)
91	136		cmd.Env = g.env
92	137		cmd.Dir = g.dataDir


@@ -12,6 +12,7 @@ import (
12		"fmt"	12		"fmt"
13		"os/exec"	13		"os/exec"
14		"path/filepath"	14		"path/filepath"
			15	+	"runtime"
15		"strings"	16		"strings"
16		"sync"	17		"sync"
17		"time"	18		"time"

@@ -36,6 +37,16 @@ type Config struct {
36		CatFileIdle time.Duration	37		CatFileIdle time.Duration
37		// MaxCatFiles bounds the long-lived cat-file processes. Zero means 16.	38		// MaxCatFiles bounds the long-lived cat-file processes. Zero means 16.
38		MaxCatFiles int	39		MaxCatFiles int
			40	+	// MaxExec bounds the one-shot git processes that run at once (every
			41	+	// command but the long-lived cat-file processes and the blob streams);
			42	+	// callers beyond it wait for a slot until their context ends. Zero means
			43	+	// max(4, 2*GOMAXPROCS).
			44	+	MaxExec int
			45	+	// MaxBlobStreams bounds the open blob streams, each of which holds a git
			46	+	// process until its reader is closed. Streams beyond it are refused with
			47	+	// domain.ErrResourceExhausted rather than queued: a slow reader could hold
			48	+	// a slot for long. Zero means 32.
			49	+	MaxBlobStreams int
39		}	50		}
40			51
41		// Git is the git CLI adapter for ports.GitRepos.	52		// Git is the git CLI adapter for ports.GitRepos.

@@ -49,14 +60,31 @@ type Git struct {
49		backend string // git-http-backend, resolved on first use	60		backend string // git-http-backend, resolved on first use
50			61
51		cat *catPool // long-lived cat-file processes of the read side	62		cat *catPool // long-lived cat-file processes of the read side
			63	+
			64	+	exec chan struct{} // slots of the one-shot processes, see Config.MaxExec
			65	+	blobs chan struct{} // slots of the blob streams, see Config.MaxBlobStreams
52		}	66		}
53			67
			68	+	// Defaults of the process bounds.
			69	+	const defaultMaxBlobStreams = 32
			70	+
			71	+	func defaultMaxExec() int { return max(4, 2*runtime.GOMAXPROCS(0)) }
			72	+
54		// New builds an adapter. It does not run git: call CheckVersion.	73		// New builds an adapter. It does not run git: call CheckVersion.
55		func New(cfg Config) *Git {	74		func New(cfg Config) *Git {
56		env := buildEnv(cfg.DataDir, cfg.Path)	75		env := buildEnv(cfg.DataDir, cfg.Path)
			76	+	maxExec, maxBlobs := cfg.MaxExec, cfg.MaxBlobStreams
			77	+	if maxExec <= 0 {
			78	+	maxExec = defaultMaxExec()
			79	+	}
			80	+	if maxBlobs <= 0 {
			81	+	maxBlobs = defaultMaxBlobStreams
			82	+	}
57		return &Git{	83		return &Git{
58		bin: cfg.Bin, dataDir: cfg.DataDir, env: env, maxRecv: cfg.MaxReceiveBytes,	84		bin: cfg.Bin, dataDir: cfg.DataDir, env: env, maxRecv: cfg.MaxReceiveBytes,
59	-	cat: newCatPool(cfg.Bin, env, cfg.CatFileIdle, cfg.MaxCatFiles),	85	+	cat: newCatPool(cfg.Bin, env, cfg.CatFileIdle, cfg.MaxCatFiles),
			86	+	exec: make(chan struct{}, maxExec),
			87	+	blobs: make(chan struct{}, maxBlobs),
60		}	88		}
61		}	89		}
62			90

@@ -80,6 +108,18 @@ func buildEnv(dataDir, path string) []string {
80		// commandTimeout bounds one git invocation when ctx has no deadline.	108		// commandTimeout bounds one git invocation when ctx has no deadline.
81		const commandTimeout = 2 * time.Minute	109		const commandTimeout = 2 * time.Minute
82			110
			111	+	// acquireExec takes a slot of the one-shot processes, waiting until ctx ends.
			112	+	// The wait counts against the deadline of ctx, so a saturated server answers a
			113	+	// request with its deadline rather than hanging it.
			114	+	func (g *Git) acquireExec(ctx context.Context) (release func(), err error) {
			115	+	select {
			116	+	case g.exec <- struct{}{}:
			117	+	return func() { <-g.exec }, nil
			118	+	case <-ctx.Done():
			119	+	return nil, ctx.Err()
			120	+	}
			121	+	}
			122	+
83		// run executes git with args in the isolated environment and returns stdout.	123		// run executes git with args in the isolated environment and returns stdout.
84		func (g *Git) run(ctx context.Context, args ...string) (string, error) {	124		func (g *Git) run(ctx context.Context, args ...string) (string, error) {
85		if _, ok := ctx.Deadline(); !ok {	125		if _, ok := ctx.Deadline(); !ok {

@@ -87,6 +127,11 @@ func (g *Git) run(ctx context.Context, args ...string) (string, error) {
87		ctx, cancel = context.WithTimeout(ctx, commandTimeout)	127		ctx, cancel = context.WithTimeout(ctx, commandTimeout)
88		defer cancel()	128		defer cancel()
89		}	129		}
			130	+	release, err := g.acquireExec(ctx)
			131	+	if err != nil {
			132	+	return "", fmt.Errorf("git %s: %w", subcommand(args), err)
			133	+	}
			134	+	defer release()
90		cmd := exec.CommandContext(ctx, g.bin, args...)	135		cmd := exec.CommandContext(ctx, g.bin, args...)
91		cmd.Env = g.env	136		cmd.Env = g.env
92		cmd.Dir = g.dataDir	137		cmd.Dir = g.dataDir

M internal/adapters/gitcli/pipe.go +5 -0


@@ -40,6 +40,11 @@ func (g Git) pipe(ctx context.Context, args []string, fn func(bufio.Reader) er
40	40		ctx, cancel = context.WithTimeout(ctx, commandTimeout)
41	41		defer cancel()
42	42		}
	43	+	release, err := g.acquireExec(ctx)
	44	+	if err != nil {
	45	+	return fmt.Errorf("git %s: %w", subcommand(args), err)
	46	+	}
	47	+	defer release()
43	48		base := ctx
44	49		ctx, cancel := context.WithCancel(base)
45	50		defer cancel()


@@ -40,6 +40,11 @@ func (g Git) pipe(ctx context.Context, args []string, fn func(bufio.Reader) er
40	ctx, cancel = context.WithTimeout(ctx, commandTimeout)	40		ctx, cancel = context.WithTimeout(ctx, commandTimeout)
41	defer cancel()	41		defer cancel()
42	}	42		}
		43	+	release, err := g.acquireExec(ctx)
		44	+	if err != nil {
		45	+	return fmt.Errorf("git %s: %w", subcommand(args), err)
		46	+	}
		47	+	defer release()
43	base := ctx	48		base := ctx
44	ctx, cancel := context.WithCancel(base)	49		ctx, cancel := context.WithCancel(base)
45	defer cancel()	50		defer cancel()

M internal/adapters/gitcli/pool_test.go +79 -9


@@ -8,6 +8,7 @@ import (
8	8		"errors"
9	9		"fmt"
10	10		"io"
	11	+	"os"
11	12		"runtime"
12	13		"sync"
13	14		"syscall"

@@ -39,7 +40,9 @@ func requireGone(t *testing.T, pids ...int) {
39	40
40	41		func TestCatFileProcessIsReusedAndStopsWhenIdle(t *testing.T) {
41	42		ctx := context.Background()
42		-	g, data := newReader(t, func(c gitcli.Config) { c.CatFileIdle = 300 time.Millisecond })
	43	+	// The default idle timeout (60 s) never fires during the test: the hooks
	44	+	// do what the timers would, deterministically.
	45	+	g, data := newReader(t, nil)
43	46		h := newHistory(t, data)
44	47
45	48		assert.Empty(t, g.CatFilePIDs(), "nothing runs before the first read")

@@ -55,16 +58,13 @@ func TestCatFileProcessIsReusedAndStopsWhenIdle(t *testing.T) {
55	58		}
56	59		assert.Equal(t, pids, g.CatFilePIDs(), "one long-lived process serves every request")
57	60
58		-	// Use keeps it alive past the idle timeout...
59		-	for range 4 {
60		-	time.Sleep(150 * time.Millisecond)
61		-	_, err = g.ResolveRevision(ctx, h.name, domain.Revision{})
62		-	require.NoError(t, err)
63		-	}
	61	+	// A timer armed before the last use is stale: use keeps the process alive.
	62	+	g.FireStaleIdleTimers()
64	63		assert.Equal(t, pids, g.CatFilePIDs())
65	64
66		-	// ...and idleness stops it.
67		-	require.Eventually(t, func() bool { return len(g.CatFilePIDs()) == 0 }, 5time.Second, 10time.Millisecond)
	65	+	// The timer of the last use stops it once nobody used it since.
	66	+	g.FireIdleTimers()
	67	+	require.Empty(t, g.CatFilePIDs())
68	68		requireGone(t, pids...)
69	69
70	70		// The next request starts a new one.

@@ -75,6 +75,17 @@ func TestCatFileProcessIsReusedAndStopsWhenIdle(t *testing.T) {
75	75		assert.NotEqual(t, pids, next)
76	76		}
77	77
	78	+	func TestCatFileIdleTimerStopsAnUnusedProcess(t *testing.T) {
	79	+	ctx := context.Background()
	80	+	g, data := newReader(t, func(c gitcli.Config) { c.CatFileIdle = 20 time.Millisecond })
	81	+	h := newHistory(t, data)
	82	+	_, err := g.ResolveRevision(ctx, h.name, domain.Revision{})
	83	+	require.NoError(t, err)
	84	+	pids := g.CatFilePIDs()
	85	+	require.Eventually(t, func() bool { return len(g.CatFilePIDs()) == 0 }, 5time.Second, 5time.Millisecond)
	86	+	requireGone(t, pids...)
	87	+	}
	88	+
78	89		func TestCloseStopsEveryProcess(t *testing.T) {
79	90		ctx := context.Background()
80	91		g, data := newReader(t, nil) // default idle: 60 s, so only Close can stop them

@@ -335,3 +346,62 @@ func TestOpenBlobStreams50MiB(t *testing.T) {
335	346		assert.Less(t, growth, int64(8<<20), "heap growth while streaming 50 MiB: %d", growth)
336	347		t.Logf("streamed %d MiB: allocated %d KiB, peak heap growth %d KiB", size>>20, allocated>>10, growth>>10)
337	348		}
	349	+
	350	+	// The one-shot git commands share a bounded number of slots: a burst of
	351	+	// requests waits for a slot, and gives up with its context.
	352	+	func TestExecSemaphoreBoundsOneShotCommands(t *testing.T) {
	353	+	ctx := context.Background()
	354	+	gate := t.TempDir() + "/gate"
	355	+	// A git that blocks until the gate file exists.
	356	+	script := writeScript(t, `while [ ! -e `+gate+` ]; do sleep 0.01; done; echo "git version 2.50.0"`)
	357	+	g := gitcli.New(gitcli.Config{Bin: script, DataDir: t.TempDir(), Path: os.Getenv("PATH"), MaxExec: 2})
	358	+	size, taken := g.ExecSlots()
	359	+	require.Equal(t, 2, size)
	360	+	require.Zero(t, taken)
	361	+
	362	+	errs := make(chan error, 2)
	363	+	for range 2 {
	364	+	go func() { _, err := g.CheckVersion(ctx); errs <- err }()
	365	+	}
	366	+	require.Eventually(t, func() bool { _, taken := g.ExecSlots(); return taken == 2 }, 5*time.Second, time.Millisecond)
	367	+
	368	+	// A third waits and gives up with its deadline: it never started git.
	369	+	short, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
	370	+	defer cancel()
	371	+	_, err := g.CheckVersion(short)
	372	+	assert.ErrorIs(t, err, context.DeadlineExceeded)
	373	+
	374	+	require.NoError(t, os.WriteFile(gate, nil, 0o600))
	375	+	for range 2 {
	376	+	<-errs // their outcome is the version check's, not the subject
	377	+	}
	378	+	require.Eventually(t, func() bool { _, taken := g.ExecSlots(); return taken == 0 }, 5*time.Second, time.Millisecond)
	379	+	}
	380	+
	381	+	// Open blob streams are bounded; a refused stream leaves nothing behind and a
	382	+	// closed one frees its slot.
	383	+	func TestBlobStreamsAreBounded(t *testing.T) {
	384	+	ctx := context.Background()
	385	+	g, data := newReader(t, func(c *gitcli.Config) { c.MaxBlobStreams = 2 })
	386	+	f := newFixtureRepo(t, data, "bigone")
	387	+	f.writeBytes("big", randomBytes(600<<10, 5)).commit("big")
	388	+	open := func() (io.ReadCloser, error) {
	389	+	_, rc, err := g.OpenBlob(ctx, f.name, domain.Revision{}, tpath(t, "big"))
	390	+	return rc, err
	391	+	}
	392	+
	393	+	a, err := open()
	394	+	require.NoError(t, err)
	395	+	b, err := open()
	396	+	require.NoError(t, err)
	397	+	_, err = open()
	398	+	assert.ErrorIs(t, err, domain.ErrResourceExhausted)
	399	+	assert.Equal(t, 2, g.BlobStreams())
	400	+
	401	+	require.NoError(t, a.Close())
	402	+	c, err := open()
	403	+	require.NoError(t, err, "a closed stream frees its slot")
	404	+	require.NoError(t, b.Close())
	405	+	require.NoError(t, c.Close())
	406	+	assert.Zero(t, g.BlobStreams())
	407	+	}


@@ -8,6 +8,7 @@ import (
8		"errors"	8		"errors"
9		"fmt"	9		"fmt"
10		"io"	10		"io"
			11	+	"os"
11		"runtime"	12		"runtime"
12		"sync"	13		"sync"
13		"syscall"	14		"syscall"

@@ -39,7 +40,9 @@ func requireGone(t *testing.T, pids ...int) {
39			40
40		func TestCatFileProcessIsReusedAndStopsWhenIdle(t *testing.T) {	41		func TestCatFileProcessIsReusedAndStopsWhenIdle(t *testing.T) {
41		ctx := context.Background()	42		ctx := context.Background()
42	-	g, data := newReader(t, func(c gitcli.Config) { c.CatFileIdle = 300 time.Millisecond })	43	+	// The default idle timeout (60 s) never fires during the test: the hooks
			44	+	// do what the timers would, deterministically.
			45	+	g, data := newReader(t, nil)
43		h := newHistory(t, data)	46		h := newHistory(t, data)
44			47
45		assert.Empty(t, g.CatFilePIDs(), "nothing runs before the first read")	48		assert.Empty(t, g.CatFilePIDs(), "nothing runs before the first read")

@@ -55,16 +58,13 @@ func TestCatFileProcessIsReusedAndStopsWhenIdle(t *testing.T) {
55		}	58		}
56		assert.Equal(t, pids, g.CatFilePIDs(), "one long-lived process serves every request")	59		assert.Equal(t, pids, g.CatFilePIDs(), "one long-lived process serves every request")
57			60
58	-	// Use keeps it alive past the idle timeout...	61	+	// A timer armed before the last use is stale: use keeps the process alive.
59	-	for range 4 {	62	+	g.FireStaleIdleTimers()
60	-	time.Sleep(150 * time.Millisecond)
61	-	_, err = g.ResolveRevision(ctx, h.name, domain.Revision{})
62	-	require.NoError(t, err)
63	-	}
64		assert.Equal(t, pids, g.CatFilePIDs())	63		assert.Equal(t, pids, g.CatFilePIDs())
65			64
66	-	// ...and idleness stops it.	65	+	// The timer of the last use stops it once nobody used it since.
67	-	require.Eventually(t, func() bool { return len(g.CatFilePIDs()) == 0 }, 5time.Second, 10time.Millisecond)	66	+	g.FireIdleTimers()
			67	+	require.Empty(t, g.CatFilePIDs())
68		requireGone(t, pids...)	68		requireGone(t, pids...)
69			69
70		// The next request starts a new one.	70		// The next request starts a new one.

@@ -75,6 +75,17 @@ func TestCatFileProcessIsReusedAndStopsWhenIdle(t *testing.T) {
75		assert.NotEqual(t, pids, next)	75		assert.NotEqual(t, pids, next)
76		}	76		}
77			77
			78	+	func TestCatFileIdleTimerStopsAnUnusedProcess(t *testing.T) {
			79	+	ctx := context.Background()
			80	+	g, data := newReader(t, func(c gitcli.Config) { c.CatFileIdle = 20 time.Millisecond })
			81	+	h := newHistory(t, data)
			82	+	_, err := g.ResolveRevision(ctx, h.name, domain.Revision{})
			83	+	require.NoError(t, err)
			84	+	pids := g.CatFilePIDs()
			85	+	require.Eventually(t, func() bool { return len(g.CatFilePIDs()) == 0 }, 5time.Second, 5time.Millisecond)
			86	+	requireGone(t, pids...)
			87	+	}
			88	+
78		func TestCloseStopsEveryProcess(t *testing.T) {	89		func TestCloseStopsEveryProcess(t *testing.T) {
79		ctx := context.Background()	90		ctx := context.Background()
80		g, data := newReader(t, nil) // default idle: 60 s, so only Close can stop them	91		g, data := newReader(t, nil) // default idle: 60 s, so only Close can stop them

@@ -335,3 +346,62 @@ func TestOpenBlobStreams50MiB(t *testing.T) {
335		assert.Less(t, growth, int64(8<<20), "heap growth while streaming 50 MiB: %d", growth)	346		assert.Less(t, growth, int64(8<<20), "heap growth while streaming 50 MiB: %d", growth)
336		t.Logf("streamed %d MiB: allocated %d KiB, peak heap growth %d KiB", size>>20, allocated>>10, growth>>10)	347		t.Logf("streamed %d MiB: allocated %d KiB, peak heap growth %d KiB", size>>20, allocated>>10, growth>>10)
337		}	348		}
			349	+
			350	+	// The one-shot git commands share a bounded number of slots: a burst of
			351	+	// requests waits for a slot, and gives up with its context.
			352	+	func TestExecSemaphoreBoundsOneShotCommands(t *testing.T) {
			353	+	ctx := context.Background()
			354	+	gate := t.TempDir() + "/gate"
			355	+	// A git that blocks until the gate file exists.
			356	+	script := writeScript(t, `while [ ! -e `+gate+` ]; do sleep 0.01; done; echo "git version 2.50.0"`)
			357	+	g := gitcli.New(gitcli.Config{Bin: script, DataDir: t.TempDir(), Path: os.Getenv("PATH"), MaxExec: 2})
			358	+	size, taken := g.ExecSlots()
			359	+	require.Equal(t, 2, size)
			360	+	require.Zero(t, taken)
			361	+
			362	+	errs := make(chan error, 2)
			363	+	for range 2 {
			364	+	go func() { _, err := g.CheckVersion(ctx); errs <- err }()
			365	+	}
			366	+	require.Eventually(t, func() bool { _, taken := g.ExecSlots(); return taken == 2 }, 5*time.Second, time.Millisecond)
			367	+
			368	+	// A third waits and gives up with its deadline: it never started git.
			369	+	short, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
			370	+	defer cancel()
			371	+	_, err := g.CheckVersion(short)
			372	+	assert.ErrorIs(t, err, context.DeadlineExceeded)
			373	+
			374	+	require.NoError(t, os.WriteFile(gate, nil, 0o600))
			375	+	for range 2 {
			376	+	<-errs // their outcome is the version check's, not the subject
			377	+	}
			378	+	require.Eventually(t, func() bool { _, taken := g.ExecSlots(); return taken == 0 }, 5*time.Second, time.Millisecond)
			379	+	}
			380	+
			381	+	// Open blob streams are bounded; a refused stream leaves nothing behind and a
			382	+	// closed one frees its slot.
			383	+	func TestBlobStreamsAreBounded(t *testing.T) {
			384	+	ctx := context.Background()
			385	+	g, data := newReader(t, func(c *gitcli.Config) { c.MaxBlobStreams = 2 })
			386	+	f := newFixtureRepo(t, data, "bigone")
			387	+	f.writeBytes("big", randomBytes(600<<10, 5)).commit("big")
			388	+	open := func() (io.ReadCloser, error) {
			389	+	_, rc, err := g.OpenBlob(ctx, f.name, domain.Revision{}, tpath(t, "big"))
			390	+	return rc, err
			391	+	}
			392	+
			393	+	a, err := open()
			394	+	require.NoError(t, err)
			395	+	b, err := open()
			396	+	require.NoError(t, err)
			397	+	_, err = open()
			398	+	assert.ErrorIs(t, err, domain.ErrResourceExhausted)
			399	+	assert.Equal(t, 2, g.BlobStreams())
			400	+
			401	+	require.NoError(t, a.Close())
			402	+	c, err := open()
			403	+	require.NoError(t, err, "a closed stream frees its slot")
			404	+	require.NoError(t, b.Close())
			405	+	require.NoError(t, c.Close())
			406	+	assert.Zero(t, g.BlobStreams())
			407	+	}