fix(task): dedupe repository listing and close local-path create race (#1897)
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 == "" {
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user