Provisioning: Improve connection deletion error handling with undelete

When finalizer removal fails, the connection is now "undeleted" by removing
the DeletionTimestamp. This prevents connections from being stuck in deletion
state and allows users to retry deletion later.

Additionally:
- Expand retry logic to handle more transient errors (not just ServiceUnavailable)
- Add isTransientError helper to detect retriable errors
- Add comprehensive tests for undelete behavior and transient error detection

This ensures that if the controller cannot remove the finalizer due to
transient errors (network issues, API timeouts, etc.), the connection returns
to normal state rather than remaining stuck in deletion.

apps/provisioning/pkg/connection/finalizers.go

@@ -0,0 +1,4 @@
+package connection
+
+// BlockDeletionFinalizer prevents deletion of connections while repositories reference them
+const BlockDeletionFinalizer = "block-deletion-while-repositories-exist"

pkg/apiserver/registry/generic/storage.go

@@ -1,26 +1,58 @@
 package generic
 
 import (
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apiserver/pkg/registry/generic"
 	"k8s.io/apiserver/pkg/registry/generic/registry"
+	"k8s.io/apiserver/pkg/storage"
 
 	"github.com/grafana/grafana/pkg/apimachinery/utils"
 )
 
+// SelectableFieldsOptions allows customizing field selector behavior for a resource.
+type SelectableFieldsOptions struct {
+	// GetAttrs returns labels and fields for the object.
+	// If nil, the default GetAttrs is used which only exposes metadata.name.
+	GetAttrs func(obj runtime.Object) (labels.Set, fields.Set, error)
+}
+
 func NewRegistryStore(scheme *runtime.Scheme, resourceInfo utils.ResourceInfo, optsGetter generic.RESTOptionsGetter) (*registry.Store, error) {
+	return NewRegistryStoreWithSelectableFields(scheme, resourceInfo, optsGetter, SelectableFieldsOptions{})
+}
+
+// NewRegistryStoreWithSelectableFields creates a registry store with custom selectable fields support.
+// Use this when you need to filter resources by custom fields like spec.connection.name.
+func NewRegistryStoreWithSelectableFields(scheme *runtime.Scheme, resourceInfo utils.ResourceInfo, optsGetter generic.RESTOptionsGetter, fieldOpts SelectableFieldsOptions) (*registry.Store, error) {
 	gv := resourceInfo.GroupVersion()
 	gv.Version = runtime.APIVersionInternal
 	strategy := NewStrategy(scheme, gv)
 	if resourceInfo.IsClusterScoped() {
 		strategy = strategy.WithClusterScope()
 	}
+
+	// Use custom GetAttrs if provided, otherwise use default
+	attrFunc := GetAttrs
+	predicateFunc := Matcher
+	if fieldOpts.GetAttrs != nil {
+		attrFunc = fieldOpts.GetAttrs
+		// Create a matcher that uses the custom GetAttrs
+		predicateFunc = func(label labels.Selector, field fields.Selector) storage.SelectionPredicate {
+			return storage.SelectionPredicate{
+				Label:    label,
+				Field:    field,
+				GetAttrs: attrFunc,
+			}
+		}
+	}
+
 	store := &registry.Store{
 		NewFunc:                   resourceInfo.NewFunc,
 		NewListFunc:               resourceInfo.NewListFunc,
 		KeyRootFunc:               KeyRootFunc(resourceInfo.GroupResource()),
 		KeyFunc:                   NamespaceKeyFunc(resourceInfo.GroupResource()),
-		PredicateFunc:             Matcher,
+		PredicateFunc:             predicateFunc,
 		DefaultQualifiedResource:  resourceInfo.GroupResource(),
 		SingularQualifiedResource: resourceInfo.SingularGroupResource(),
 		TableConvertor:            resourceInfo.TableConverter(),
@@ -28,7 +60,7 @@ func NewRegistryStore(scheme *runtime.Scheme, resourceInfo utils.ResourceInfo, o
 		UpdateStrategy:            strategy,
 		DeleteStrategy:            strategy,
 	}
-	options := &generic.StoreOptions{RESTOptions: optsGetter, AttrFunc: GetAttrs}
+	options := &generic.StoreOptions{RESTOptions: optsGetter, AttrFunc: attrFunc}
 	if err := store.CompleteWithOptions(options); err != nil {
 		return nil, err
 	}

pkg/registry/apis/provisioning/controller/connection.go

@@ -4,9 +4,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net"
+	"strings"
 	"time"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
@@ -14,9 +20,11 @@ import (
 
 	"github.com/grafana/grafana-app-sdk/logging"
 	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+	connectionvalidation "github.com/grafana/grafana/apps/provisioning/pkg/connection"
 	client "github.com/grafana/grafana/apps/provisioning/pkg/generated/clientset/versioned/typed/provisioning/v0alpha1"
 	informer "github.com/grafana/grafana/apps/provisioning/pkg/generated/informers/externalversions/provisioning/v0alpha1"
 	listers "github.com/grafana/grafana/apps/provisioning/pkg/generated/listers/provisioning/v0alpha1"
+	"k8s.io/apimachinery/pkg/fields"
 )
 
 const connectionLoggerName = "provisioning-connection-controller"
@@ -41,6 +49,11 @@ type ConnectionStatusPatcher interface {
 	Patch(ctx context.Context, conn *provisioning.Connection, patchOperations ...map[string]interface{}) error
 }
 
+// RepositoryLister interface for listing repositories
+type RepositoryLister interface {
+	List(ctx context.Context, options *internalversion.ListOptions) (runtime.Object, error)
+}
+
 // ConnectionController controls Connection resources.
 type ConnectionController struct {
 	client     client.ProvisioningV0alpha1Interface
@@ -49,6 +62,7 @@ type ConnectionController struct {
 	logger     logging.Logger
 
 	statusPatcher ConnectionStatusPatcher
+	repoLister    RepositoryLister
 
 	queue workqueue.TypedRateLimitingInterface[*connectionQueueItem]
 }
@@ -58,6 +72,7 @@ func NewConnectionController(
 	provisioningClient client.ProvisioningV0alpha1Interface,
 	connInformer informer.ConnectionInformer,
 	statusPatcher ConnectionStatusPatcher,
+	repoLister RepositoryLister,
 ) (*ConnectionController, error) {
 	cc := &ConnectionController{
 		client:     provisioningClient,
@@ -70,6 +85,7 @@ func NewConnectionController(
 			},
 		),
 		statusPatcher: statusPatcher,
+		repoLister:    repoLister,
 		logger:        logging.DefaultLogger.With("logger", connectionLoggerName),
 	}
 
@@ -141,13 +157,14 @@ func (cc *ConnectionController) processNextWorkItem(ctx context.Context) bool {
 		return true
 	}
 
-	if !apierrors.IsServiceUnavailable(err) {
-		logger.Info("ConnectionController will not retry")
+	// Check if error is transient and should be retried
+	if !isTransientError(err) {
+		logger.Info("ConnectionController will not retry (non-transient error)")
 		cc.queue.Forget(item)
 		return true
 	}
 
-	logger.Info("ConnectionController will retry as service is unavailable")
+	logger.Info("ConnectionController will retry (transient error)")
 	utilruntime.HandleError(fmt.Errorf("%v failed with: %v", item, err))
 	cc.queue.AddRateLimited(item)
 
@@ -171,10 +188,9 @@ func (cc *ConnectionController) process(ctx context.Context, item *connectionQue
 		return err
 	}
 
-	// Skip if being deleted
+	// Handle deletion if being deleted
 	if conn.DeletionTimestamp != nil {
-		logger.Info("connection is being deleted, skipping")
-		return nil
+		return cc.handleDelete(ctx, conn)
 	}
 
 	hasSpecChanged := conn.Generation != conn.Status.ObservedGeneration
@@ -229,6 +245,147 @@ func (cc *ConnectionController) process(ctx context.Context, item *connectionQue
 	return nil
 }
 
+func (cc *ConnectionController) handleDelete(ctx context.Context, conn *provisioning.Connection) error {
+	logger := logging.FromContext(ctx)
+	logger.Info("handle connection delete")
+
+	// Check if finalizer is present
+	hasFinalizer := false
+	for _, f := range conn.Finalizers {
+		if f == connectionvalidation.BlockDeletionFinalizer {
+			hasFinalizer = true
+			break
+		}
+	}
+
+	if !hasFinalizer {
+		logger.Info("no finalizer to process")
+		return nil
+	}
+
+	// Check if any repositories reference this connection using field selector
+	fieldSelector := fields.OneTermEqualSelector("spec.connection.name", conn.Name)
+	var allRepos []provisioning.Repository
+	continueToken := ""
+	var err error
+
+	for {
+		var obj runtime.Object
+		obj, err = cc.repoLister.List(ctx, &internalversion.ListOptions{
+			Limit:         100,
+			Continue:      continueToken,
+			FieldSelector: fieldSelector,
+		})
+		if err != nil {
+			logger.Error("failed to check for connected repositories", "error", err)
+			return fmt.Errorf("check for connected repositories: %w", err)
+		}
+
+		repositoryList, ok := obj.(*provisioning.RepositoryList)
+		if !ok {
+			logger.Error("expected repository list", "type", fmt.Sprintf("%T", obj))
+			return fmt.Errorf("expected repository list, got %T", obj)
+		}
+
+		allRepos = append(allRepos, repositoryList.Items...)
+
+		continueToken = repositoryList.GetContinue()
+		if continueToken == "" {
+			break
+		}
+	}
+
+	if len(allRepos) > 0 {
+		repoNames := make([]string, 0, len(allRepos))
+		for _, repo := range allRepos {
+			repoNames = append(repoNames, repo.Name)
+		}
+		logger.Info("cannot delete connection while repositories reference it", "repositories", repoNames)
+		// Don't remove finalizer - this will prevent deletion
+		// The connection will remain in deletion state until repositories are removed
+		return fmt.Errorf("cannot delete connection while repositories are using it: %s", strings.Join(repoNames, ", "))
+	}
+
+	// No repositories reference this connection, remove finalizer to allow deletion
+	logger.Info("no repositories reference connection, removing finalizer")
+	_, err = cc.client.Connections(conn.GetNamespace()).
+		Patch(ctx, conn.Name, types.JSONPatchType, []byte(`[
+			{ "op": "remove", "path": "/metadata/finalizers" }
+		]`), metav1.PatchOptions{
+			FieldManager: "provisioning-connection-controller",
+		})
+	if err != nil {
+		// If we can't remove the finalizer, undelete the connection so it can be retried later
+		// This prevents the connection from being stuck in deletion state
+		logger.Error("failed to remove finalizer, undeleting connection", "error", err)
+		undeleteErr := cc.undeleteConnection(ctx, conn, err)
+		if undeleteErr != nil {
+			return fmt.Errorf("remove finalizer: %w; failed to undelete: %w", err, undeleteErr)
+		}
+		return fmt.Errorf("remove finalizer: %w (connection has been undeleted, deletion can be retried)", err)
+	}
+
+	return nil
+}
+
+// undeleteConnection removes the DeletionTimestamp to "undelete" the connection
+// This is used when finalizer removal fails, allowing the deletion to be retried later
+func (cc *ConnectionController) undeleteConnection(ctx context.Context, conn *provisioning.Connection, originalErr error) error {
+	logger := logging.FromContext(ctx)
+	logger.Info("undeleting connection due to finalizer removal failure", "error", originalErr.Error())
+
+	// Remove DeletionTimestamp by patching it to null
+	_, err := cc.client.Connections(conn.GetNamespace()).
+		Patch(ctx, conn.Name, types.JSONPatchType, []byte(`[
+			{ "op": "remove", "path": "/metadata/deletionTimestamp" }
+		]`), metav1.PatchOptions{
+			FieldManager: "provisioning-connection-controller",
+		})
+	if err != nil {
+		logger.Error("failed to undelete connection", "error", err)
+		return fmt.Errorf("undelete connection: %w", err)
+	}
+
+	logger.Info("connection undeleted successfully, deletion can be retried")
+	return nil
+}
+
+// isTransientError determines if an error is transient and should be retried
+func isTransientError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	// Check for Kubernetes API transient errors
+	if apierrors.IsServiceUnavailable(err) {
+		return true
+	}
+	if apierrors.IsServerTimeout(err) {
+		return true
+	}
+	if apierrors.IsTooManyRequests(err) {
+		return true
+	}
+	if apierrors.IsInternalError(err) {
+		return true
+	}
+	if apierrors.IsTimeout(err) {
+		return true
+	}
+
+	// Check for network errors
+	var netErr net.Error
+	if errors.As(err, &netErr) {
+		if netErr.Timeout() {
+			return true
+		}
+	}
+
+	// Check for connection errors
+	var opErr *net.OpError
+	return errors.As(err, &opErr)
+}
+
 // shouldCheckHealth determines if a connection health check should be performed.
 func (cc *ConnectionController) shouldCheckHealth(conn *provisioning.Connection) bool {
 	// If the connection has been updated, always check health

pkg/registry/apis/provisioning/controller/connection_test.go

@@ -1,13 +1,28 @@
 package controller
 
 import (
+	"context"
+	"errors"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/internalversion"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/rest"
 
 	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+	connectionvalidation "github.com/grafana/grafana/apps/provisioning/pkg/connection"
+	applyconfiguration "github.com/grafana/grafana/apps/provisioning/pkg/generated/applyconfiguration/provisioning/v0alpha1"
+	client "github.com/grafana/grafana/apps/provisioning/pkg/generated/clientset/versioned/typed/provisioning/v0alpha1"
 )
 
 func TestConnectionController_shouldCheckHealth(t *testing.T) {
@@ -285,3 +300,398 @@ func TestConnectionController_processNextWorkItem(t *testing.T) {
 		assert.NotNil(t, cc)
 	})
 }
+
+// mockRepositoryLister is a mock implementation of RepositoryLister for testing
+type mockRepositoryLister struct {
+	mock.Mock
+}
+
+func (m *mockRepositoryLister) List(ctx context.Context, options *internalversion.ListOptions) (runtime.Object, error) {
+	args := m.Called(ctx, options)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(runtime.Object), args.Error(1)
+}
+
+// mockConnectionInterface is a mock implementation of client.ConnectionInterface for testing
+type mockConnectionInterface struct {
+	mock.Mock
+}
+
+func (m *mockConnectionInterface) Create(ctx context.Context, connection *provisioning.Connection, opts metav1.CreateOptions) (*provisioning.Connection, error) {
+	args := m.Called(ctx, connection, opts)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*provisioning.Connection), args.Error(1)
+}
+
+func (m *mockConnectionInterface) Update(ctx context.Context, connection *provisioning.Connection, opts metav1.UpdateOptions) (*provisioning.Connection, error) {
+	args := m.Called(ctx, connection, opts)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*provisioning.Connection), args.Error(1)
+}
+
+func (m *mockConnectionInterface) UpdateStatus(ctx context.Context, connection *provisioning.Connection, opts metav1.UpdateOptions) (*provisioning.Connection, error) {
+	args := m.Called(ctx, connection, opts)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*provisioning.Connection), args.Error(1)
+}
+
+func (m *mockConnectionInterface) Delete(ctx context.Context, name string, opts metav1.DeleteOptions) error {
+	args := m.Called(ctx, name, opts)
+	return args.Error(0)
+}
+
+func (m *mockConnectionInterface) DeleteCollection(ctx context.Context, opts metav1.DeleteOptions, listOpts metav1.ListOptions) error {
+	args := m.Called(ctx, opts, listOpts)
+	return args.Error(0)
+}
+
+func (m *mockConnectionInterface) Get(ctx context.Context, name string, opts metav1.GetOptions) (*provisioning.Connection, error) {
+	args := m.Called(ctx, name, opts)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*provisioning.Connection), args.Error(1)
+}
+
+func (m *mockConnectionInterface) List(ctx context.Context, opts metav1.ListOptions) (*provisioning.ConnectionList, error) {
+	args := m.Called(ctx, opts)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*provisioning.ConnectionList), args.Error(1)
+}
+
+func (m *mockConnectionInterface) Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
+	args := m.Called(ctx, opts)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(watch.Interface), args.Error(1)
+}
+
+func (m *mockConnectionInterface) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts metav1.PatchOptions, subresources ...string) (*provisioning.Connection, error) {
+	args := m.Called(ctx, name, pt, data, opts, subresources)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*provisioning.Connection), args.Error(1)
+}
+
+func (m *mockConnectionInterface) Apply(ctx context.Context, connection *applyconfiguration.ConnectionApplyConfiguration, opts metav1.ApplyOptions) (*provisioning.Connection, error) {
+	args := m.Called(ctx, connection, opts)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*provisioning.Connection), args.Error(1)
+}
+
+func (m *mockConnectionInterface) ApplyStatus(ctx context.Context, connection *applyconfiguration.ConnectionApplyConfiguration, opts metav1.ApplyOptions) (*provisioning.Connection, error) {
+	args := m.Called(ctx, connection, opts)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*provisioning.Connection), args.Error(1)
+}
+
+// mockProvisioningV0alpha1InterfaceForConnections is a mock implementation of client.ProvisioningV0alpha1Interface for connection tests
+type mockProvisioningV0alpha1InterfaceForConnections struct {
+	mock.Mock
+	connections *mockConnectionInterface
+}
+
+func (m *mockProvisioningV0alpha1InterfaceForConnections) RESTClient() rest.Interface {
+	panic("not needed for testing")
+}
+
+func (m *mockProvisioningV0alpha1InterfaceForConnections) HistoricJobs(namespace string) client.HistoricJobInterface {
+	panic("not needed for testing")
+}
+
+func (m *mockProvisioningV0alpha1InterfaceForConnections) Jobs(namespace string) client.JobInterface {
+	panic("not needed for testing")
+}
+
+func (m *mockProvisioningV0alpha1InterfaceForConnections) Connections(namespace string) client.ConnectionInterface {
+	return m.connections
+}
+
+func (m *mockProvisioningV0alpha1InterfaceForConnections) Repositories(namespace string) client.RepositoryInterface {
+	panic("not needed for testing")
+}
+
+func TestConnectionController_handleDelete(t *testing.T) {
+	ctx := context.Background()
+
+	tests := []struct {
+		name                   string
+		connection             *provisioning.Connection
+		repoListerSetup        func(*mockRepositoryLister)
+		connectionSetup        func(*mockConnectionInterface)
+		expectedError          string
+		expectFinalizerRemoved bool
+	}{
+		{
+			name: "no finalizer present, should return nil",
+			connection: &provisioning.Connection{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "test-conn",
+					Namespace:         "default",
+					DeletionTimestamp: &metav1.Time{Time: time.Now()},
+					Finalizers:        []string{},
+				},
+			},
+			repoListerSetup:        func(m *mockRepositoryLister) {},
+			connectionSetup:        func(m *mockConnectionInterface) {},
+			expectedError:          "",
+			expectFinalizerRemoved: false,
+		},
+		{
+			name: "finalizer present but repositories exist, should block deletion",
+			connection: &provisioning.Connection{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "test-conn",
+					Namespace:         "default",
+					DeletionTimestamp: &metav1.Time{Time: time.Now()},
+					Finalizers:        []string{connectionvalidation.BlockDeletionFinalizer},
+				},
+			},
+			repoListerSetup: func(m *mockRepositoryLister) {
+				m.On("List", ctx, mock.MatchedBy(func(opts *internalversion.ListOptions) bool {
+					return opts.FieldSelector != nil && opts.FieldSelector.String() == "spec.connection.name=test-conn"
+				})).Return(&provisioning.RepositoryList{
+					Items: []provisioning.Repository{
+						{
+							ObjectMeta: metav1.ObjectMeta{Name: "repo-1"},
+							Spec: provisioning.RepositorySpec{
+								Connection: &provisioning.ConnectionInfo{Name: "test-conn"},
+							},
+						},
+						{
+							ObjectMeta: metav1.ObjectMeta{Name: "repo-2"},
+							Spec: provisioning.RepositorySpec{
+								Connection: &provisioning.ConnectionInfo{Name: "test-conn"},
+							},
+						},
+					},
+				}, nil)
+			},
+			connectionSetup:        func(m *mockConnectionInterface) {},
+			expectedError:          "cannot delete connection while repositories are using it: repo-1, repo-2",
+			expectFinalizerRemoved: false,
+		},
+		{
+			name: "finalizer present and no repositories, should remove finalizer",
+			connection: &provisioning.Connection{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "test-conn",
+					Namespace:         "default",
+					DeletionTimestamp: &metav1.Time{Time: time.Now()},
+					Finalizers:        []string{connectionvalidation.BlockDeletionFinalizer},
+				},
+			},
+			repoListerSetup: func(m *mockRepositoryLister) {
+				m.On("List", ctx, mock.MatchedBy(func(opts *internalversion.ListOptions) bool {
+					return opts.FieldSelector != nil && opts.FieldSelector.String() == "spec.connection.name=test-conn"
+				})).Return(&provisioning.RepositoryList{
+					Items: []provisioning.Repository{},
+				}, nil)
+			},
+			connectionSetup: func(m *mockConnectionInterface) {
+				m.On("Patch", ctx, "test-conn", types.JSONPatchType, mock.Anything, metav1.PatchOptions{
+					FieldManager: "provisioning-connection-controller",
+				}, mock.Anything).Return(&provisioning.Connection{}, nil)
+			},
+			expectedError:          "",
+			expectFinalizerRemoved: true,
+		},
+		{
+			name: "error checking repositories, should return error",
+			connection: &provisioning.Connection{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "test-conn",
+					Namespace:         "default",
+					DeletionTimestamp: &metav1.Time{Time: time.Now()},
+					Finalizers:        []string{connectionvalidation.BlockDeletionFinalizer},
+				},
+			},
+			repoListerSetup: func(m *mockRepositoryLister) {
+				m.On("List", ctx, mock.Anything).Return(nil, errors.New("list error"))
+			},
+			connectionSetup:        func(m *mockConnectionInterface) {},
+			expectedError:          "check for connected repositories: list error",
+			expectFinalizerRemoved: false,
+		},
+		{
+			name: "error removing finalizer, should undelete connection",
+			connection: &provisioning.Connection{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "test-conn",
+					Namespace:         "default",
+					DeletionTimestamp: &metav1.Time{Time: time.Now()},
+					Finalizers:        []string{connectionvalidation.BlockDeletionFinalizer},
+				},
+			},
+			repoListerSetup: func(m *mockRepositoryLister) {
+				m.On("List", ctx, mock.Anything).Return(&provisioning.RepositoryList{
+					Items: []provisioning.Repository{},
+				}, nil)
+			},
+			connectionSetup: func(m *mockConnectionInterface) {
+				// First patch fails (remove finalizer)
+				m.On("Patch", ctx, "test-conn", types.JSONPatchType, mock.MatchedBy(func(data []byte) bool {
+					return string(data) == `[
+			{ "op": "remove", "path": "/metadata/finalizers" }
+		]`
+				}), metav1.PatchOptions{
+					FieldManager: "provisioning-connection-controller",
+				}, mock.Anything).Return(nil, errors.New("patch error")).Once()
+				// Second patch succeeds (undelete - remove DeletionTimestamp)
+				m.On("Patch", ctx, "test-conn", types.JSONPatchType, mock.MatchedBy(func(data []byte) bool {
+					return string(data) == `[
+			{ "op": "remove", "path": "/metadata/deletionTimestamp" }
+		]`
+				}), metav1.PatchOptions{
+					FieldManager: "provisioning-connection-controller",
+				}, mock.Anything).Return(&provisioning.Connection{}, nil).Once()
+			},
+			expectedError:          "remove finalizer: patch error (connection has been undeleted, deletion can be retried)",
+			expectFinalizerRemoved: false,
+		},
+		{
+			name: "pagination handled correctly",
+			connection: &provisioning.Connection{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "test-conn",
+					Namespace:         "default",
+					DeletionTimestamp: &metav1.Time{Time: time.Now()},
+					Finalizers:        []string{connectionvalidation.BlockDeletionFinalizer},
+				},
+			},
+			repoListerSetup: func(m *mockRepositoryLister) {
+				// First call returns empty with continue token (testing pagination even when empty)
+				m.On("List", ctx, mock.MatchedBy(func(opts *internalversion.ListOptions) bool {
+					return opts.Continue == ""
+				})).Return(&provisioning.RepositoryList{
+					Items:    []provisioning.Repository{},
+					ListMeta: metav1.ListMeta{Continue: "continue-token"},
+				}, nil)
+				// Second call returns empty with no continue token
+				m.On("List", ctx, mock.MatchedBy(func(opts *internalversion.ListOptions) bool {
+					return opts.Continue == "continue-token"
+				})).Return(&provisioning.RepositoryList{
+					Items: []provisioning.Repository{},
+				}, nil)
+			},
+			connectionSetup: func(m *mockConnectionInterface) {
+				m.On("Patch", ctx, "test-conn", types.JSONPatchType, mock.Anything, metav1.PatchOptions{
+					FieldManager: "provisioning-connection-controller",
+				}, mock.Anything).Return(&provisioning.Connection{}, nil)
+			},
+			expectedError:          "",
+			expectFinalizerRemoved: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			repoLister := new(mockRepositoryLister)
+			connInterface := new(mockConnectionInterface)
+			client := &mockProvisioningV0alpha1InterfaceForConnections{connections: connInterface}
+
+			tt.repoListerSetup(repoLister)
+			tt.connectionSetup(connInterface)
+
+			cc := &ConnectionController{
+				client:     client,
+				repoLister: repoLister,
+				logger:     nil, // logger is optional for testing
+			}
+
+			err := cc.handleDelete(ctx, tt.connection)
+
+			if tt.expectedError != "" {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tt.expectedError)
+			} else {
+				require.NoError(t, err)
+			}
+
+			if tt.expectFinalizerRemoved {
+				connInterface.AssertCalled(t, "Patch", ctx, "test-conn", types.JSONPatchType, mock.Anything, metav1.PatchOptions{
+					FieldManager: "provisioning-connection-controller",
+				}, mock.Anything)
+			} else if tt.expectedError != "" && strings.Contains(tt.expectedError, "undeleted") {
+				// For undelete case, we expect both patches to be called (remove finalizer fails, then undelete succeeds)
+				connInterface.AssertNumberOfCalls(t, "Patch", 2)
+			}
+			// For other error cases (repositories exist), no successful patch should occur
+
+			repoLister.AssertExpectations(t)
+			connInterface.AssertExpectations(t)
+		})
+	}
+}
+
+func TestIsTransientError(t *testing.T) {
+	tests := []struct {
+		name     string
+		err      error
+		expected bool
+	}{
+		{
+			name:     "nil error",
+			err:      nil,
+			expected: false,
+		},
+		{
+			name:     "service unavailable",
+			err:      apierrors.NewServiceUnavailable("service unavailable"),
+			expected: true,
+		},
+		{
+			name:     "server timeout",
+			err:      apierrors.NewServerTimeout(schema.GroupResource{}, "operation", 0),
+			expected: true,
+		},
+		{
+			name:     "too many requests",
+			err:      apierrors.NewTooManyRequests("too many requests", 0),
+			expected: true,
+		},
+		{
+			name:     "internal error",
+			err:      apierrors.NewInternalError(errors.New("internal error")),
+			expected: true,
+		},
+		{
+			name:     "not found error",
+			err:      apierrors.NewNotFound(schema.GroupResource{}, "resource"),
+			expected: false,
+		},
+		{
+			name:     "forbidden error",
+			err:      apierrors.NewForbidden(schema.GroupResource{}, "resource", errors.New("forbidden")),
+			expected: false,
+		},
+		{
+			name:     "generic error",
+			err:      errors.New("generic error"),
+			expected: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := isTransientError(tt.err)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}

pkg/registry/apis/provisioning/register.go

@@ -559,6 +559,22 @@ func (b *APIBuilder) InstallSchema(scheme *runtime.Scheme) error {
 		return err
 	}
 
+	// Register custom field label conversion for Repository to enable field selectors like spec.connection.name
+	err = scheme.AddFieldLabelConversionFunc(
+		provisioning.SchemeGroupVersion.WithKind("Repository"),
+		func(label, value string) (string, string, error) {
+			switch label {
+			case "metadata.name", "metadata.namespace", "spec.connection.name":
+				return label, value, nil
+			default:
+				return "", "", fmt.Errorf("field label not supported for Repository: %s", label)
+			}
+		},
+	)
+	if err != nil {
+		return err
+	}
+
 	metav1.AddToGroupVersion(scheme, provisioning.SchemeGroupVersion)
 	// Only 1 version (for now?)
 	return scheme.SetVersionPriority(provisioning.SchemeGroupVersion)
@@ -569,10 +585,19 @@ func (b *APIBuilder) AllowedV0Alpha1Resources() []string {
 }
 
 func (b *APIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserver.APIGroupInfo, opts builder.APIGroupOptions) error {
-	repositoryStorage, err := grafanaregistry.NewRegistryStore(opts.Scheme, provisioning.RepositoryResourceInfo, opts.OptsGetter)
+	// Create repository storage with custom field selectors (e.g., spec.connection.name)
+	repositoryStorage, err := grafanaregistry.NewRegistryStoreWithSelectableFields(
+		opts.Scheme,
+		provisioning.RepositoryResourceInfo,
+		opts.OptsGetter,
+		grafanaregistry.SelectableFieldsOptions{
+			GetAttrs: RepositoryGetAttrs,
+		},
+	)
 	if err != nil {
 		return fmt.Errorf("failed to create repository storage: %w", err)
 	}
+
 	repositoryStatusStorage := grafanaregistry.NewRegistryStatusStore(opts.Scheme, repositoryStorage)
 	b.store = repositoryStorage
 
@@ -660,6 +685,12 @@ func (b *APIBuilder) Mutate(ctx context.Context, a admission.Attributes, o admis
 	// TODO: complete this as part of https://github.com/grafana/git-ui-sync-project/issues/700
 	c, ok := obj.(*provisioning.Connection)
 	if ok {
+		// Add finalizer on create to prevent deletion while repositories reference it
+		if len(c.Finalizers) == 0 && a.GetOperation() == admission.Create {
+			c.Finalizers = []string{
+				connectionvalidation.BlockDeletionFinalizer,
+			}
+		}
 		return connectionvalidation.MutateConnection(c)
 	}
 
@@ -695,7 +726,13 @@ func (b *APIBuilder) Mutate(ctx context.Context, a admission.Attributes, o admis
 // TODO: move logic to a more appropriate place. Probably controller/validation.go
 func (b *APIBuilder) Validate(ctx context.Context, a admission.Attributes, o admission.ObjectInterfaces) (err error) {
 	obj := a.GetObject()
-	if obj == nil || a.GetOperation() == admission.Connect || a.GetOperation() == admission.Delete {
+
+	// Handle Connection deletion - check for connected repositories
+	if a.GetOperation() == admission.Delete {
+		return b.validateDelete(ctx, a)
+	}
+
+	if obj == nil || a.GetOperation() == admission.Connect {
 		return nil // This is normal for sub-resource
 	}
 
@@ -775,6 +812,42 @@ func invalidRepositoryError(name string, list field.ErrorList) error {
 		name, list)
 }
 
+// validateDelete handles validation for delete operations
+func (b *APIBuilder) validateDelete(ctx context.Context, a admission.Attributes) error {
+	// Only validate Connection deletions
+	if a.GetResource().Resource != "connections" {
+		return nil
+	}
+
+	connectionName := a.GetName()
+	namespace := a.GetNamespace()
+
+	// Set namespace in context for the repository store query
+	ctx, _, err := identity.WithProvisioningIdentity(ctx, namespace)
+	if err != nil {
+		return apierrors.NewInternalError(fmt.Errorf("failed to set provisioning identity: %w", err))
+	}
+
+	repos, err := GetRepositoriesByConnection(ctx, b.store, connectionName)
+	if err != nil {
+		return apierrors.NewInternalError(fmt.Errorf("failed to check for connected repositories: %w", err))
+	}
+
+	if len(repos) > 0 {
+		repoNames := make([]string, 0, len(repos))
+		for _, repo := range repos {
+			repoNames = append(repoNames, repo.Name)
+		}
+		return apierrors.NewForbidden(
+			provisioning.ConnectionResourceInfo.GroupResource(),
+			connectionName,
+			fmt.Errorf("cannot delete connection while repositories are using it: %s", strings.Join(repoNames, ", ")),
+		)
+	}
+
+	return nil
+}
+
 func (b *APIBuilder) VerifyAgainstExistingRepositories(ctx context.Context, cfg *provisioning.Repository) *field.Error {
 	return VerifyAgainstExistingRepositories(ctx, b.store, cfg)
 }
@@ -947,6 +1020,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
 				b.GetClient(),
 				connInformer,
 				connStatusPatcher,
+				b.store,
 			)
 			if err != nil {
 				return err

pkg/registry/apis/provisioning/repository_fields.go

@@ -0,0 +1,44 @@
+package provisioning
+
+import (
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/registry/generic"
+
+	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+)
+
+// RepositoryToSelectableFields returns a field set that can be used for field selectors.
+// This includes standard metadata fields plus custom fields like spec.connection.name.
+func RepositoryToSelectableFields(obj *provisioning.Repository) fields.Set {
+	objectMetaFields := generic.ObjectMetaFieldsSet(&obj.ObjectMeta, true)
+
+	// Add custom selectable fields
+	specificFields := fields.Set{
+		"spec.connection.name": getConnectionName(obj),
+	}
+
+	return generic.MergeFieldsSets(objectMetaFields, specificFields)
+}
+
+// getConnectionName safely extracts the connection name from a Repository.
+// Returns empty string if no connection is configured.
+func getConnectionName(obj *provisioning.Repository) string {
+	if obj == nil || obj.Spec.Connection == nil {
+		return ""
+	}
+	return obj.Spec.Connection.Name
+}
+
+// RepositoryGetAttrs returns labels and fields of a Repository object.
+// This is used by the storage layer for filtering.
+func RepositoryGetAttrs(obj runtime.Object) (labels.Set, fields.Set, error) {
+	repo, ok := obj.(*provisioning.Repository)
+	if !ok {
+		return nil, nil, fmt.Errorf("given object is not a Repository")
+	}
+	return labels.Set(repo.Labels), RepositoryToSelectableFields(repo), nil
+}

pkg/registry/apis/provisioning/repository_fields_test.go

@@ -0,0 +1,184 @@
+package provisioning
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+)
+
+func TestGetConnectionName(t *testing.T) {
+	tests := []struct {
+		name     string
+		repo     *provisioning.Repository
+		expected string
+	}{
+		{
+			name:     "nil repository returns empty string",
+			repo:     nil,
+			expected: "",
+		},
+		{
+			name: "repository without connection returns empty string",
+			repo: &provisioning.Repository{
+				Spec: provisioning.RepositorySpec{
+					Title: "test-repo",
+				},
+			},
+			expected: "",
+		},
+		{
+			name: "repository with connection returns connection name",
+			repo: &provisioning.Repository{
+				Spec: provisioning.RepositorySpec{
+					Title: "test-repo",
+					Connection: &provisioning.ConnectionInfo{
+						Name: "my-connection",
+					},
+				},
+			},
+			expected: "my-connection",
+		},
+		{
+			name: "repository with empty connection name returns empty string",
+			repo: &provisioning.Repository{
+				Spec: provisioning.RepositorySpec{
+					Title: "test-repo",
+					Connection: &provisioning.ConnectionInfo{
+						Name: "",
+					},
+				},
+			},
+			expected: "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := getConnectionName(tt.repo)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}
+
+func TestRepositoryToSelectableFields(t *testing.T) {
+	tests := []struct {
+		name           string
+		repo           *provisioning.Repository
+		expectedFields map[string]string
+	}{
+		{
+			name: "includes metadata.name and metadata.namespace",
+			repo: &provisioning.Repository{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-repo",
+					Namespace: "default",
+				},
+				Spec: provisioning.RepositorySpec{
+					Title: "Test Repository",
+				},
+			},
+			expectedFields: map[string]string{
+				"metadata.name":        "test-repo",
+				"metadata.namespace":   "default",
+				"spec.connection.name": "",
+			},
+		},
+		{
+			name: "includes spec.connection.name when set",
+			repo: &provisioning.Repository{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "repo-with-connection",
+					Namespace: "org-1",
+				},
+				Spec: provisioning.RepositorySpec{
+					Title: "Repo With Connection",
+					Connection: &provisioning.ConnectionInfo{
+						Name: "github-connection",
+					},
+				},
+			},
+			expectedFields: map[string]string{
+				"metadata.name":        "repo-with-connection",
+				"metadata.namespace":   "org-1",
+				"spec.connection.name": "github-connection",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fields := RepositoryToSelectableFields(tt.repo)
+
+			for key, expectedValue := range tt.expectedFields {
+				actualValue, exists := fields[key]
+				assert.True(t, exists, "field %s should exist", key)
+				assert.Equal(t, expectedValue, actualValue, "field %s should have correct value", key)
+			}
+		})
+	}
+}
+
+func TestRepositoryGetAttrs(t *testing.T) {
+	t.Run("returns error for non-Repository object", func(t *testing.T) {
+		// Pass a different runtime.Object type instead of a Repository
+		connection := &provisioning.Connection{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "not-a-repository",
+			},
+		}
+		_, _, err := RepositoryGetAttrs(connection)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "not a Repository")
+	})
+
+	t.Run("returns labels and fields for valid Repository", func(t *testing.T) {
+		repo := &provisioning.Repository{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-repo",
+				Namespace: "default",
+				Labels: map[string]string{
+					"app": "grafana",
+					"env": "test",
+				},
+			},
+			Spec: provisioning.RepositorySpec{
+				Title: "Test Repository",
+				Connection: &provisioning.ConnectionInfo{
+					Name: "my-connection",
+				},
+			},
+		}
+
+		labels, fields, err := RepositoryGetAttrs(repo)
+		require.NoError(t, err)
+
+		// Check labels
+		assert.Equal(t, "grafana", labels["app"])
+		assert.Equal(t, "test", labels["env"])
+
+		// Check fields
+		assert.Equal(t, "test-repo", fields["metadata.name"])
+		assert.Equal(t, "default", fields["metadata.namespace"])
+		assert.Equal(t, "my-connection", fields["spec.connection.name"])
+	})
+
+	t.Run("returns empty connection name when not set", func(t *testing.T) {
+		repo := &provisioning.Repository{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-repo",
+				Namespace: "default",
+			},
+			Spec: provisioning.RepositorySpec{
+				Title: "Test Repository",
+			},
+		}
+
+		_, fields, err := RepositoryGetAttrs(repo)
+		require.NoError(t, err)
+		assert.Equal(t, "", fields["spec.connection.name"])
+	})
+}

pkg/registry/apis/provisioning/validation.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/apiserver/pkg/endpoints/request"
@@ -50,6 +51,39 @@ func GetRepositoriesInNamespace(ctx context.Context, store RepositoryLister) ([]
 	return allRepositories, nil
 }
 
+// GetRepositoriesByConnection retrieves all repositories that reference a specific connection
+func GetRepositoriesByConnection(ctx context.Context, store RepositoryLister, connectionName string) ([]provisioning.Repository, error) {
+	var allRepositories []provisioning.Repository
+	continueToken := ""
+
+	fieldSelector := fields.OneTermEqualSelector("spec.connection.name", connectionName)
+
+	for {
+		obj, err := store.List(ctx, &internalversion.ListOptions{
+			Limit:         100,
+			Continue:      continueToken,
+			FieldSelector: fieldSelector,
+		})
+		if err != nil {
+			return nil, err
+		}
+
+		repositoryList, ok := obj.(*provisioning.RepositoryList)
+		if !ok {
+			return nil, fmt.Errorf("expected repository list")
+		}
+
+		allRepositories = append(allRepositories, repositoryList.Items...)
+
+		continueToken = repositoryList.GetContinue()
+		if continueToken == "" {
+			break
+		}
+	}
+
+	return allRepositories, nil
+}
+
 // VerifyAgainstExistingRepositories validates a repository configuration against existing repositories
 func VerifyAgainstExistingRepositories(ctx context.Context, store RepositoryLister, cfg *provisioning.Repository) *field.Error {
 	ctx, _, err := identity.WithProvisioningIdentity(ctx, cfg.Namespace)

pkg/registry/apis/provisioning/validation_test.go

@@ -0,0 +1,200 @@
+package provisioning
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+)
+
+// mockRepositoryLister is a mock implementation of RepositoryLister for testing
+type mockRepositoryLister struct {
+	repositories []provisioning.Repository
+	listErr      error
+	// Track the field selector used in List calls
+	lastFieldSelector fields.Selector
+}
+
+func (m *mockRepositoryLister) List(ctx context.Context, options *internalversion.ListOptions) (runtime.Object, error) {
+	if m.listErr != nil {
+		return nil, m.listErr
+	}
+
+	// Store the field selector for verification
+	m.lastFieldSelector = options.FieldSelector
+
+	// Filter repositories based on field selector if present
+	filteredRepos := m.repositories
+	if options.FieldSelector != nil && !options.FieldSelector.Empty() {
+		filteredRepos = make([]provisioning.Repository, 0)
+		for _, repo := range m.repositories {
+			// Simulate field selector matching for spec.connection.name
+			repoFields := fields.Set{
+				"spec.connection.name": getRepoConnectionName(&repo),
+			}
+			if options.FieldSelector.Matches(repoFields) {
+				filteredRepos = append(filteredRepos, repo)
+			}
+		}
+	}
+
+	return &provisioning.RepositoryList{
+		Items: filteredRepos,
+	}, nil
+}
+
+func getRepoConnectionName(repo *provisioning.Repository) string {
+	if repo.Spec.Connection == nil {
+		return ""
+	}
+	return repo.Spec.Connection.Name
+}
+
+func TestGetRepositoriesByConnection(t *testing.T) {
+	tests := []struct {
+		name           string
+		repositories   []provisioning.Repository
+		connectionName string
+		expectedCount  int
+		expectedNames  []string
+		expectedErr    bool
+	}{
+		{
+			name:           "empty repository list returns empty",
+			repositories:   []provisioning.Repository{},
+			connectionName: "test-conn",
+			expectedCount:  0,
+			expectedNames:  []string{},
+		},
+		{
+			name: "finds single matching repository",
+			repositories: []provisioning.Repository{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "repo-1"},
+					Spec: provisioning.RepositorySpec{
+						Connection: &provisioning.ConnectionInfo{Name: "conn-a"},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "repo-2"},
+					Spec: provisioning.RepositorySpec{
+						Connection: &provisioning.ConnectionInfo{Name: "conn-b"},
+					},
+				},
+			},
+			connectionName: "conn-a",
+			expectedCount:  1,
+			expectedNames:  []string{"repo-1"},
+		},
+		{
+			name: "finds multiple matching repositories",
+			repositories: []provisioning.Repository{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "repo-1"},
+					Spec: provisioning.RepositorySpec{
+						Connection: &provisioning.ConnectionInfo{Name: "shared-conn"},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "repo-2"},
+					Spec: provisioning.RepositorySpec{
+						Connection: &provisioning.ConnectionInfo{Name: "shared-conn"},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "repo-3"},
+					Spec: provisioning.RepositorySpec{
+						Connection: &provisioning.ConnectionInfo{Name: "different-conn"},
+					},
+				},
+			},
+			connectionName: "shared-conn",
+			expectedCount:  2,
+			expectedNames:  []string{"repo-1", "repo-2"},
+		},
+		{
+			name: "no matches returns empty list",
+			repositories: []provisioning.Repository{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "repo-1"},
+					Spec: provisioning.RepositorySpec{
+						Connection: &provisioning.ConnectionInfo{Name: "conn-a"},
+					},
+				},
+			},
+			connectionName: "non-existent",
+			expectedCount:  0,
+			expectedNames:  []string{},
+		},
+		{
+			name: "empty connection name matches repos without connection",
+			repositories: []provisioning.Repository{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "repo-with-conn"},
+					Spec: provisioning.RepositorySpec{
+						Connection: &provisioning.ConnectionInfo{Name: "some-conn"},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "repo-without-conn"},
+					Spec: provisioning.RepositorySpec{
+						Connection: nil,
+					},
+				},
+			},
+			connectionName: "",
+			expectedCount:  1,
+			expectedNames:  []string{"repo-without-conn"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mock := &mockRepositoryLister{repositories: tt.repositories}
+			ctx := context.Background()
+
+			repos, err := GetRepositoriesByConnection(ctx, mock, tt.connectionName)
+
+			if tt.expectedErr {
+				require.Error(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Len(t, repos, tt.expectedCount)
+
+			// Verify the field selector was used
+			require.NotNil(t, mock.lastFieldSelector, "field selector should have been set")
+			expectedSelector := fields.OneTermEqualSelector("spec.connection.name", tt.connectionName)
+			assert.Equal(t, expectedSelector.String(), mock.lastFieldSelector.String())
+
+			// Verify the correct repositories were returned
+			actualNames := make([]string, len(repos))
+			for i, repo := range repos {
+				actualNames[i] = repo.Name
+			}
+			for _, expectedName := range tt.expectedNames {
+				assert.Contains(t, actualNames, expectedName)
+			}
+		})
+	}
+}
+
+func TestGetRepositoriesByConnection_ListError(t *testing.T) {
+	mock := &mockRepositoryLister{
+		listErr: assert.AnError,
+	}
+	ctx := context.Background()
+
+	repos, err := GetRepositoriesByConnection(ctx, mock, "any-conn")
+
+	require.Error(t, err)
+	assert.Nil(t, repos)
+}

