Improve cached dir initialization (#401)

* Add a fast and slow path for cached dir initialization

* Improve cachedir cleanup

* Use testify in saferepomap unit tests

Author: kushnaidu

pkg/cache/repomap/saferepomap.go

@@ -25,30 +25,20 @@ type SafeRepoMap struct {
 	syncMap sync.Map
 }
 
-func (s *SafeRepoMap) Store(key repository.RepositoryKey, value repository.Repository) {
-	s.syncMap.Store(key, value)
-}
-
 func (s *SafeRepoMap) Load(key repository.RepositoryKey) (repository.Repository, bool) {
 	v, ok := s.syncMap.Load(key)
 	if !ok {
 		return nil, false
 	}
-	if loader, isLoader := v.(*repoLoader); isLoader {
-		return loader.repo, true
-	}
-	return v.(repository.Repository), true
+	return v.(*repoLoader).repo, true
 }
 
 func (s *SafeRepoMap) LoadAndDelete(key repository.RepositoryKey) (repository.Repository, bool) {
 	v, ok := s.syncMap.LoadAndDelete(key)
 	if !ok {
 		return nil, false
 	}
-	if loader, isLoader := v.(*repoLoader); isLoader {
-		return loader.repo, true
-	}
-	return v.(repository.Repository), true
+	return v.(*repoLoader).repo, true
 }
 
 func (s *SafeRepoMap) Range(f func(key, value any) bool) {
@@ -63,13 +53,13 @@ type repoLoader struct {
 
 func (s *SafeRepoMap) LoadOrCreate(key repository.RepositoryKey, create func() (repository.Repository, error)) (repository.Repository, error) {
 	loader := &repoLoader{}
-	actual, _ := s.syncMap.LoadOrStore(key, loader)
+	actual, loaded := s.syncMap.LoadOrStore(key, loader)
 	l := actual.(*repoLoader)
 
 	l.once.Do(func() {
 		l.repo, l.err = create()
-		if l.err != nil {
-			// Remove failed entry so subsequent calls can retry
+		if l.err != nil && !loaded {
+			// Remove failed entry only if this thread created it, so subsequent calls can retry
 			s.syncMap.Delete(key)
 		}
 	})

pkg/cache/repomap/saferepomap_test.go

@@ -0,0 +1,229 @@
+// Copyright 2026 The kpt and Nephio Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package repomap
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/nephio-project/porch/pkg/repository"
+	mockrepository "github.com/nephio-project/porch/test/mockery/mocks/porch/pkg/repository"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLoadOrCreate_Success(t *testing.T) {
+	m := &SafeRepoMap{}
+	key := repository.RepositoryKey{Name: "test-repo"}
+
+	repo, err := m.LoadOrCreate(key, func() (repository.Repository, error) {
+		mock := &mockrepository.MockRepository{}
+		mock.On("KubeObjectName").Return("created")
+		return mock, nil
+	})
+
+	assert.NoError(t, err)
+	assert.NotNil(t, repo)
+}
+
+func TestLoadOrCreate_Reuse(t *testing.T) {
+	m := &SafeRepoMap{}
+	key := repository.RepositoryKey{Name: "test-repo"}
+
+	var callCount int32
+	create := func() (repository.Repository, error) {
+		atomic.AddInt32(&callCount, 1)
+		return &mockrepository.MockRepository{}, nil
+	}
+
+	repo1, err1 := m.LoadOrCreate(key, create)
+	repo2, err2 := m.LoadOrCreate(key, create)
+
+	assert.NoError(t, err1)
+	assert.NoError(t, err2)
+	assert.Same(t, repo1, repo2)
+	assert.Equal(t, int32(1), atomic.LoadInt32(&callCount))
+}
+
+func TestLoadOrCreate_ErrorRetry(t *testing.T) {
+	m := &SafeRepoMap{}
+	key := repository.RepositoryKey{Name: "test-repo"}
+
+	var callCount int32
+	create := func() (repository.Repository, error) {
+		count := atomic.AddInt32(&callCount, 1)
+		if count == 1 {
+			return nil, errors.New("first attempt failed")
+		}
+		return &mockrepository.MockRepository{}, nil
+	}
+
+	// First call should fail
+	repo1, err1 := m.LoadOrCreate(key, create)
+	assert.Error(t, err1)
+	assert.Nil(t, repo1)
+
+	// Second call should succeed (retry)
+	repo2, err2 := m.LoadOrCreate(key, create)
+	assert.NoError(t, err2)
+	assert.NotNil(t, repo2)
+	assert.Equal(t, int32(2), atomic.LoadInt32(&callCount))
+}
+
+func TestLoadOrCreate_Concurrent(t *testing.T) {
+	m := &SafeRepoMap{}
+	key := repository.RepositoryKey{Name: "test-repo"}
+
+	var callCount int32
+	create := func() (repository.Repository, error) {
+		atomic.AddInt32(&callCount, 1)
+		return &mockrepository.MockRepository{}, nil
+	}
+
+	const goroutines = 10
+	var wg sync.WaitGroup
+	results := make([]repository.Repository, goroutines)
+
+	for i := 0; i < goroutines; i++ {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			repo, err := m.LoadOrCreate(key, create)
+			assert.NoError(t, err)
+			results[idx] = repo
+		}(i)
+	}
+
+	wg.Wait()
+
+	// All goroutines should get the same instance
+	for i := 1; i < goroutines; i++ {
+		assert.Same(t, results[0], results[i])
+	}
+
+	// Create should only be called once
+	assert.Equal(t, int32(1), atomic.LoadInt32(&callCount))
+}
+
+func TestLoadOrCreate_ConcurrentError(t *testing.T) {
+	m := &SafeRepoMap{}
+	key := repository.RepositoryKey{Name: "test-repo"}
+
+	var callCount int32
+	create := func() (repository.Repository, error) {
+		atomic.AddInt32(&callCount, 1)
+		return nil, errors.New("always fails")
+	}
+
+	const goroutines = 5
+	var wg sync.WaitGroup
+
+	for i := 0; i < goroutines; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_, err := m.LoadOrCreate(key, create)
+			assert.Error(t, err)
+		}()
+	}
+
+	wg.Wait()
+
+	// With concurrent errors and deletion, multiple goroutines may call create
+	// This is expected behavior - failed entries are deleted and retried
+	count := atomic.LoadInt32(&callCount)
+	assert.GreaterOrEqual(t, count, int32(1))
+	assert.LessOrEqual(t, count, int32(goroutines))
+}
+
+func TestLoad(t *testing.T) {
+	m := &SafeRepoMap{}
+	key := repository.RepositoryKey{Name: "test-repo"}
+
+	// Load non-existent key
+	repo, ok := m.Load(key)
+	assert.False(t, ok)
+	assert.Nil(t, repo)
+
+	// Create a repo
+	m.LoadOrCreate(key, func() (repository.Repository, error) {
+		return &mockrepository.MockRepository{}, nil
+	})
+
+	// Load existing key
+	repo, ok = m.Load(key)
+	assert.True(t, ok)
+	assert.NotNil(t, repo)
+}
+
+func TestLoadAndDelete(t *testing.T) {
+	m := &SafeRepoMap{}
+	key := repository.RepositoryKey{Name: "test-repo"}
+
+	// LoadAndDelete non-existent key
+	repo, ok := m.LoadAndDelete(key)
+	assert.False(t, ok)
+	assert.Nil(t, repo)
+
+	// Create a repo
+	m.LoadOrCreate(key, func() (repository.Repository, error) {
+		return &mockrepository.MockRepository{}, nil
+	})
+
+	// LoadAndDelete existing key
+	repo, ok = m.LoadAndDelete(key)
+	assert.True(t, ok)
+	assert.NotNil(t, repo)
+
+	// Verify it's deleted
+	repo, ok = m.Load(key)
+	assert.False(t, ok)
+}
+
+func TestRange(t *testing.T) {
+	m := &SafeRepoMap{}
+
+	// Create multiple repos
+	keys := []repository.RepositoryKey{
+		{Name: "repo1"},
+		{Name: "repo2"},
+		{Name: "repo3"},
+	}
+
+	for _, key := range keys {
+		m.LoadOrCreate(key, func() (repository.Repository, error) {
+			return &mockrepository.MockRepository{}, nil
+		})
+	}
+
+	// Range over all entries
+	var count int
+	m.Range(func(key, value any) bool {
+		count++
+		return true
+	})
+
+	assert.Equal(t, len(keys), count)
+
+	// Test early termination
+	count = 0
+	m.Range(func(key, value any) bool {
+		count++
+		return count < 2 // Stop after 2 iterations
+	})
+
+	assert.Equal(t, 2, count)
+}

pkg/externalrepo/git/cachedir_pool.go

@@ -42,17 +42,30 @@ var globalDirectoryPool = &DirectoryPool{
 
 // GetOrCreateSharedRepository safely initializes or reuses a cached git directory
 func (p *DirectoryPool) GetOrCreateSharedRepository(dir, reponame string) (*SharedDirectory, error) {
+	// Fast path: check if directory already exists
+	if sharedDir, exists := p.directories.Load(dir); exists {
+		p.mutex.Lock()
+		shared := sharedDir.(*SharedDirectory)
+		shared.refCount++
+		klog.V(2).Infof("Repo %s is reusing shared directory %s, refCount now: %d", reponame, dir, shared.refCount)
+		p.mutex.Unlock()
+		return shared, nil
+	}
+
+	// Slow path: create new directory with write lock - needed to avoid race condition during parallel operations
 	p.mutex.Lock()
 	defer p.mutex.Unlock()
 
-	// Check if directory already exists
+	// Double-check after acquiring lock
 	if sharedDir, exists := p.directories.Load(dir); exists {
 		shared := sharedDir.(*SharedDirectory)
 		shared.refCount++
 		klog.V(2).Infof("Repo %s is reusing shared directory %s, refCount now: %d", reponame, dir, shared.refCount)
 		return shared, nil
 	}
 
+	klog.V(2).Infof("Repo %s initializing shared directory %s", reponame, dir)
+
 	// Initialize repository safely
 	var repo *git.Repository
 	if fi, err := os.Stat(dir); err != nil {
@@ -61,15 +74,22 @@ func (p *DirectoryPool) GetOrCreateSharedRepository(dir, reponame string) (*Shar
 		}
 		repo, err = initEmptyRepository(dir)
 		if err != nil {
+			if removeErr := os.RemoveAll(dir); removeErr != nil {
+				klog.Errorf("Failed to remove partially created directory %s: %v", dir, removeErr)
+			}
 			return nil, err
 		}
 	} else if !fi.IsDir() {
 		return nil, fmt.Errorf("cache location %q is not a directory", dir)
 	} else {
 		repo, err = openRepository(dir)
 		if err != nil {
-			klog.Errorf("Failed to open repository %s: %v", dir, err)
-			return nil, fmt.Errorf("open of cached git directory failed in gogit (check the local git cache): %w", err)
+			if removeErr := os.RemoveAll(dir); removeErr != nil {
+				klog.Errorf("Failed to open repository %s: %v (also failed to remove corrupted directory: %v)", dir, err, removeErr)
+				return nil, fmt.Errorf("open of cached git directory failed in gogit (check the local git cache): %w", err)
+			}
+			klog.Infof("Removed corrupted cached directory %s, will retry on next attempt", dir)
+			return nil, fmt.Errorf("open of cached git directory failed in gogit (corrupted cache was removed): %w", err)
 		}
 	}
 
@@ -118,7 +138,7 @@ func (s *SharedDirectory) WithLock(fn func(*git.Repository) error) error {
 	return fn(s.repo)
 }
 
-// WithLock executes function with exclusive access to the cached git directory
+// WithRLock executes function with read-only access to the cached git directory
 func (s *SharedDirectory) WithRLock(fn func(*git.Repository) error) error {
 	s.mutex.RLock()
 	defer s.mutex.RUnlock()