From 01de99dfa33af3a5d4a3d70d407b781134521bf1 Mon Sep 17 00:00:00 2001 From: Yushan Lin Date: Thu, 9 Jul 2026 13:01:25 -0700 Subject: [PATCH 1/3] Ensure PoolSize is set --- core/repomanager/repo_manager.go | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/core/repomanager/repo_manager.go b/core/repomanager/repo_manager.go index 2bd5f60..e82266c 100644 --- a/core/repomanager/repo_manager.go +++ b/core/repomanager/repo_manager.go @@ -86,7 +86,10 @@ type Params struct { // appCtx is the application-lifetime context. Cancel it when the process is // shutting down (e.g. wire it to SIGTERM/SIGINT in main) to abort any // background goroutines the manager spawns. -func NewRepoManager(appCtx context.Context, p Params) RepoManager { +func NewRepoManager(appCtx context.Context, p Params) (RepoManager, error) { + if p.PoolSize <= 0 { + return nil, fmt.Errorf("pool size must be > 0, got %d", p.PoolSize) + } return &repoManager{ git: p.Git, repoManagerClonePath: p.RepoManagerClonePath, @@ -105,10 +108,6 @@ func (r *repoManager) poolFor(repo string) *workerPool { if pool, ok := r.pools[repo]; ok { return pool } - // default to 1 worker workspace per repo - if r.poolSize <= 0 { - r.poolSize = 1 - } pool := &workerPool{ originDir: filepath.Join(r.repoManagerClonePath, repo), avail: make(chan *workerSlot, r.poolSize), From 8cc0d9611310d725edeb709e2f9ec88908c64765 Mon Sep 17 00:00:00 2001 From: Yushan Lin Date: Thu, 9 Jul 2026 13:17:40 -0700 Subject: [PATCH 2/3] Update --- core/repomanager/repo_manager.go | 2 +- core/repomanager/repo_manager_test.go | 134 +++++++++++++++++++++++--- example/main.go | 5 +- 3 files changed, 128 insertions(+), 13 deletions(-) diff --git a/core/repomanager/repo_manager.go b/core/repomanager/repo_manager.go index e82266c..337439b 100644 --- a/core/repomanager/repo_manager.go +++ b/core/repomanager/repo_manager.go @@ -98,7 +98,7 @@ func NewRepoManager(appCtx context.Context, p Params) (RepoManager, error) { poolSize: p.PoolSize, pools: make(map[string]*workerPool), appCtx: appCtx, - } + }, nil } func (r *repoManager) poolFor(repo string) *workerPool { diff --git a/core/repomanager/repo_manager_test.go b/core/repomanager/repo_manager_test.go index 1d0b951..5e37990 100644 --- a/core/repomanager/repo_manager_test.go +++ b/core/repomanager/repo_manager_test.go @@ -30,6 +30,23 @@ import ( "go.uber.org/zap" ) +func newTestRepoManager(t *testing.T, appCtx context.Context, p Params) RepoManager { + t.Helper() + rm, err := NewRepoManager(appCtx, p) + require.NoError(t, err) + return rm +} + +func TestNewRepoManager_InvalidPoolSize(t *testing.T) { + t.Parallel() + _, err := NewRepoManager(context.Background(), Params{ + Logger: zap.NewNop().Sugar(), + PoolSize: 0, + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "pool size must be > 0") +} + func TestLease_ClonesOriginAndCreatesWorker(t *testing.T) { t.Parallel() ctrl := gomock.NewController(t) @@ -43,7 +60,7 @@ func TestLease_ClonesOriginAndCreatesWorker(t *testing.T) { g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil) g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(nil) - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) ws, err := rm.Lease(context.Background(), tangopb.BuildDescription{Remote: remote}) require.NoError(t, err) assert.Equal(t, workerDir, ws.Path()) @@ -65,7 +82,7 @@ func TestLease_SkipsOriginClone_WhenExists(t *testing.T) { // Only worker clone expected g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(nil) - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) ws, err := rm.Lease(context.Background(), tangopb.BuildDescription{Remote: remote}) require.NoError(t, err) assert.Equal(t, workerDir, ws.Path()) @@ -86,7 +103,7 @@ func TestLease_ReusesWorker_AfterRelease(t *testing.T) { g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil) g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(nil) - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) ctx := context.Background() ws1, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) @@ -115,7 +132,7 @@ func TestLease_CreatesMultipleWorkers(t *testing.T) { g.EXPECT().Clone(gomock.Any(), originDir, dir, "--local", "-c", "gc.auto=0").Return(nil) } - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 2}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 2}) ctx := context.Background() ws1, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) @@ -141,7 +158,7 @@ func TestLease_BlocksUntilReturn(t *testing.T) { g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil) g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(nil) - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) ctx := context.Background() ws1, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) @@ -187,7 +204,7 @@ func TestLease_CtxCanceled(t *testing.T) { g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil) g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(nil) - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) ws1, err := rm.Lease(context.Background(), tangopb.BuildDescription{Remote: remote}) require.NoError(t, err) @@ -209,7 +226,7 @@ func TestLease_OriginCloneFails(t *testing.T) { remote := "git@github.com:org/repo" g.EXPECT().Clone(gomock.Any(), remote, filepath.Join(root, "org/repo"), "-c", "gc.auto=0").Return(assert.AnError) - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) _, err := rm.Lease(context.Background(), tangopb.BuildDescription{Remote: remote}) require.Error(t, err) assert.Contains(t, err.Error(), "clone origin") @@ -228,7 +245,7 @@ func TestLease_WorkerCloneFails(t *testing.T) { g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil) g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(assert.AnError) - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) _, err := rm.Lease(context.Background(), tangopb.BuildDescription{Remote: remote}) require.Error(t, err) assert.Contains(t, err.Error(), "create worker") @@ -247,7 +264,7 @@ func TestLease_DiscoversExistingWorker(t *testing.T) { require.NoError(t, os.MkdirAll(filepath.Join(root, ".workers", "org/repo", "worker-1", ".git"), 0o755)) // No Clone calls — everything already exists - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) ws, err := rm.Lease(context.Background(), tangopb.BuildDescription{Remote: remote}) require.NoError(t, err) assert.Contains(t, ws.Path(), "worker-1") @@ -273,7 +290,7 @@ func TestLease_DifferentRepos_IndependentPools(t *testing.T) { g.EXPECT().Clone(gomock.Any(), remote2, origin2, "-c", "gc.auto=0").Return(nil) g.EXPECT().Clone(gomock.Any(), origin2, worker2, "--local", "-c", "gc.auto=0").Return(nil) - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) ctx := context.Background() // Both repos can be leased concurrently even with pool size 1 @@ -306,7 +323,7 @@ func TestLease_WorkerCloneFails_SlotReturnedToPool(t *testing.T) { g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(nil), ) - rm := NewRepoManager(context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) ctx := context.Background() // First attempt fails @@ -318,3 +335,98 @@ func TestLease_WorkerCloneFails_SlotReturnedToPool(t *testing.T) { require.NoError(t, err) require.NoError(t, ws.Release()) } + +func TestLease_PoolBoundsWorkerCount(t *testing.T) { + t.Parallel() + ctrl := gomock.NewController(t) + g := gitmock.NewMockInterface(ctrl) + + root := t.TempDir() + remote := "git@github.com:org/repo" + originDir := filepath.Join(root, "org/repo") + + g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil) + for i := 1; i <= 3; i++ { + dir := filepath.Join(root, ".workers", "org/repo", fmt.Sprintf("worker-%d", i)) + g.EXPECT().Clone(gomock.Any(), originDir, dir, "--local", "-c", "gc.auto=0").Return(nil) + } + + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 3}) + ctx := context.Background() + + // Lease all 3 slots + workspaces := make([]interface{ Release() error }, 3) + for i := 0; i < 3; i++ { + ws, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) + require.NoError(t, err) + workspaces[i] = ws + } + + // 4th lease should block — verify with a short-lived context + shortCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + _, err := rm.Lease(shortCtx, tangopb.BuildDescription{Remote: remote}) + require.Error(t, err) + + // Release all + for _, ws := range workspaces { + require.NoError(t, ws.Release()) + } + + // Now a lease succeeds again + ws, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) + require.NoError(t, err) + require.NoError(t, ws.Release()) +} + +func TestLease_ConcurrentRequestsServedSequentially(t *testing.T) { + t.Parallel() + ctrl := gomock.NewController(t) + g := gitmock.NewMockInterface(ctrl) + + root := t.TempDir() + remote := "git@github.com:org/repo" + originDir := filepath.Join(root, "org/repo") + workerDir := filepath.Join(root, ".workers", "org/repo", "worker-1") + + g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil) + g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(nil) + + rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) + ctx := context.Background() + + const numRequests = 5 + results := make(chan string, numRequests) + + ws, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) + require.NoError(t, err) + + // Launch multiple goroutines all waiting for the single slot + for i := 0; i < numRequests; i++ { + go func() { + ws, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) + if err != nil { + results <- "error" + return + } + results <- ws.Path() + ws.Release() + }() + } + + // None should have completed yet + time.Sleep(50 * time.Millisecond) + assert.Empty(t, len(results)) + + // Release the slot — goroutines should proceed one at a time + require.NoError(t, ws.Release()) + + for i := 0; i < numRequests; i++ { + select { + case path := <-results: + assert.Equal(t, workerDir, path) + case <-time.After(5 * time.Second): + t.Fatalf("goroutine %d did not complete", i) + } + } +} diff --git a/example/main.go b/example/main.go index 5141088..4c47725 100644 --- a/example/main.go +++ b/example/main.go @@ -82,13 +82,16 @@ func run() error { } defer os.RemoveAll(workerRootPath) - rm := repomanager.NewRepoManager(appCtx, repomanager.Params{ + rm, err := repomanager.NewRepoManager(appCtx, repomanager.Params{ Git: git.New(repoManagerClonePath), Logger: logger, RepoManagerClonePath: repoManagerClonePath, WorkerRootPath: workerRootPath, PoolSize: cfg.Service.WorkerPoolSize, }) + if err != nil { + return fmt.Errorf("failed to create repo manager: %w", err) + } orch, err := orchestrator.NewNativeOrchestrator(appCtx, orchestrator.Params{ Storage: store, RepoManager: rm, From 672ed7e1d9031a816f03f90767e35809cdd40cbe Mon Sep 17 00:00:00 2001 From: Yushan Lin Date: Thu, 9 Jul 2026 13:37:43 -0700 Subject: [PATCH 3/3] Update --- core/repomanager/repo_manager_test.go | 98 +-------------------------- 1 file changed, 1 insertion(+), 97 deletions(-) diff --git a/core/repomanager/repo_manager_test.go b/core/repomanager/repo_manager_test.go index 5e37990..dcda796 100644 --- a/core/repomanager/repo_manager_test.go +++ b/core/repomanager/repo_manager_test.go @@ -43,8 +43,7 @@ func TestNewRepoManager_InvalidPoolSize(t *testing.T) { Logger: zap.NewNop().Sugar(), PoolSize: 0, }) - require.Error(t, err) - assert.Contains(t, err.Error(), "pool size must be > 0") + require.Error(t, err, "expected setting invalid poolsize to error out") } func TestLease_ClonesOriginAndCreatesWorker(t *testing.T) { @@ -335,98 +334,3 @@ func TestLease_WorkerCloneFails_SlotReturnedToPool(t *testing.T) { require.NoError(t, err) require.NoError(t, ws.Release()) } - -func TestLease_PoolBoundsWorkerCount(t *testing.T) { - t.Parallel() - ctrl := gomock.NewController(t) - g := gitmock.NewMockInterface(ctrl) - - root := t.TempDir() - remote := "git@github.com:org/repo" - originDir := filepath.Join(root, "org/repo") - - g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil) - for i := 1; i <= 3; i++ { - dir := filepath.Join(root, ".workers", "org/repo", fmt.Sprintf("worker-%d", i)) - g.EXPECT().Clone(gomock.Any(), originDir, dir, "--local", "-c", "gc.auto=0").Return(nil) - } - - rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 3}) - ctx := context.Background() - - // Lease all 3 slots - workspaces := make([]interface{ Release() error }, 3) - for i := 0; i < 3; i++ { - ws, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) - require.NoError(t, err) - workspaces[i] = ws - } - - // 4th lease should block — verify with a short-lived context - shortCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) - defer cancel() - _, err := rm.Lease(shortCtx, tangopb.BuildDescription{Remote: remote}) - require.Error(t, err) - - // Release all - for _, ws := range workspaces { - require.NoError(t, ws.Release()) - } - - // Now a lease succeeds again - ws, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) - require.NoError(t, err) - require.NoError(t, ws.Release()) -} - -func TestLease_ConcurrentRequestsServedSequentially(t *testing.T) { - t.Parallel() - ctrl := gomock.NewController(t) - g := gitmock.NewMockInterface(ctrl) - - root := t.TempDir() - remote := "git@github.com:org/repo" - originDir := filepath.Join(root, "org/repo") - workerDir := filepath.Join(root, ".workers", "org/repo", "worker-1") - - g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil) - g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(nil) - - rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop().Sugar(), RepoManagerClonePath: root, WorkerRootPath: filepath.Join(root, ".workers"), PoolSize: 1}) - ctx := context.Background() - - const numRequests = 5 - results := make(chan string, numRequests) - - ws, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) - require.NoError(t, err) - - // Launch multiple goroutines all waiting for the single slot - for i := 0; i < numRequests; i++ { - go func() { - ws, err := rm.Lease(ctx, tangopb.BuildDescription{Remote: remote}) - if err != nil { - results <- "error" - return - } - results <- ws.Path() - ws.Release() - }() - } - - // None should have completed yet - time.Sleep(50 * time.Millisecond) - assert.Empty(t, len(results)) - - // Release the slot — goroutines should proceed one at a time - require.NoError(t, ws.Release()) - - for i := 0; i < numRequests; i++ { - select { - case path := <-results: - assert.Equal(t, workerDir, path) - case <-time.After(5 * time.Second): - t.Fatalf("goroutine %d did not complete", i) - } - } -}