pkg/tests/apis/provisioning/connection_test.go

@@ -559,3 +559,376 @@ func TestIntegrationConnectionController_HealthCheckUpdates(t *testing.T) {
 		assert.True(t, final.Status.Health.Healthy, "connection should remain healthy")
 	})
 }
+
+func TestIntegrationProvisioning_RepositoryFieldSelectorByConnection(t *testing.T) {
+	testutil.SkipIntegrationTestInShortMode(t)
+
+	helper := runGrafana(t)
+	ctx := context.Background()
+	createOptions := metav1.CreateOptions{FieldValidation: "Strict"}
+
+	// Create a connection first
+	connection := &unstructured.Unstructured{Object: map[string]any{
+		"apiVersion": "provisioning.grafana.app/v0alpha1",
+		"kind":       "Connection",
+		"metadata": map[string]any{
+			"name":      "test-conn-for-field-selector",
+			"namespace": "default",
+		},
+		"spec": map[string]any{
+			"type": "github",
+			"github": map[string]any{
+				"appID":          "123456",
+				"installationID": "789012",
+			},
+		},
+		"secure": map[string]any{
+			"privateKey": map[string]any{
+				"create": "test-private-key",
+			},
+		},
+	}}
+
+	_, err := helper.Connections.Resource.Create(ctx, connection, createOptions)
+	require.NoError(t, err, "failed to create connection")
+
+	t.Cleanup(func() {
+		// Clean up repositories first
+		_ = helper.Repositories.Resource.Delete(ctx, "repo-with-connection", metav1.DeleteOptions{})
+		_ = helper.Repositories.Resource.Delete(ctx, "repo-without-connection", metav1.DeleteOptions{})
+		_ = helper.Repositories.Resource.Delete(ctx, "repo-with-different-connection", metav1.DeleteOptions{})
+		// Then clean up the connection
+		_ = helper.Connections.Resource.Delete(ctx, "test-conn-for-field-selector", metav1.DeleteOptions{})
+	})
+
+	// Create a repository WITH the connection
+	repoWithConnection := &unstructured.Unstructured{Object: map[string]any{
+		"apiVersion": "provisioning.grafana.app/v0alpha1",
+		"kind":       "Repository",
+		"metadata": map[string]any{
+			"name":      "repo-with-connection",
+			"namespace": "default",
+		},
+		"spec": map[string]any{
+			"title": "Repo With Connection",
+			"type":  "local",
+			"sync": map[string]any{
+				"enabled": false,
+				"target":  "folder",
+			},
+			"local": map[string]any{
+				"path": helper.ProvisioningPath,
+			},
+			"connection": map[string]any{
+				"name": "test-conn-for-field-selector",
+			},
+		},
+	}}
+
+	_, err = helper.Repositories.Resource.Create(ctx, repoWithConnection, createOptions)
+	require.NoError(t, err, "failed to create repository with connection")
+
+	// Create a repository WITHOUT the connection
+	repoWithoutConnection := &unstructured.Unstructured{Object: map[string]any{
+		"apiVersion": "provisioning.grafana.app/v0alpha1",
+		"kind":       "Repository",
+		"metadata": map[string]any{
+			"name":      "repo-without-connection",
+			"namespace": "default",
+		},
+		"spec": map[string]any{
+			"title": "Repo Without Connection",
+			"type":  "local",
+			"sync": map[string]any{
+				"enabled": false,
+				"target":  "folder",
+			},
+			"local": map[string]any{
+				"path": helper.ProvisioningPath,
+			},
+		},
+	}}
+
+	_, err = helper.Repositories.Resource.Create(ctx, repoWithoutConnection, createOptions)
+	require.NoError(t, err, "failed to create repository without connection")
+
+	// Create a repository with a DIFFERENT connection name (non-existent)
+	repoWithDifferentConnection := &unstructured.Unstructured{Object: map[string]any{
+		"apiVersion": "provisioning.grafana.app/v0alpha1",
+		"kind":       "Repository",
+		"metadata": map[string]any{
+			"name":      "repo-with-different-connection",
+			"namespace": "default",
+		},
+		"spec": map[string]any{
+			"title": "Repo With Different Connection",
+			"type":  "local",
+			"sync": map[string]any{
+				"enabled": false,
+				"target":  "folder",
+			},
+			"local": map[string]any{
+				"path": helper.ProvisioningPath,
+			},
+			"connection": map[string]any{
+				"name": "some-other-connection",
+			},
+		},
+	}}
+
+	_, err = helper.Repositories.Resource.Create(ctx, repoWithDifferentConnection, createOptions)
+	require.NoError(t, err, "failed to create repository with different connection")
+
+	t.Run("filter repositories by spec.connection.name", func(t *testing.T) {
+		// List repositories with field selector for the specific connection
+		list, err := helper.Repositories.Resource.List(ctx, metav1.ListOptions{
+			FieldSelector: "spec.connection.name=test-conn-for-field-selector",
+		})
+		require.NoError(t, err, "failed to list repositories with field selector")
+
+		// Should only return the repository with the matching connection
+		assert.Len(t, list.Items, 1, "should return exactly one repository")
+		assert.Equal(t, "repo-with-connection", list.Items[0].GetName(), "should return the correct repository")
+	})
+
+	t.Run("filter repositories by non-existent connection returns empty", func(t *testing.T) {
+		// List repositories with field selector for a non-existent connection
+		list, err := helper.Repositories.Resource.List(ctx, metav1.ListOptions{
+			FieldSelector: "spec.connection.name=non-existent-connection",
+		})
+		require.NoError(t, err, "failed to list repositories with field selector")
+
+		// Should return empty list
+		assert.Len(t, list.Items, 0, "should return no repositories for non-existent connection")
+	})
+
+	t.Run("filter repositories by empty connection name", func(t *testing.T) {
+		// List repositories with field selector for empty connection (repos without connection)
+		list, err := helper.Repositories.Resource.List(ctx, metav1.ListOptions{
+			FieldSelector: "spec.connection.name=",
+		})
+		require.NoError(t, err, "failed to list repositories with empty connection field selector")
+
+		// Should return the repository without a connection
+		assert.Len(t, list.Items, 1, "should return exactly one repository without connection")
+		assert.Equal(t, "repo-without-connection", list.Items[0].GetName(), "should return the repository without connection")
+	})
+
+	t.Run("list all repositories without field selector", func(t *testing.T) {
+		// List all repositories without field selector
+		list, err := helper.Repositories.Resource.List(ctx, metav1.ListOptions{})
+		require.NoError(t, err, "failed to list all repositories")
+
+		// Should return all three repositories
+		assert.Len(t, list.Items, 3, "should return all three repositories")
+
+		names := make([]string, len(list.Items))
+		for i, item := range list.Items {
+			names[i] = item.GetName()
+		}
+		assert.Contains(t, names, "repo-with-connection")
+		assert.Contains(t, names, "repo-without-connection")
+		assert.Contains(t, names, "repo-with-different-connection")
+	})
+}
+
+func TestIntegrationProvisioning_ConnectionDeletionBlocking(t *testing.T) {
+	testutil.SkipIntegrationTestInShortMode(t)
+
+	helper := runGrafana(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	t.Cleanup(cancel)
+
+	createOptions := metav1.CreateOptions{}
+
+	// Create a connection for testing deletion blocking
+	connName := "test-conn-delete-blocking"
+	_, err := helper.Connections.Resource.Create(ctx, &unstructured.Unstructured{
+		Object: map[string]any{
+			"apiVersion": "provisioning.grafana.app/v0alpha1",
+			"kind":       "Connection",
+			"metadata": map[string]any{
+				"name":      connName,
+				"namespace": "default",
+			},
+			"spec": map[string]any{
+				"type": "github",
+				"github": map[string]any{
+					"appID":          "123456",
+					"installationID": "454545",
+				},
+			},
+			"secure": map[string]any{
+				"privateKey": map[string]any{
+					"create": "someSecret",
+				},
+			},
+		},
+	}, createOptions)
+	require.NoError(t, err, "failed to create test connection")
+
+	t.Run("delete connection without connected repositories succeeds", func(t *testing.T) {
+		// Create a connection that has no repositories
+		emptyConnName := "test-conn-no-repos"
+		_, err := helper.Connections.Resource.Create(ctx, &unstructured.Unstructured{
+			Object: map[string]any{
+				"apiVersion": "provisioning.grafana.app/v0alpha1",
+				"kind":       "Connection",
+				"metadata": map[string]any{
+					"name":      emptyConnName,
+					"namespace": "default",
+				},
+				"spec": map[string]any{
+					"type": "github",
+					"github": map[string]any{
+						"appID":          "123457",
+						"installationID": "454546",
+					},
+				},
+				"secure": map[string]any{
+					"privateKey": map[string]any{
+						"create": "someSecret",
+					},
+				},
+			},
+		}, createOptions)
+		require.NoError(t, err, "failed to create test connection without repos")
+
+		// Delete should succeed since no repositories reference it
+		err = helper.Connections.Resource.Delete(ctx, emptyConnName, metav1.DeleteOptions{})
+		require.NoError(t, err, "deleting connection without connected repositories should succeed")
+
+		// Verify the connection is deleted
+		_, err = helper.Connections.Resource.Get(ctx, emptyConnName, metav1.GetOptions{})
+		require.True(t, k8serrors.IsNotFound(err), "connection should be deleted")
+	})
+
+	t.Run("delete connection with connected repository fails", func(t *testing.T) {
+		// Create a repository that uses the connection
+		repoName := "repo-using-connection"
+		_, err := helper.Repositories.Resource.Create(ctx, &unstructured.Unstructured{
+			Object: map[string]any{
+				"apiVersion": "provisioning.grafana.app/v0alpha1",
+				"kind":       "Repository",
+				"metadata": map[string]any{
+					"name":      repoName,
+					"namespace": "default",
+				},
+				"spec": map[string]any{
+					"title": "Test Repository",
+					"type":  "local",
+					"sync": map[string]any{
+						"enabled": false,
+						"target":  "folder",
+					},
+					"local": map[string]any{
+						"path": helper.ProvisioningPath,
+					},
+					"connection": map[string]any{
+						"name": connName,
+					},
+				},
+			},
+		}, createOptions)
+		require.NoError(t, err, "failed to create repository using connection")
+
+		// Attempt to delete the connection - should fail
+		err = helper.Connections.Resource.Delete(ctx, connName, metav1.DeleteOptions{})
+		require.Error(t, err, "deleting connection with connected repository should fail")
+		require.True(t, k8serrors.IsForbidden(err), "error should be Forbidden, got: %v", err)
+		assert.Contains(t, err.Error(), repoName, "error should mention the connected repository name")
+		assert.Contains(t, err.Error(), "cannot delete connection while repositories are using it", "error should explain why deletion is blocked")
+
+		// Clean up: delete the repository first
+		err = helper.Repositories.Resource.Delete(ctx, repoName, metav1.DeleteOptions{})
+		require.NoError(t, err, "failed to delete test repository")
+
+		// Wait for the repository to be deleted
+		require.Eventually(t, func() bool {
+			_, err := helper.Repositories.Resource.Get(ctx, repoName, metav1.GetOptions{})
+			return k8serrors.IsNotFound(err)
+		}, 10*time.Second, 100*time.Millisecond, "repository should be deleted")
+	})
+
+	t.Run("delete connection after disconnecting repository succeeds", func(t *testing.T) {
+		// Now that the repository is deleted, the connection should be deletable
+		err = helper.Connections.Resource.Delete(ctx, connName, metav1.DeleteOptions{})
+		require.NoError(t, err, "deleting connection after removing connected repositories should succeed")
+
+		// Verify the connection is deleted
+		_, err = helper.Connections.Resource.Get(ctx, connName, metav1.GetOptions{})
+		require.True(t, k8serrors.IsNotFound(err), "connection should be deleted")
+	})
+
+	t.Run("delete connection with multiple connected repositories lists all", func(t *testing.T) {
+		// Create a new connection
+		multiConnName := "test-conn-multi-repos"
+		_, err := helper.Connections.Resource.Create(ctx, &unstructured.Unstructured{
+			Object: map[string]any{
+				"apiVersion": "provisioning.grafana.app/v0alpha1",
+				"kind":       "Connection",
+				"metadata": map[string]any{
+					"name":      multiConnName,
+					"namespace": "default",
+				},
+				"spec": map[string]any{
+					"type": "github",
+					"github": map[string]any{
+						"appID":          "123458",
+						"installationID": "454547",
+					},
+				},
+				"secure": map[string]any{
+					"privateKey": map[string]any{
+						"create": "someSecret",
+					},
+				},
+			},
+		}, createOptions)
+		require.NoError(t, err, "failed to create multi-repo test connection")
+
+		// Create multiple repositories using this connection
+		repoNames := []string{"multi-repo-1", "multi-repo-2"}
+		for _, repoName := range repoNames {
+			_, err := helper.Repositories.Resource.Create(ctx, &unstructured.Unstructured{
+				Object: map[string]any{
+					"apiVersion": "provisioning.grafana.app/v0alpha1",
+					"kind":       "Repository",
+					"metadata": map[string]any{
+						"name":      repoName,
+						"namespace": "default",
+					},
+					"spec": map[string]any{
+						"title": "Test Repository " + repoName,
+						"type":  "local",
+						"sync": map[string]any{
+							"enabled": false,
+							"target":  "folder",
+						},
+						"local": map[string]any{
+							"path": helper.ProvisioningPath,
+						},
+						"connection": map[string]any{
+							"name": multiConnName,
+						},
+					},
+				},
+			}, createOptions)
+			require.NoError(t, err, "failed to create repository %s", repoName)
+		}
+
+		// Attempt to delete - should fail and list all repos
+		err = helper.Connections.Resource.Delete(ctx, multiConnName, metav1.DeleteOptions{})
+		require.Error(t, err, "deleting connection with multiple repos should fail")
+		require.True(t, k8serrors.IsForbidden(err), "error should be Forbidden")
+		for _, repoName := range repoNames {
+			assert.Contains(t, err.Error(), repoName, "error should mention repository %s", repoName)
+		}
+
+		// Clean up
+		for _, repoName := range repoNames {
+			_ = helper.Repositories.Resource.Delete(ctx, repoName, metav1.DeleteOptions{})
+		}
+		_ = helper.Connections.Resource.Delete(ctx, multiConnName, metav1.DeleteOptions{})
+	})
+}