From 7a044b7e027352d5336c2740d00cb2c97d6c8be6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9ment=20Fiere?= Date: Thu, 23 Jul 2026 18:46:25 +0200 Subject: [PATCH] fix(task): dedupe repository listing and close local-path create race (#1897) --- .../executor/executor_mocks_test.go | 3 + .../task/handlers/process_handlers_test.go | 3 + .../internal/task/repository/interface.go | 9 + .../repository/sqlite/repository_entity.go | 35 ++- .../sqlite/repository_entity_test.go | 38 +++ apps/backend/internal/task/service/service.go | 8 + .../task/service/service_resources.go | 105 ++++++- .../service/service_resources_dedupe_test.go | 292 ++++++++++++++++++ .../internal/task/service/service_tasks.go | 27 +- 9 files changed, 510 insertions(+), 10 deletions(-) create mode 100644 apps/backend/internal/task/service/service_resources_dedupe_test.go diff --git a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go index 5f1c900d8e..4e5bf8314b 100644 --- a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go @@ -964,6 +964,9 @@ func (m *mockRepository) ListScriptsByRepositoryIDs(_ context.Context, _ []strin func (m *mockRepository) GetRepositoryByProviderInfo(_ context.Context, _, _, _, _, _ string) (*models.Repository, error) { return nil, nil } +func (m *mockRepository) GetRepositoryByLocalPath(_ context.Context, _, _ string) (*models.Repository, error) { + return nil, nil +} // Executor operations func (m *mockRepository) CreateExecutor(ctx context.Context, executor *models.Executor) error { diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index 4b595fd956..8db333e6d6 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -381,6 +381,9 @@ func (m *mockRepository) ListScriptsByRepositoryIDs(_ context.Context, _ []strin func (m *mockRepository) GetRepositoryByProviderInfo(_ context.Context, _, _, _, _, _ string) (*models.Repository, error) { return nil, nil } +func (m *mockRepository) GetRepositoryByLocalPath(_ context.Context, _, _ string) (*models.Repository, error) { + return nil, nil +} func (m *mockRepository) CreateExecutor(ctx context.Context, executor *models.Executor) error { return nil } diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index e0760f3d84..f12312aead 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -271,6 +271,15 @@ type RepositoryEntityRepository interface { ListRepositoryScripts(ctx context.Context, repositoryID string) ([]*models.RepositoryScript, error) ListScriptsByRepositoryIDs(ctx context.Context, repoIDs []string) (map[string][]*models.RepositoryScript, error) GetRepositoryByProviderInfo(ctx context.Context, workspaceID, provider, host, owner, name string) (*models.Repository, error) + // GetRepositoryByLocalPath finds a live repository by workspace and canonical + // local_path. Returns nil, nil if not found. Used by + // Service.FindOrCreateRepositoryByLocalPath to check for an existing row by + // canonical path immediately before insert (serialized via repoResolveMu), + // instead of relying solely on a batch snapshot that can go stale across + // concurrent callers within this process. This closes the common + // single-process race; it is not a substitute for a database-level + // uniqueness constraint against writers outside this process. + GetRepositoryByLocalPath(ctx context.Context, workspaceID, localPath string) (*models.Repository, error) } // ExecutorRepository handles executor CRUD, executor profiles, and running state. diff --git a/apps/backend/internal/task/repository/sqlite/repository_entity.go b/apps/backend/internal/task/repository/sqlite/repository_entity.go index 7960c713d0..d3a650f96b 100644 --- a/apps/backend/internal/task/repository/sqlite/repository_entity.go +++ b/apps/backend/internal/task/repository/sqlite/repository_entity.go @@ -154,7 +154,13 @@ func (r *Repository) ListRepositories(ctx context.Context, workspaceID string) ( } // GetRepositoryByProviderInfo finds a repository by workspace, provider, owner, and name. -// Returns nil, nil if not found. +// Returns nil, nil if not found. When duplicate rows share the same provider +// identity (e.g. left behind by a resolver race that predates +// Service.repoResolveMu), orders by created_at then id so the row returned +// here is always the same earliest-created row that +// dedupeRepositoriesByIdentity keeps as the canonical winner in +// ListRepositories — otherwise a caller could resolve to, and write +// backfilled fields onto, a duplicate that ListRepositories hides. func (r *Repository) GetRepositoryByProviderInfo(ctx context.Context, workspaceID, provider, host, owner, name string) (*models.Repository, error) { repository := &models.Repository{} err := r.ro.QueryRowContext(ctx, r.ro.Rebind(` @@ -163,6 +169,8 @@ func (r *Repository) GetRepositoryByProviderInfo(ctx context.Context, workspaceI FROM repositories WHERE workspace_id = ? AND provider = ? AND provider_host = ? AND provider_owner = ? AND provider_name = ? AND deleted_at IS NULL + ORDER BY created_at ASC, id ASC + LIMIT 1 `), workspaceID, provider, host, owner, name).Scan( &repository.ID, &repository.WorkspaceID, &repository.Name, &repository.SourceType, &repository.LocalPath, &repository.Provider, &repository.ProviderRepoID, &repository.ProviderHost, &repository.ProviderOwner, &repository.ProviderName, &repository.RemoteURL, @@ -174,6 +182,31 @@ func (r *Repository) GetRepositoryByProviderInfo(ctx context.Context, workspaceI return repository, err } +// GetRepositoryByLocalPath finds a live repository by workspace and canonical +// local_path. Returns nil, nil if not found. Mirrors GetRepositoryByProviderInfo, +// including the created_at/id tiebreak, so local-path resolution can do the +// same lookup-then-create as provider-URL resolution instead of trusting a +// possibly-stale in-memory snapshot. +func (r *Repository) GetRepositoryByLocalPath(ctx context.Context, workspaceID, localPath string) (*models.Repository, error) { + repository := &models.Repository{} + err := r.ro.QueryRowContext(ctx, r.ro.Rebind(` + SELECT id, workspace_id, name, source_type, local_path, provider, provider_repo_id, provider_host, provider_owner, + provider_name, remote_url, default_branch, worktree_branch_prefix, worktree_branch_template, pull_before_worktree, setup_script, cleanup_script, dev_script, copy_files, created_at, updated_at, deleted_at + FROM repositories + WHERE workspace_id = ? AND local_path = ? AND local_path != '' AND deleted_at IS NULL + ORDER BY created_at ASC, id ASC + LIMIT 1 + `), workspaceID, localPath).Scan( + &repository.ID, &repository.WorkspaceID, &repository.Name, &repository.SourceType, &repository.LocalPath, + &repository.Provider, &repository.ProviderRepoID, &repository.ProviderHost, &repository.ProviderOwner, &repository.ProviderName, &repository.RemoteURL, + &repository.DefaultBranch, &repository.WorktreeBranchPrefix, &repository.WorktreeBranchTemplate, &repository.PullBeforeWorktree, &repository.SetupScript, &repository.CleanupScript, &repository.DevScript, &repository.CopyFiles, &repository.CreatedAt, &repository.UpdatedAt, &repository.DeletedAt, + ) + if err == sql.ErrNoRows { + return nil, nil + } + return repository, err +} + // CreateRepositoryScript creates a new repository script func (r *Repository) CreateRepositoryScript(ctx context.Context, script *models.RepositoryScript) error { if script.ID == "" { diff --git a/apps/backend/internal/task/repository/sqlite/repository_entity_test.go b/apps/backend/internal/task/repository/sqlite/repository_entity_test.go index 6b2bca2937..3aecdb810b 100644 --- a/apps/backend/internal/task/repository/sqlite/repository_entity_test.go +++ b/apps/backend/internal/task/repository/sqlite/repository_entity_test.go @@ -5,6 +5,7 @@ import ( "errors" "path/filepath" "testing" + "time" "github.com/jmoiron/sqlx" @@ -121,6 +122,43 @@ func TestGetRepositoryByProviderInfoSeparatesGitLabHosts(t *testing.T) { } } +// TestGetRepositoryByProviderInfoReturnsEarliestCreatedDuplicate guards the +// Greptile-flagged race window: when two rows already share the same +// provider identity (left over from a resolver race that predates +// Service.repoResolveMu), GetRepositoryByProviderInfo must resolve to the +// same row ListRepositories' dedupeRepositoriesByIdentity keeps as the +// canonical winner (earliest created_at, ties broken by the smaller id) — +// not an arbitrary one of the two — otherwise a caller can attach a task to, +// or backfill fields onto, the duplicate ListRepositories hides. +func TestGetRepositoryByProviderInfoReturnsEarliestCreatedDuplicate(t *testing.T) { + repo := newRepoForEntityTests(t) + ctx := context.Background() + seedWorkspace(t, repo, "ws-provider-dup") + for _, item := range []*models.Repository{ + {ID: "repo-dup-later", WorkspaceID: "ws-provider-dup", Name: "later", SourceType: "provider", Provider: "github", ProviderHost: "https://github.com", ProviderOwner: "kdlbs", ProviderName: "kandev"}, + {ID: "repo-dup-earlier", WorkspaceID: "ws-provider-dup", Name: "earlier", SourceType: "provider", Provider: "github", ProviderHost: "https://github.com", ProviderOwner: "kdlbs", ProviderName: "kandev"}, + } { + if err := repo.CreateRepository(ctx, item); err != nil { + t.Fatalf("create repository %s: %v", item.ID, err) + } + } + // CreateRepository always stamps created_at = time.Now(), so backdate the + // intended winner directly to make ordering deterministic regardless of + // wall-clock resolution. + earlier := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC) + if _, err := repo.db.ExecContext(ctx, repo.db.Rebind(`UPDATE repositories SET created_at = ? WHERE id = ?`), earlier, "repo-dup-earlier"); err != nil { + t.Fatalf("backdate repo-dup-earlier: %v", err) + } + + got, err := repo.GetRepositoryByProviderInfo(ctx, "ws-provider-dup", "github", "https://github.com", "kdlbs", "kandev") + if err != nil { + t.Fatalf("GetRepositoryByProviderInfo: %v", err) + } + if got == nil || got.ID != "repo-dup-earlier" { + t.Fatalf("GetRepositoryByProviderInfo = %+v, want repo-dup-earlier (the row ListRepositories keeps as canonical)", got) + } +} + func TestRepositoryProviderHostMigrationBackfillsOnlyUnambiguousGitHubRows(t *testing.T) { repo := newRepoForEntityTests(t) ctx := context.Background() diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index e9a92e08bc..c9595f47df 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -224,6 +224,14 @@ type Service struct { cleanupWorkerWake chan struct{} cleanupRunsMu sync.Mutex cleanupRuns map[*taskResourceCleanupRun]struct{} + // repoResolveMu serializes the check-then-create sections of + // FindOrCreateRepository and FindOrCreateRepositoryByLocalPath so two + // resolvers racing to register the same not-yet-known repository (by + // provider identity or by canonical local_path) converge on a single row + // instead of each inserting a duplicate. Covers concurrent requests + // within this backend process only — this backend is single-process per + // SQLite database, so that is the complete threat model today. + repoResolveMu sync.Mutex } // NewService creates a new task service diff --git a/apps/backend/internal/task/service/service_resources.go b/apps/backend/internal/task/service/service_resources.go index ad082a2f9b..3b904c8a97 100644 --- a/apps/backend/internal/task/service/service_resources.go +++ b/apps/backend/internal/task/service/service_resources.go @@ -695,6 +695,8 @@ func (s *Service) GetRepositoryByProviderInfo(ctx context.Context, workspaceID, // create race between snapshot and lookup, so a snapshot-miss does NOT // mean this call created the row. func (s *Service) FindOrCreateRepository(ctx context.Context, req *FindOrCreateRepositoryRequest) (*models.Repository, bool, error) { + s.repoResolveMu.Lock() + defer s.repoResolveMu.Unlock() req.ProviderHost = normalizeProviderHost(req.Provider, req.ProviderHost) existing, err := s.repoEntities.GetRepositoryByProviderInfo( ctx, req.WorkspaceID, req.Provider, req.ProviderHost, req.ProviderOwner, req.ProviderName, @@ -766,6 +768,45 @@ func (s *Service) FindOrCreateRepository(ctx context.Context, req *FindOrCreateR return created, true, nil } +// FindOrCreateRepositoryByLocalPath looks up a repository by its canonical +// local_path within workspaceID, creating one from req if none exists. +// Mirrors FindOrCreateRepository's provider-identity flow: the lookup and the +// insert both happen while holding repoResolveMu, so two resolvers racing to +// register the same not-yet-known on-disk repo (e.g. two task-creation +// requests naming the same local_path) converge on one row instead of each +// inserting its own. canonicalPath must already be resolved (see +// resolveExplicitLocalRepositoryPath); pass "" when canonicalization failed +// or was skipped, which disables the lookup and always creates — matching +// prior behavior for that edge case, since there is no reliable identity to +// dedupe against. +// +// Returns created=true only when this call inserted the new row. +func (s *Service) FindOrCreateRepositoryByLocalPath( + ctx context.Context, workspaceID, canonicalPath string, req *CreateRepositoryRequest, +) (*models.Repository, bool, error) { + s.repoResolveMu.Lock() + defer s.repoResolveMu.Unlock() + + if canonicalPath != "" { + existing, err := s.repoEntities.GetRepositoryByLocalPath(ctx, workspaceID, canonicalPath) + if err != nil { + return nil, false, fmt.Errorf("lookup repository by local path: %w", err) + } + if existing != nil { + replacement, replacementCreated, replaceErr := s.replaceTaskWorktreeRepositoryMatch(ctx, workspaceID, existing) + if replaceErr != nil { + return nil, false, replaceErr + } + return replacement, replacementCreated, nil + } + } + created, createErr := s.CreateRepository(ctx, req) + if createErr != nil { + return nil, false, createErr + } + return created, true, nil +} + func (s *Service) UpdateRepository(ctx context.Context, id string, req *UpdateRepositoryRequest) (*models.Repository, error) { repository, err := s.repoEntities.GetRepository(ctx, id) if err != nil { @@ -938,7 +979,69 @@ func (s *Service) ListRepositories(ctx context.Context, workspaceID string) ([]* live = append(live, repository) } } - return live, nil + return dedupeRepositoriesByIdentity(live), nil +} + +// repositoryIdentityKey returns a stable dedup key for repo: local_path when +// set (a local checkout is identified by where it lives on disk), otherwise +// the provider identity tuple when host/owner/name are all present (a +// provider-backed repo is identified by where it lives upstream), otherwise +// the row's own ID. A missing provider_host (legacy rows, or self-managed +// providers we've never normalized a host for) means we cannot tell two +// same-namespace repos on different unknown hosts apart, so those rows fail +// closed to their own ID instead of risking a false-positive collapse. +// Placeholder rows with neither local_path nor provider also fall back to ID +// so they never collide with one another. +func repositoryIdentityKey(repo *models.Repository) string { + if repo.LocalPath != "" { + return "local\x00" + repo.LocalPath + } + if repo.Provider != "" && repo.ProviderHost != "" && repo.ProviderOwner != "" && repo.ProviderName != "" { + return "provider\x00" + repo.Provider + "\x00" + repo.ProviderHost + "\x00" + repo.ProviderOwner + "\x00" + repo.ProviderName + } + return "id\x00" + repo.ID +} + +// dedupeRepositoriesByIdentity collapses rows that share a +// repositoryIdentityKey — e.g. two rows for the same local_path left behind +// by a resolver race, or the same provider repo registered twice — down to +// one. Keeps the earliest-created row per key (ties broken by the smaller +// ID), matching the winner FindOrCreateRepository / +// FindOrCreateRepositoryByLocalPath resolve future references to via +// GetRepositoryByProviderInfo / GetRepositoryByLocalPath's +// `ORDER BY created_at ASC, id ASC` — so callers do not add a +// task_repositories link the UI would then de-list. This is a read-time +// safety net: it hides pre-existing duplicate rows from callers without +// touching the underlying table, so a caller that still deletes by ID (e.g. +// DeleteRepository) must use the ID this function returned, not one filtered +// out. Preserves the relative order of first occurrence. +func dedupeRepositoriesByIdentity(repos []*models.Repository) []*models.Repository { + winners := make(map[string]*models.Repository, len(repos)) + for _, repo := range repos { + if repo == nil { + continue + } + key := repositoryIdentityKey(repo) + current, ok := winners[key] + if !ok || repo.CreatedAt.Before(current.CreatedAt) || + (repo.CreatedAt.Equal(current.CreatedAt) && repo.ID < current.ID) { + winners[key] = repo + } + } + deduped := make([]*models.Repository, 0, len(winners)) + seen := make(map[string]struct{}, len(winners)) + for _, repo := range repos { + if repo == nil { + continue + } + key := repositoryIdentityKey(repo) + if _, done := seen[key]; done { + continue + } + seen[key] = struct{}{} + deduped = append(deduped, winners[key]) + } + return deduped } // CountActiveSessionsByRepository returns the number of agent sessions in an diff --git a/apps/backend/internal/task/service/service_resources_dedupe_test.go b/apps/backend/internal/task/service/service_resources_dedupe_test.go new file mode 100644 index 0000000000..fa9ff50a59 --- /dev/null +++ b/apps/backend/internal/task/service/service_resources_dedupe_test.go @@ -0,0 +1,292 @@ +package service + +import ( + "context" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" +) + +// TestDedupeRepositoriesByIdentity exercises the pure winner-selection logic +// directly with explicit CreatedAt values, so the assertion is deterministic +// and independent of wall-clock/DB timestamp precision. Covers: local-path +// duplicates collapse to the earliest-created row, provider-identity +// duplicates collapse the same way, an unrelated repository of each kind is +// preserved, and two placeholder rows sharing neither identity are never +// merged into each other. +func TestDedupeRepositoriesByIdentity(t *testing.T) { + older := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + newer := older.Add(time.Hour) + + localWinner := &models.Repository{ID: "repo-a-first", LocalPath: "/repos/dup", CreatedAt: older} + localLoser := &models.Repository{ID: "repo-a-second", LocalPath: "/repos/dup", CreatedAt: newer} + localDistinct := &models.Repository{ID: "repo-b", LocalPath: "/repos/distinct", CreatedAt: newer} + providerWinner := &models.Repository{ + ID: "repo-gh-first", Provider: "github", ProviderHost: githubProviderHost, + ProviderOwner: "kdlbs", ProviderName: "kandev", CreatedAt: older, + } + providerLoser := &models.Repository{ + ID: "repo-gh-second", Provider: "github", ProviderHost: githubProviderHost, + ProviderOwner: "kdlbs", ProviderName: "kandev", CreatedAt: newer, + } + providerDistinct := &models.Repository{ + ID: "repo-gh-other", Provider: "github", ProviderHost: githubProviderHost, + ProviderOwner: "kdlbs", ProviderName: "other", CreatedAt: newer, + } + placeholder1 := &models.Repository{ID: "repo-placeholder-1", CreatedAt: newer} + placeholder2 := &models.Repository{ID: "repo-placeholder-2", CreatedAt: newer} + + deduped := dedupeRepositoriesByIdentity([]*models.Repository{ + localLoser, localWinner, localDistinct, + providerLoser, providerWinner, providerDistinct, + placeholder1, placeholder2, + }) + + gotIDs := make(map[string]bool, len(deduped)) + for _, r := range deduped { + gotIDs[r.ID] = true + } + wantIDs := []string{ + "repo-a-first", "repo-b", + "repo-gh-first", "repo-gh-other", + "repo-placeholder-1", "repo-placeholder-2", + } + if len(deduped) != len(wantIDs) { + t.Fatalf("dedupeRepositoriesByIdentity returned %d repositories, want %d: %#v", len(deduped), len(wantIDs), deduped) + } + for _, id := range wantIDs { + if !gotIDs[id] { + t.Errorf("expected surviving repository %q, missing from result %#v", id, deduped) + } + } + for _, loserID := range []string{"repo-a-second", "repo-gh-second"} { + if gotIDs[loserID] { + t.Errorf("later-created duplicate %q should have been dropped in favor of the earliest-created winner", loserID) + } + } +} + +// TestDedupeRepositoriesByIdentity_EmptyProviderHostNeverCollapses guards the +// Codex-flagged gap: two provider rows sharing owner/name but with an empty +// provider_host (legacy rows, or a self-managed provider we've never +// normalized a host for) must NOT be treated as the same identity, since an +// empty host cannot distinguish which upstream they actually point at. +// dedupeRepositoriesByIdentity must fail closed to per-row IDs here instead +// of collapsing them. +func TestDedupeRepositoriesByIdentity_EmptyProviderHostNeverCollapses(t *testing.T) { + same := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + unknownHostA := &models.Repository{ID: "repo-unknown-a", Provider: "gitlab", ProviderOwner: "group", ProviderName: "project", CreatedAt: same} + unknownHostB := &models.Repository{ID: "repo-unknown-b", Provider: "gitlab", ProviderOwner: "group", ProviderName: "project", CreatedAt: same} + + deduped := dedupeRepositoriesByIdentity([]*models.Repository{unknownHostA, unknownHostB}) + + if len(deduped) != 2 { + t.Fatalf("dedupeRepositoriesByIdentity collapsed %d empty-provider_host rows into %d, want both preserved: %#v", + 2, len(deduped), deduped) + } +} + +// TestService_ListRepositoriesDedupesByLocalPath is the integration path for +// the local-path case: it exercises the real Service.ListRepositories (so +// pruning/filtering runs before dedupe), asserting the duplicate pair +// collapses to one entry and the unrelated repository survives. Winner +// selection itself is covered deterministically by +// TestDedupeRepositoriesByIdentity above; this test avoids asserting which of +// the two duplicate IDs wins so it never depends on DB timestamp resolution. +func TestService_ListRepositoriesDedupesByLocalPath(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + + dupPath := "/repos/dup" + for _, id := range []string{"repo-a-first", "repo-a-second"} { + if err := repo.CreateRepository(ctx, &models.Repository{ + ID: id, WorkspaceID: "ws-1", Name: id, SourceType: sourceTypeLocal, LocalPath: dupPath, + }); err != nil { + t.Fatalf("CreateRepository(%s): %v", id, err) + } + } + if err := repo.CreateRepository(ctx, &models.Repository{ + ID: "repo-b", WorkspaceID: "ws-1", Name: "distinct", SourceType: sourceTypeLocal, LocalPath: "/repos/distinct", + }); err != nil { + t.Fatalf("CreateRepository(distinct): %v", err) + } + + repositories, err := svc.ListRepositories(ctx, "ws-1") + if err != nil { + t.Fatalf("ListRepositories: %v", err) + } + if len(repositories) != 2 { + t.Fatalf("ListRepositories returned %d repositories, want 2 (one per identity): %#v", len(repositories), repositories) + } + dupWinners, sawDistinct := 0, false + for _, r := range repositories { + switch r.ID { + case "repo-a-first", "repo-a-second": + dupWinners++ + case "repo-b": + sawDistinct = true + default: + t.Fatalf("unexpected repository id %q", r.ID) + } + } + if dupWinners != 1 || !sawDistinct { + t.Fatalf("expected exactly one repo-a-* survivor plus distinct repo-b, got %#v", repositories) + } +} + +// TestService_ListRepositoriesDedupesByProviderIdentity mirrors the +// local-path case above for provider-backed repositories through the real +// Service.ListRepositories path. +func TestService_ListRepositoriesDedupesByProviderIdentity(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + + for _, id := range []string{"repo-gh-first", "repo-gh-second"} { + if err := repo.CreateRepository(ctx, &models.Repository{ + ID: id, WorkspaceID: "ws-1", Name: id, SourceType: sourceTypeProvider, + Provider: "github", ProviderHost: githubProviderHost, ProviderOwner: "kdlbs", ProviderName: "kandev", + }); err != nil { + t.Fatalf("CreateRepository(%s): %v", id, err) + } + } + if err := repo.CreateRepository(ctx, &models.Repository{ + ID: "repo-gh-other", WorkspaceID: "ws-1", Name: "kdlbs/other", SourceType: sourceTypeProvider, + Provider: "github", ProviderHost: githubProviderHost, ProviderOwner: "kdlbs", ProviderName: "other", + }); err != nil { + t.Fatalf("CreateRepository(distinct): %v", err) + } + + repositories, err := svc.ListRepositories(ctx, "ws-1") + if err != nil { + t.Fatalf("ListRepositories: %v", err) + } + if len(repositories) != 2 { + t.Fatalf("ListRepositories returned %d repositories, want 2 (one per provider identity): %#v", len(repositories), repositories) + } + dupWinners, sawDistinct := 0, false + for _, r := range repositories { + switch r.ID { + case "repo-gh-first", "repo-gh-second": + dupWinners++ + case "repo-gh-other": + sawDistinct = true + default: + t.Fatalf("unexpected repository id %q", r.ID) + } + } + if dupWinners != 1 || !sawDistinct { + t.Fatalf("expected exactly one repo-gh-* survivor plus distinct repo-gh-other, got %#v", repositories) + } +} + +// TestService_ListRepositoriesKeepsDistinctPlaceholderRepositories guards +// against an over-broad dedupe key: two rows that both carry an empty +// local_path and no provider identity (e.g. mid-creation placeholders) share +// no real identity and must never collapse into one another. +func TestService_ListRepositoriesKeepsDistinctPlaceholderRepositories(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + for _, id := range []string{"repo-placeholder-1", "repo-placeholder-2"} { + if err := repo.CreateRepository(ctx, &models.Repository{ + ID: id, WorkspaceID: "ws-1", Name: id, SourceType: sourceTypeLocal, + }); err != nil { + t.Fatalf("CreateRepository(%s): %v", id, err) + } + } + + repositories, err := svc.ListRepositories(ctx, "ws-1") + if err != nil { + t.Fatalf("ListRepositories: %v", err) + } + if len(repositories) != 2 { + t.Fatalf("ListRepositories returned %d repositories, want both placeholders preserved: %#v", len(repositories), repositories) + } +} + +// TestService_FindOrCreateRepositoryByLocalPath_ConcurrentCallsConvergeOnOneRow +// is the regression for the race this change closes: many goroutines +// resolving the same on-disk repository path at once (e.g. several +// create_task_kandev calls naming the same not-yet-registered local_path) +// must converge on a single repository row instead of each inserting its +// own. All goroutines are released together via a closed channel barrier so +// they genuinely contend for repoResolveMu rather than running sequentially. +func TestService_FindOrCreateRepositoryByLocalPath_ConcurrentCallsConvergeOnOneRow(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + + repoPath := filepath.Join(t.TempDir(), "concurrent-repo") + makeRepo(t, repoPath) + canonicalPath, _, pathErr := resolveExplicitLocalRepositoryPath(repoPath) + if pathErr != nil { + t.Fatalf("resolveExplicitLocalRepositoryPath: %v", pathErr) + } + + const n = 20 + var wg sync.WaitGroup + start := make(chan struct{}) + ids := make([]string, n) + createdFlags := make([]bool, n) + errs := make([]error, n) + for i := range n { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + resolved, created, err := svc.FindOrCreateRepositoryByLocalPath(ctx, "ws-1", canonicalPath, &CreateRepositoryRequest{ + WorkspaceID: "ws-1", + Name: "concurrent-repo", + SourceType: sourceTypeLocal, + LocalPath: repoPath, + }) + errs[i] = err + if resolved != nil { + ids[i] = resolved.ID + } + createdFlags[i] = created + }(i) + } + close(start) // release every goroutine at once + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Fatalf("call %d: FindOrCreateRepositoryByLocalPath: %v", i, err) + } + } + firstID := ids[0] + createdCount := 0 + for i := range ids { + if ids[i] != firstID { + t.Fatalf("call %d resolved to repository %q, want %q (all concurrent callers must converge on one row)", i, ids[i], firstID) + } + if createdFlags[i] { + createdCount++ + } + } + if createdCount != 1 { + t.Fatalf("createdCount = %d, want exactly 1 (exactly one call should have inserted the row)", createdCount) + } + + all, err := repo.ListRepositories(ctx, "ws-1") + if err != nil { + t.Fatalf("ListRepositories: %v", err) + } + if len(all) != 1 { + t.Fatalf("ListRepositories returned %d rows after concurrent resolution, want exactly 1: %#v", len(all), all) + } +} diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index f24eb8b719..311fa9d363 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -612,9 +612,12 @@ func pathAtOrInsideRoot(path, root string) bool { } // resolveRepoInputLocal handles the LocalPath branch of resolveRepoInput. -// Looks the path up in the workspace snapshot; on miss, calls -// CreateRepository (and reports created=true). Extracted to keep -// resolveRepoInput inside the cyclomatic-complexity budget. +// Looks the path up in the workspace snapshot first; on miss, delegates to +// FindOrCreateRepositoryByLocalPath, which re-checks the canonical path +// against the database immediately before inserting (see repoResolveMu) so a +// second resolver racing this one onto the same on-disk repo reuses its row +// instead of creating a sibling duplicate. Extracted to keep resolveRepoInput +// inside the cyclomatic-complexity budget. func (s *Service) resolveRepoInputLocal( ctx context.Context, workspaceID string, repoInput TaskRepositoryInput, repoByPath map[string]*models.Repository, baseBranch string, @@ -651,22 +654,30 @@ func (s *Service) resolveRepoInputLocal( defaultBranch = probedBranch } } - createdRepo, createErr := s.CreateRepository(ctx, &CreateRepositoryRequest{ + // identityPath is left empty when canonicalization failed: there is no + // reliable identity to dedupe against in that case, so + // FindOrCreateRepositoryByLocalPath always creates — same as prior + // behavior for that edge case. + var identityPath string + if pathErr == nil { + identityPath = canonicalPath + } + resolved, wasCreated, resolveErr := s.FindOrCreateRepositoryByLocalPath(ctx, workspaceID, identityPath, &CreateRepositoryRequest{ WorkspaceID: workspaceID, Name: name, SourceType: "local", LocalPath: repoInput.LocalPath, DefaultBranch: defaultBranch, }) - if createErr != nil { - return "", "", false, createErr + if resolveErr != nil { + return "", "", false, resolveErr } - repo = createdRepo + repo = resolved if repoByPath != nil { repoByPath[repoInput.LocalPath] = repo repoByPath[repo.LocalPath] = repo } - created = true + created = wasCreated } else { replacement, replacementCreated, replaceErr := s.replaceTaskWorktreeRepositoryMatch(ctx, workspaceID, repo) if replaceErr != nil {