Implement hierarchical error handling for folder creation failures

This commit implements hierarchical error handling to improve sync robustness
when folder creation fails. Instead of failing the entire sync, the system now:

1. Tracks failed folder creations and automatically skips nested resources
2. Records skipped resources with FileActionIgnored (doesn't count toward error limits)
3. Allows other folder hierarchies to continue processing
4. Prevents folder deletion when child resource deletions fail

Key Changes:

- Add PathCreationError type to track which folder path failed
- Modify progress recorder to automatically detect and track failures via Record()
- Add IsNestedUnderFailedCreation() and HasFailedDeletionsUnder() checks
- Update full and incremental sync to skip nested resources after folder failures
- Deletions proceed even if parent folder creation failed (resource may exist from previous sync)
- FileActionIgnored results don't count toward error limits

Example behavior improvement:
Before: /monitoring folder creation fails → all nested resources fail → other folders never processed
After: /monitoring folder creation fails → nested resources ignored → /applications folder succeeds

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

pkg/registry/apis/provisioning/jobs/job_progress_recorder_mock.go

@@ -71,6 +71,98 @@ func (_c *MockJobProgressRecorder_Complete_Call) RunAndReturn(run func(context.C
 	return _c
 }
 
+// HasFailedDeletionsUnder provides a mock function with given fields: folderPath
+func (_m *MockJobProgressRecorder) HasFailedDeletionsUnder(folderPath string) bool {
+	ret := _m.Called(folderPath)
+
+	if len(ret) == 0 {
+		panic("no return value specified for HasFailedDeletionsUnder")
+	}
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(string) bool); ok {
+		r0 = rf(folderPath)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	return r0
+}
+
+// MockJobProgressRecorder_HasFailedDeletionsUnder_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'HasFailedDeletionsUnder'
+type MockJobProgressRecorder_HasFailedDeletionsUnder_Call struct {
+	*mock.Call
+}
+
+// HasFailedDeletionsUnder is a helper method to define mock.On call
+//   - folderPath string
+func (_e *MockJobProgressRecorder_Expecter) HasFailedDeletionsUnder(folderPath interface{}) *MockJobProgressRecorder_HasFailedDeletionsUnder_Call {
+	return &MockJobProgressRecorder_HasFailedDeletionsUnder_Call{Call: _e.mock.On("HasFailedDeletionsUnder", folderPath)}
+}
+
+func (_c *MockJobProgressRecorder_HasFailedDeletionsUnder_Call) Run(run func(folderPath string)) *MockJobProgressRecorder_HasFailedDeletionsUnder_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(string))
+	})
+	return _c
+}
+
+func (_c *MockJobProgressRecorder_HasFailedDeletionsUnder_Call) Return(_a0 bool) *MockJobProgressRecorder_HasFailedDeletionsUnder_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *MockJobProgressRecorder_HasFailedDeletionsUnder_Call) RunAndReturn(run func(string) bool) *MockJobProgressRecorder_HasFailedDeletionsUnder_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// IsNestedUnderFailedCreation provides a mock function with given fields: path
+func (_m *MockJobProgressRecorder) IsNestedUnderFailedCreation(path string) bool {
+	ret := _m.Called(path)
+
+	if len(ret) == 0 {
+		panic("no return value specified for IsNestedUnderFailedCreation")
+	}
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(string) bool); ok {
+		r0 = rf(path)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	return r0
+}
+
+// MockJobProgressRecorder_IsNestedUnderFailedCreation_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'IsNestedUnderFailedCreation'
+type MockJobProgressRecorder_IsNestedUnderFailedCreation_Call struct {
+	*mock.Call
+}
+
+// IsNestedUnderFailedCreation is a helper method to define mock.On call
+//   - path string
+func (_e *MockJobProgressRecorder_Expecter) IsNestedUnderFailedCreation(path interface{}) *MockJobProgressRecorder_IsNestedUnderFailedCreation_Call {
+	return &MockJobProgressRecorder_IsNestedUnderFailedCreation_Call{Call: _e.mock.On("IsNestedUnderFailedCreation", path)}
+}
+
+func (_c *MockJobProgressRecorder_IsNestedUnderFailedCreation_Call) Run(run func(path string)) *MockJobProgressRecorder_IsNestedUnderFailedCreation_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(string))
+	})
+	return _c
+}
+
+func (_c *MockJobProgressRecorder_IsNestedUnderFailedCreation_Call) Return(_a0 bool) *MockJobProgressRecorder_IsNestedUnderFailedCreation_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *MockJobProgressRecorder_IsNestedUnderFailedCreation_Call) RunAndReturn(run func(string) bool) *MockJobProgressRecorder_IsNestedUnderFailedCreation_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Record provides a mock function with given fields: ctx, result
 func (_m *MockJobProgressRecorder) Record(ctx context.Context, result JobResourceResult) {
 	_m.Called(ctx, result)

pkg/registry/apis/provisioning/jobs/progress.go

@@ -2,6 +2,7 @@ package jobs
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -9,6 +10,8 @@ import (
 	"github.com/grafana/grafana-app-sdk/logging"
 	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
 	"github.com/grafana/grafana/apps/provisioning/pkg/repository"
+	"github.com/grafana/grafana/apps/provisioning/pkg/safepath"
+	"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
 )
 
 // maybeNotifyProgress will only notify if a certain amount of time has passed
@@ -57,6 +60,8 @@ type jobProgressRecorder struct {
 	notifyImmediatelyFn ProgressFn
 	maybeNotifyFn       ProgressFn
 	summaries           map[string]*provisioning.JobResourceSummary
+	failedCreations     []string // Tracks folder paths that failed to be created
+	failedDeletions     []string // Tracks resource paths that failed to be deleted
 }
 
 func newJobProgressRecorder(ProgressFn ProgressFn) JobProgressRecorder {
@@ -83,10 +88,26 @@ func (r *jobProgressRecorder) Record(ctx context.Context, result JobResourceResu
 	if result.Error != nil {
 		shouldLogError = true
 		logErr = result.Error
-		if len(r.errors) < 20 {
-			r.errors = append(r.errors, result.Error.Error())
+
+		// Don't count ignored actions as errors in error count or error list
+		if result.Action != repository.FileActionIgnored {
+			if len(r.errors) < 20 {
+				r.errors = append(r.errors, result.Error.Error())
+			}
+			r.errorCount++
+		}
+
+		// Automatically track failed operations based on error type and action
+		// Check if this is a PathCreationError (folder creation failure)
+		var pathErr *resources.PathCreationError
+		if errors.As(result.Error, &pathErr) {
+			r.failedCreations = append(r.failedCreations, pathErr.Path)
+		}
+
+		// Track failed deletions
+		if result.Action == repository.FileActionDeleted {
+			r.failedDeletions = append(r.failedDeletions, result.Path)
 		}
-		r.errorCount++
 	}
 
 	r.updateSummary(result)
@@ -111,6 +132,8 @@ func (r *jobProgressRecorder) ResetResults() {
 	r.errorCount = 0
 	r.errors = nil
 	r.summaries = make(map[string]*provisioning.JobResourceSummary)
+	r.failedCreations = nil
+	r.failedDeletions = nil
 }
 
 func (r *jobProgressRecorder) SetMessage(ctx context.Context, msg string) {
@@ -292,3 +315,29 @@ func (r *jobProgressRecorder) Complete(ctx context.Context, err error) provision
 
 	return jobStatus
 }
+
+// IsNestedUnderFailedCreation checks if a path is nested under any failed folder creation
+func (r *jobProgressRecorder) IsNestedUnderFailedCreation(path string) bool {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	for _, failedCreation := range r.failedCreations {
+		if safepath.InDir(path, failedCreation) {
+			return true
+		}
+	}
+	return false
+}
+
+// HasFailedDeletionsUnder checks if any resource deletions failed under a folder path
+func (r *jobProgressRecorder) HasFailedDeletionsUnder(folderPath string) bool {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	for _, failedDeletion := range r.failedDeletions {
+		if safepath.InDir(failedDeletion, folderPath) {
+			return true
+		}
+	}
+	return false
+}

pkg/registry/apis/provisioning/jobs/progress_fn_mock.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v2.52.4. DO NOT EDIT.
+// Code generated by mockery v2.53.4. DO NOT EDIT.
 
 package jobs
 

pkg/registry/apis/provisioning/jobs/progress_test.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"testing"
 
+	"github.com/grafana/grafana/apps/provisioning/pkg/repository"
 	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+	"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -83,3 +85,266 @@ func TestJobProgressRecorderCompleteIncludesRefURLs(t *testing.T) {
 	assert.Equal(t, provisioning.JobStateSuccess, finalStatus.State)
 	assert.Equal(t, "completed successfully", finalStatus.Message)
 }
+
+func TestJobProgressRecorderAutomaticFailureTracking(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a progress recorder
+	mockProgressFn := func(ctx context.Context, status provisioning.JobStatus) error {
+		return nil
+	}
+	recorder := newJobProgressRecorder(mockProgressFn).(*jobProgressRecorder)
+
+	// Record a folder creation failure with PathCreationError
+	pathErr := &resources.PathCreationError{
+		Path: "folder1/",
+		Err:  assert.AnError,
+	}
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder1/file.json",
+		Action: repository.FileActionCreated,
+		Error:  pathErr,
+	})
+
+	// Record another PathCreationError for a different folder
+	pathErr2 := &resources.PathCreationError{
+		Path: "folder2/subfolder/",
+		Err:  assert.AnError,
+	}
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder2/subfolder/file.json",
+		Action: repository.FileActionCreated,
+		Error:  pathErr2,
+	})
+
+	// Record a deletion failure
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder3/file1.json",
+		Action: repository.FileActionDeleted,
+		Error:  assert.AnError,
+	})
+
+	// Record another deletion failure
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder4/subfolder/file2.json",
+		Action: repository.FileActionDeleted,
+		Error:  assert.AnError,
+	})
+
+	// Verify failed creations are tracked
+	recorder.mu.RLock()
+	assert.Len(t, recorder.failedCreations, 2)
+	assert.Contains(t, recorder.failedCreations, "folder1/")
+	assert.Contains(t, recorder.failedCreations, "folder2/subfolder/")
+
+	// Verify failed deletions are tracked
+	assert.Len(t, recorder.failedDeletions, 2)
+	assert.Contains(t, recorder.failedDeletions, "folder3/file1.json")
+	assert.Contains(t, recorder.failedDeletions, "folder4/subfolder/file2.json")
+	recorder.mu.RUnlock()
+}
+
+func TestJobProgressRecorderIsNestedUnderFailedCreation(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a progress recorder
+	mockProgressFn := func(ctx context.Context, status provisioning.JobStatus) error {
+		return nil
+	}
+	recorder := newJobProgressRecorder(mockProgressFn).(*jobProgressRecorder)
+
+	// Add failed creations via Record
+	pathErr1 := &resources.PathCreationError{
+		Path: "folder1/",
+		Err:  assert.AnError,
+	}
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder1/file.json",
+		Action: repository.FileActionCreated,
+		Error:  pathErr1,
+	})
+
+	pathErr2 := &resources.PathCreationError{
+		Path: "folder2/subfolder/",
+		Err:  assert.AnError,
+	}
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder2/subfolder/file.json",
+		Action: repository.FileActionCreated,
+		Error:  pathErr2,
+	})
+
+	// Test nested paths
+	assert.True(t, recorder.IsNestedUnderFailedCreation("folder1/file.json"))
+	assert.True(t, recorder.IsNestedUnderFailedCreation("folder1/nested/file.json"))
+	assert.True(t, recorder.IsNestedUnderFailedCreation("folder2/subfolder/file.json"))
+
+	// Test non-nested paths
+	assert.False(t, recorder.IsNestedUnderFailedCreation("other/file.json"))
+	assert.False(t, recorder.IsNestedUnderFailedCreation("folder3/file.json"))
+	assert.False(t, recorder.IsNestedUnderFailedCreation("file.json"))
+}
+
+func TestJobProgressRecorderHasFailedDeletionsUnder(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a progress recorder
+	mockProgressFn := func(ctx context.Context, status provisioning.JobStatus) error {
+		return nil
+	}
+	recorder := newJobProgressRecorder(mockProgressFn).(*jobProgressRecorder)
+
+	// Add failed deletions via Record
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder1/file1.json",
+		Action: repository.FileActionDeleted,
+		Error:  assert.AnError,
+	})
+
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder2/subfolder/file2.json",
+		Action: repository.FileActionDeleted,
+		Error:  assert.AnError,
+	})
+
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder3/nested/deep/file3.json",
+		Action: repository.FileActionDeleted,
+		Error:  assert.AnError,
+	})
+
+	// Test folder paths with failed deletions
+	assert.True(t, recorder.HasFailedDeletionsUnder("folder1/"))
+	assert.True(t, recorder.HasFailedDeletionsUnder("folder2/"))
+	assert.True(t, recorder.HasFailedDeletionsUnder("folder2/subfolder/"))
+	assert.True(t, recorder.HasFailedDeletionsUnder("folder3/"))
+	assert.True(t, recorder.HasFailedDeletionsUnder("folder3/nested/"))
+	assert.True(t, recorder.HasFailedDeletionsUnder("folder3/nested/deep/"))
+
+	// Test folder paths without failed deletions
+	assert.False(t, recorder.HasFailedDeletionsUnder("other/"))
+	assert.False(t, recorder.HasFailedDeletionsUnder("different/"))
+}
+
+func TestJobProgressRecorderResetResults(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a progress recorder
+	mockProgressFn := func(ctx context.Context, status provisioning.JobStatus) error {
+		return nil
+	}
+	recorder := newJobProgressRecorder(mockProgressFn).(*jobProgressRecorder)
+
+	// Add some data via Record
+	pathErr := &resources.PathCreationError{
+		Path: "folder1/",
+		Err:  assert.AnError,
+	}
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder1/file.json",
+		Action: repository.FileActionCreated,
+		Error:  pathErr,
+	})
+
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder2/file.json",
+		Action: repository.FileActionDeleted,
+		Error:  assert.AnError,
+	})
+
+	// Verify data is stored
+	recorder.mu.RLock()
+	assert.Len(t, recorder.failedCreations, 1)
+	assert.Len(t, recorder.failedDeletions, 1)
+	recorder.mu.RUnlock()
+
+	// Reset results
+	recorder.ResetResults()
+
+	// Verify data is cleared
+	recorder.mu.RLock()
+	assert.Nil(t, recorder.failedCreations)
+	assert.Nil(t, recorder.failedDeletions)
+	recorder.mu.RUnlock()
+}
+
+func TestJobProgressRecorderConcurrentAccess(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a progress recorder
+	mockProgressFn := func(ctx context.Context, status provisioning.JobStatus) error {
+		return nil
+	}
+	recorder := newJobProgressRecorder(mockProgressFn)
+
+	// Test concurrent writes and reads
+	done := make(chan bool)
+
+	// Writer goroutines
+	for i := 0; i < 10; i++ {
+		go func(idx int) {
+			pathErr := &resources.PathCreationError{
+				Path: "folder/",
+				Err:  assert.AnError,
+			}
+			recorder.Record(ctx, JobResourceResult{
+				Path:   "test/path",
+				Action: repository.FileActionCreated,
+				Error:  pathErr,
+			})
+			recorder.Record(ctx, JobResourceResult{
+				Path:   "test/file.json",
+				Action: repository.FileActionDeleted,
+				Error:  assert.AnError,
+			})
+			done <- true
+		}(i)
+	}
+
+	// Reader goroutines
+	for i := 0; i < 10; i++ {
+		go func() {
+			recorder.IsNestedUnderFailedCreation("test/path")
+			recorder.HasFailedDeletionsUnder("test/")
+			done <- true
+		}()
+	}
+
+	// Wait for all goroutines
+	for i := 0; i < 20; i++ {
+		<-done
+	}
+
+	// Just verify no panics occurred and basic functionality works
+	assert.NotNil(t, recorder)
+}
+
+func TestJobProgressRecorderIgnoredActionsDontCountAsErrors(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a progress recorder
+	mockProgressFn := func(ctx context.Context, status provisioning.JobStatus) error {
+		return nil
+	}
+	recorder := newJobProgressRecorder(mockProgressFn).(*jobProgressRecorder)
+
+	// Record an ignored action with error
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder1/file1.json",
+		Action: repository.FileActionIgnored,
+		Error:  assert.AnError,
+	})
+
+	// Record a real error for comparison
+	recorder.Record(ctx, JobResourceResult{
+		Path:   "folder2/file2.json",
+		Action: repository.FileActionCreated,
+		Error:  assert.AnError,
+	})
+
+	// Verify error count doesn't include ignored actions
+	recorder.mu.RLock()
+	assert.Equal(t, 1, recorder.errorCount, "ignored actions should not be counted as errors")
+	assert.Len(t, recorder.errors, 1, "ignored action errors should not be in error list")
+	recorder.mu.RUnlock()
+}

pkg/registry/apis/provisioning/jobs/queue.go

@@ -29,6 +29,10 @@ type JobProgressRecorder interface {
 	StrictMaxErrors(maxErrors int)
 	SetRefURLs(ctx context.Context, refURLs *provisioning.RepositoryURLs)
 	Complete(ctx context.Context, err error) provisioning.JobStatus
+	// IsNestedUnderFailedCreation checks if a path is nested under any failed folder creation
+	IsNestedUnderFailedCreation(path string) bool
+	// HasFailedDeletionsUnder checks if any resource deletions failed under a folder path
+	HasFailedDeletionsUnder(folderPath string) bool
 }
 
 // Worker is a worker that can process a job

pkg/registry/apis/provisioning/jobs/sync/full.go

@@ -80,6 +80,20 @@ func applyChange(ctx context.Context, change ResourceFileChange, clients resourc
 		return
 	}
 
+	// Check if this resource is nested under a failed folder creation
+	// This only applies to creation/update operations, not deletions
+	if change.Action != repository.FileActionDeleted && progress.IsNestedUnderFailedCreation(change.Path) {
+		// Skip this resource since its parent folder failed to be created
+		skipCtx, skipSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes.skip_nested_resource")
+		progress.Record(skipCtx, jobs.JobResourceResult{
+			Path:   change.Path,
+			Action: repository.FileActionIgnored,
+			Error:  fmt.Errorf("skipped: parent folder creation failed"),
+		})
+		skipSpan.End()
+		return
+	}
+
 	if change.Action == repository.FileActionDeleted {
 		deleteCtx, deleteSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes.delete")
 		result := jobs.JobResourceResult{
@@ -123,6 +137,23 @@ func applyChange(ctx context.Context, change ResourceFileChange, clients resourc
 
 	// Handle folders based on action type
 	if safepath.IsDir(change.Path) {
+		// Check if this is a folder deletion - need to ensure no children failed to delete
+		if change.Action == repository.FileActionDeleted {
+			// Check if any resources under this folder failed to delete
+			if progress.HasFailedDeletionsUnder(change.Path) {
+				skipCtx, skipSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes.skip_folder_with_failed_deletions")
+				progress.Record(skipCtx, jobs.JobResourceResult{
+					Path:   change.Path,
+					Action: repository.FileActionIgnored,
+					Group:  resources.FolderKind.Group,
+					Kind:   resources.FolderKind.Kind,
+					Error:  fmt.Errorf("skipped: child resource deletions failed"),
+				})
+				skipSpan.End()
+				return
+			}
+		}
+
 		// For non-deletions, ensure folder exists
 		ensureFolderCtx, ensureFolderSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes.ensure_folder_exists")
 		result := jobs.JobResourceResult{
@@ -138,6 +169,7 @@ func applyChange(ctx context.Context, change ResourceFileChange, clients resourc
 			ensureFolderSpan.RecordError(err)
 			ensureFolderSpan.End()
 			progress.Record(ctx, result)
+
 			return
 		}
 
@@ -159,6 +191,7 @@ func applyChange(ctx context.Context, change ResourceFileChange, clients resourc
 	if err != nil {
 		writeSpan.RecordError(err)
 		result.Error = fmt.Errorf("writing resource from file %s: %w", change.Path, err)
+
 	}
 
 	progress.Record(writeCtx, result)

pkg/registry/apis/provisioning/jobs/sync/full_hierarchical_test.go

@@ -0,0 +1,294 @@
+package sync
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	dynamicfake "k8s.io/client-go/dynamic/fake"
+	k8testing "k8s.io/client-go/testing"
+
+	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+	"github.com/grafana/grafana/apps/provisioning/pkg/repository"
+	"github.com/grafana/grafana/pkg/infra/tracing"
+	"github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs"
+	"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
+)
+
+/*
+TestFullSync_HierarchicalErrorHandling tests the hierarchical error handling behavior:
+
+FOLDER CREATION FAILURES:
+- When a folder fails to be created with PathCreationError, all nested resources are skipped
+- Nested resources are recorded with FileActionIgnored and error "skipped: parent folder creation failed"
+- Only the folder creation error counts toward error limits
+- Nested resource skips do NOT count toward error limits
+
+FOLDER DELETION FAILURES:
+- When a file deletion fails, it's tracked in failedDeletions
+- When cleaning up folders, we check HasFailedDeletionsUnder()
+- If children failed to delete, folder deletion is skipped with FileActionIgnored
+- This prevents orphaning resources that still exist
+
+DELETIONS NOT AFFECTED BY CREATION FAILURES:
+- If a folder creation fails, deletion operations for resources in that folder still proceed
+- This is because the resource might already exist from a previous sync
+- Only creations/updates/renames are affected by failed folder creation
+
+AUTOMATIC TRACKING:
+- Record() automatically detects PathCreationError and adds to failedCreations
+- Record() automatically detects deletion failures and adds to failedDeletions
+- No manual calls to AddFailedCreation/AddFailedDeletion needed
+*/
+func TestFullSync_HierarchicalErrorHandling(t *testing.T) {
+	tests := []struct {
+		name          string
+		setupMocks    func(*repository.MockRepository, *resources.MockRepositoryResources, *resources.MockResourceClients, *jobs.MockJobProgressRecorder, *dynamicfake.FakeDynamicClient)
+		changes       []ResourceFileChange
+		description   string
+		expectError   bool
+		errorContains string
+	}{
+		{
+			name:        "folder creation fails, nested file skipped",
+			description: "When folder1/ fails to create, folder1/file.json should be skipped with FileActionIgnored",
+			changes: []ResourceFileChange{
+				{Path: "folder1/file.json", Action: repository.FileActionCreated},
+			},
+			setupMocks: func(repo *repository.MockRepository, repoResources *resources.MockRepositoryResources, clients *resources.MockResourceClients, progress *jobs.MockJobProgressRecorder, _ *dynamicfake.FakeDynamicClient) {
+				// First, check if nested under failed creation - not yet
+				progress.On("IsNestedUnderFailedCreation", "folder1/file.json").Return(false).Once()
+
+				// WriteResourceFromFile fails with PathCreationError for folder1/
+				folderErr := &resources.PathCreationError{Path: "folder1/", Err: fmt.Errorf("permission denied")}
+				repoResources.On("WriteResourceFromFile", mock.Anything, "folder1/file.json", "").
+					Return("", schema.GroupVersionKind{}, folderErr).Once()
+
+				// File will be recorded with error, triggering automatic tracking of folder1/ failure
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/file.json" && r.Error != nil && r.Action == repository.FileActionCreated
+				})).Return().Once()
+			},
+		},
+		{
+			name:        "folder creation fails, multiple nested resources skipped",
+			description: "When folder1/ fails to create, all nested resources (subfolder, files) are skipped",
+			changes: []ResourceFileChange{
+				{Path: "folder1/file1.json", Action: repository.FileActionCreated},
+				{Path: "folder1/subfolder/file2.json", Action: repository.FileActionCreated},
+				{Path: "folder1/file3.json", Action: repository.FileActionCreated},
+			},
+			setupMocks: func(repo *repository.MockRepository, repoResources *resources.MockRepositoryResources, clients *resources.MockResourceClients, progress *jobs.MockJobProgressRecorder, _ *dynamicfake.FakeDynamicClient) {
+				// First file triggers folder creation failure
+				progress.On("IsNestedUnderFailedCreation", "folder1/file1.json").Return(false).Once()
+				folderErr := &resources.PathCreationError{Path: "folder1/", Err: fmt.Errorf("permission denied")}
+				repoResources.On("WriteResourceFromFile", mock.Anything, "folder1/file1.json", "").
+					Return("", schema.GroupVersionKind{}, folderErr).Once()
+
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/file1.json" && r.Error != nil
+				})).Return().Once()
+
+				// Subsequent files in same folder are skipped
+				progress.On("IsNestedUnderFailedCreation", "folder1/subfolder/file2.json").Return(true).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/subfolder/file2.json" &&
+						r.Action == repository.FileActionIgnored &&
+						r.Error != nil &&
+						r.Error.Error() == "skipped: parent folder creation failed"
+				})).Return().Once()
+
+				progress.On("IsNestedUnderFailedCreation", "folder1/file3.json").Return(true).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/file3.json" &&
+						r.Action == repository.FileActionIgnored &&
+						r.Error != nil &&
+						r.Error.Error() == "skipped: parent folder creation failed"
+				})).Return().Once()
+			},
+		},
+		{
+			name:        "file deletion failure tracked",
+			description: "When a file deletion fails, it's automatically tracked in failedDeletions",
+			changes: []ResourceFileChange{
+				{
+					Path:   "folder1/file.json",
+					Action: repository.FileActionDeleted,
+					Existing: &provisioning.ResourceListItem{
+						Name:     "file1",
+						Group:    "dashboard.grafana.app",
+						Resource: "dashboards",
+					},
+				},
+			},
+			setupMocks: func(repo *repository.MockRepository, repoResources *resources.MockRepositoryResources, clients *resources.MockResourceClients, progress *jobs.MockJobProgressRecorder, dynamicClient *dynamicfake.FakeDynamicClient) {
+				gvk := schema.GroupVersionKind{Group: "dashboard.grafana.app", Kind: "Dashboard", Version: "v1"}
+				gvr := schema.GroupVersionResource{Group: "dashboard.grafana.app", Resource: "dashboards", Version: "v1"}
+
+				clients.On("ForResource", mock.Anything, mock.MatchedBy(func(gvr schema.GroupVersionResource) bool {
+					return gvr.Group == "dashboard.grafana.app"
+				})).Return(dynamicClient.Resource(gvr), gvk, nil)
+
+				// File deletion fails
+				dynamicClient.PrependReactor("delete", "dashboards", func(action k8testing.Action) (bool, runtime.Object, error) {
+					return true, nil, fmt.Errorf("permission denied")
+				})
+
+				// File deletion recorded with error, automatically tracked in failedDeletions
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/file.json" &&
+						r.Action == repository.FileActionDeleted &&
+						r.Error != nil
+				})).Return().Once()
+			},
+		},
+		{
+			name:        "deletion proceeds despite creation failure",
+			description: "When folder1/ fails to create, deletion of folder1/file2.json still proceeds (resource might exist from previous sync)",
+			changes: []ResourceFileChange{
+				{Path: "folder1/file1.json", Action: repository.FileActionCreated},
+				{
+					Path:   "folder1/file2.json",
+					Action: repository.FileActionDeleted,
+					Existing: &provisioning.ResourceListItem{
+						Name:     "file2",
+						Group:    "dashboard.grafana.app",
+						Resource: "dashboards",
+					},
+				},
+			},
+			setupMocks: func(repo *repository.MockRepository, repoResources *resources.MockRepositoryResources, clients *resources.MockResourceClients, progress *jobs.MockJobProgressRecorder, dynamicClient *dynamicfake.FakeDynamicClient) {
+				// Creation fails
+				progress.On("IsNestedUnderFailedCreation", "folder1/file1.json").Return(false).Once()
+				folderErr := &resources.PathCreationError{Path: "folder1/", Err: fmt.Errorf("permission denied")}
+				repoResources.On("WriteResourceFromFile", mock.Anything, "folder1/file1.json", "").
+					Return("", schema.GroupVersionKind{}, folderErr).Once()
+
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/file1.json" && r.Error != nil
+				})).Return().Once()
+
+				// Deletion proceeds (NOT checking IsNestedUnderFailedCreation for deletions)
+				// Note: deletion will fail because resource doesn't exist, but that's fine for this test
+				gvk := schema.GroupVersionKind{Group: "dashboard.grafana.app", Kind: "Dashboard", Version: "v1"}
+				gvr := schema.GroupVersionResource{Group: "dashboard.grafana.app", Resource: "dashboards", Version: "v1"}
+
+				clients.On("ForResource", mock.Anything, mock.MatchedBy(func(gvr schema.GroupVersionResource) bool {
+					return gvr.Group == "dashboard.grafana.app"
+				})).Return(dynamicClient.Resource(gvr), gvk, nil)
+
+				// Record deletion attempt (will have error since resource doesn't exist, but that's ok)
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/file2.json" &&
+						r.Action == repository.FileActionDeleted
+					// Not checking r.Error because resource doesn't exist in fake client
+				})).Return().Once()
+			},
+		},
+		{
+			name:        "multi-level nesting - all skipped",
+			description: "When level1/ fails, level1/level2/level3/file.json is also skipped",
+			changes: []ResourceFileChange{
+				{Path: "level1/file1.json", Action: repository.FileActionCreated},
+				{Path: "level1/level2/file2.json", Action: repository.FileActionCreated},
+				{Path: "level1/level2/level3/file3.json", Action: repository.FileActionCreated},
+			},
+			setupMocks: func(repo *repository.MockRepository, repoResources *resources.MockRepositoryResources, clients *resources.MockResourceClients, progress *jobs.MockJobProgressRecorder, _ *dynamicfake.FakeDynamicClient) {
+				// First file triggers level1/ failure
+				progress.On("IsNestedUnderFailedCreation", "level1/file1.json").Return(false).Once()
+				folderErr := &resources.PathCreationError{Path: "level1/", Err: fmt.Errorf("permission denied")}
+				repoResources.On("WriteResourceFromFile", mock.Anything, "level1/file1.json", "").
+					Return("", schema.GroupVersionKind{}, folderErr).Once()
+
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "level1/file1.json" && r.Error != nil
+				})).Return().Once()
+
+				// All nested files are skipped
+				for _, path := range []string{"level1/level2/file2.json", "level1/level2/level3/file3.json"} {
+					progress.On("IsNestedUnderFailedCreation", path).Return(true).Once()
+					progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+						return r.Path == path && r.Action == repository.FileActionIgnored
+					})).Return().Once()
+				}
+			},
+		},
+		{
+			name:        "mixed success and failure",
+			description: "When success/ works and failure/ fails, only failure/* are skipped",
+			changes: []ResourceFileChange{
+				{Path: "success/file1.json", Action: repository.FileActionCreated},
+				{Path: "failure/file2.json", Action: repository.FileActionCreated},
+				{Path: "failure/nested/file3.json", Action: repository.FileActionCreated},
+			},
+			setupMocks: func(repo *repository.MockRepository, repoResources *resources.MockRepositoryResources, clients *resources.MockResourceClients, progress *jobs.MockJobProgressRecorder, _ *dynamicfake.FakeDynamicClient) {
+				// Success path works
+				progress.On("IsNestedUnderFailedCreation", "success/file1.json").Return(false).Once()
+				repoResources.On("WriteResourceFromFile", mock.Anything, "success/file1.json", "").
+					Return("resource1", schema.GroupVersionKind{Kind: "Dashboard"}, nil).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "success/file1.json" && r.Error == nil
+				})).Return().Once()
+
+				// Failure path fails
+				progress.On("IsNestedUnderFailedCreation", "failure/file2.json").Return(false).Once()
+				folderErr := &resources.PathCreationError{Path: "failure/", Err: fmt.Errorf("disk full")}
+				repoResources.On("WriteResourceFromFile", mock.Anything, "failure/file2.json", "").
+					Return("", schema.GroupVersionKind{}, folderErr).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "failure/file2.json" && r.Error != nil
+				})).Return().Once()
+
+				// Nested file in failure path is skipped
+				progress.On("IsNestedUnderFailedCreation", "failure/nested/file3.json").Return(true).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "failure/nested/file3.json" && r.Action == repository.FileActionIgnored
+				})).Return().Once()
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			scheme := runtime.NewScheme()
+			dynamicClient := dynamicfake.NewSimpleDynamicClient(scheme)
+
+			repo := repository.NewMockRepository(t)
+			repoResources := resources.NewMockRepositoryResources(t)
+			clients := resources.NewMockResourceClients(t)
+			progress := jobs.NewMockJobProgressRecorder(t)
+			compareFn := NewMockCompareFn(t)
+
+			repo.On("Config").Return(&provisioning.Repository{
+				ObjectMeta: metav1.ObjectMeta{Name: "test-repo"},
+				Spec:       provisioning.RepositorySpec{Title: "Test Repo"},
+			})
+
+			tt.setupMocks(repo, repoResources, clients, progress, dynamicClient)
+
+			compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(tt.changes, nil)
+			progress.On("SetTotal", mock.Anything, len(tt.changes)).Return()
+			progress.On("TooManyErrors").Return(nil).Maybe()
+
+			err := FullSync(context.Background(), repo, compareFn.Execute, clients, "ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+
+			if tt.expectError {
+				require.Error(t, err)
+				if tt.errorContains != "" {
+					require.Contains(t, err.Error(), tt.errorContains)
+				}
+			} else {
+				require.NoError(t, err)
+			}
+
+			progress.AssertExpectations(t)
+			repoResources.AssertExpectations(t)
+		})
+	}
+}

pkg/registry/apis/provisioning/jobs/sync/full_test.go

@@ -754,3 +754,376 @@ func TestFullSync_ApplyChanges(t *testing.T) { //nolint:gocyclo
 		})
 	}
 }
+
+// TestFullSync_HierarchicalErrorHandling_FailedFolderCreation tests that when a folder
+// creation fails, all nested resources are skipped with FileActionIgnored
+func TestFullSync_HierarchicalErrorHandling_FailedFolderCreation(t *testing.T) {
+	repo := repository.NewMockRepository(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	clients := resources.NewMockResourceClients(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+	compareFn := NewMockCompareFn(t)
+
+	repo.On("Config").Return(&provisioning.Repository{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-repo"},
+		Spec:       provisioning.RepositorySpec{Title: "Test Repo"},
+	})
+
+	changes := []ResourceFileChange{
+		{Path: "folder1/", Action: repository.FileActionCreated},
+		{Path: "folder1/subfolder/", Action: repository.FileActionCreated},
+		{Path: "folder1/file1.json", Action: repository.FileActionCreated},
+		{Path: "folder1/subfolder/file2.json", Action: repository.FileActionCreated},
+	}
+
+	folderErr := &resources.PathCreationError{Path: "folder1/", Err: fmt.Errorf("permission denied")}
+	repoResources.On("EnsureFolderPathExist", mock.Anything, "folder1/").Return("", folderErr).Once()
+
+	progress.On("IsNestedUnderFailedCreation", "folder1/subfolder/").Return(true).Once()
+	progress.On("IsNestedUnderFailedCreation", "folder1/file1.json").Return(true).Once()
+	progress.On("IsNestedUnderFailedCreation", "folder1/subfolder/file2.json").Return(true).Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "folder1/" && r.Error != nil
+	})).Return().Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "folder1/subfolder/" && r.Action == repository.FileActionIgnored
+	})).Return().Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "folder1/file1.json" && r.Action == repository.FileActionIgnored
+	})).Return().Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "folder1/subfolder/file2.json" && r.Action == repository.FileActionIgnored
+	})).Return().Once()
+
+	compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, len(changes)).Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	err := FullSync(context.Background(), repo, compareFn.Execute, clients, "ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+}
+
+// TestFullSync_HierarchicalErrorHandling_FailedFileDeletion tests folder deletion is prevented when child deletion fails
+func TestFullSync_HierarchicalErrorHandling_FailedFileDeletion(t *testing.T) {
+	scheme := runtime.NewScheme()
+	dynamicClient := dynamicfake.NewSimpleDynamicClient(scheme)
+
+	repo := repository.NewMockRepository(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	clients := resources.NewMockResourceClients(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+	compareFn := NewMockCompareFn(t)
+
+	repo.On("Config").Return(&provisioning.Repository{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-repo"},
+		Spec:       provisioning.RepositorySpec{Title: "Test Repo"},
+	})
+
+	changes := []ResourceFileChange{
+		{
+			Path:     "folder1/file1.json",
+			Action:   repository.FileActionDeleted,
+			Existing: &provisioning.ResourceListItem{Name: "file1", Group: "dashboard.grafana.app", Resource: "dashboards"},
+		},
+		{Path: "folder1/", Action: repository.FileActionDeleted},
+	}
+
+	gvk := schema.GroupVersionKind{Group: "dashboard.grafana.app", Kind: "Dashboard", Version: "v1"}
+	gvr := schema.GroupVersionResource{Group: "dashboard.grafana.app", Resource: "dashboards", Version: "v1"}
+
+	clients.On("ForResource", mock.Anything, mock.MatchedBy(func(gvr schema.GroupVersionResource) bool {
+		return gvr.Group == "dashboard.grafana.app"
+	})).Return(dynamicClient.Resource(gvr), gvk, nil)
+
+	dynamicClient.PrependReactor("delete", "dashboards", func(action k8testing.Action) (bool, runtime.Object, error) {
+		return true, nil, fmt.Errorf("permission denied")
+	})
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "folder1/file1.json" && r.Error != nil
+	})).Return().Once()
+
+	progress.On("HasFailedDeletionsUnder", "folder1/").Return(true).Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "folder1/" && r.Action == repository.FileActionIgnored
+	})).Return().Once()
+
+	compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, len(changes)).Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	err := FullSync(context.Background(), repo, compareFn.Execute, clients, "ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+}
+
+// TestFullSync_HierarchicalErrorHandling_DeletionNotAffectedByCreationFailure tests deletions proceed despite creation failures
+func TestFullSync_HierarchicalErrorHandling_DeletionNotAffectedByCreationFailure(t *testing.T) {
+	scheme := runtime.NewScheme()
+	dynamicClient := dynamicfake.NewSimpleDynamicClient(scheme)
+
+	repo := repository.NewMockRepository(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	clients := resources.NewMockResourceClients(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+	compareFn := NewMockCompareFn(t)
+
+	repo.On("Config").Return(&provisioning.Repository{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-repo"},
+		Spec:       provisioning.RepositorySpec{Title: "Test Repo"},
+	})
+
+	changes := []ResourceFileChange{
+		{Path: "folder1/", Action: repository.FileActionCreated},
+		{
+			Path:     "folder1/file1.json",
+			Action:   repository.FileActionDeleted,
+			Existing: &provisioning.ResourceListItem{Name: "file1", Group: "dashboard.grafana.app", Resource: "dashboards"},
+		},
+	}
+
+	folderErr := &resources.PathCreationError{Path: "folder1/", Err: fmt.Errorf("permission denied")}
+	repoResources.On("EnsureFolderPathExist", mock.Anything, "folder1/").Return("", folderErr).Once()
+
+	gvk := schema.GroupVersionKind{Group: "dashboard.grafana.app", Kind: "Dashboard", Version: "v1"}
+	gvr := schema.GroupVersionResource{Group: "dashboard.grafana.app", Resource: "dashboards", Version: "v1"}
+	clients.On("ForResource", mock.Anything, mock.MatchedBy(func(gvr schema.GroupVersionResource) bool {
+		return gvr.Group == "dashboard.grafana.app"
+	})).Return(dynamicClient.Resource(gvr), gvk, nil)
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "folder1/" && r.Error != nil
+	})).Return().Once()
+
+	// Deletion should proceed (not check IsNestedUnderFailedCreation)
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "folder1/file1.json" && r.Action == repository.FileActionDeleted && r.Error == nil
+	})).Return().Once()
+
+	compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, len(changes)).Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	err := FullSync(context.Background(), repo, compareFn.Execute, clients, "ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+}
+
+// TestFullSync_HierarchicalErrorHandling_MultiLevelNesting tests errors cascade through multiple nesting levels
+func TestFullSync_HierarchicalErrorHandling_MultiLevelNesting(t *testing.T) {
+	repo := repository.NewMockRepository(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	clients := resources.NewMockResourceClients(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+	compareFn := NewMockCompareFn(t)
+
+	repo.On("Config").Return(&provisioning.Repository{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-repo"},
+		Spec:       provisioning.RepositorySpec{Title: "Test Repo"},
+	})
+
+	changes := []ResourceFileChange{
+		{Path: "level1/", Action: repository.FileActionCreated},
+		{Path: "level1/level2/", Action: repository.FileActionCreated},
+		{Path: "level1/level2/level3/", Action: repository.FileActionCreated},
+		{Path: "level1/level2/level3/file.json", Action: repository.FileActionCreated},
+	}
+
+	folderErr := &resources.PathCreationError{Path: "level1/", Err: fmt.Errorf("permission denied")}
+	repoResources.On("EnsureFolderPathExist", mock.Anything, "level1/").Return("", folderErr).Once()
+
+	progress.On("IsNestedUnderFailedCreation", "level1/level2/").Return(true).Once()
+	progress.On("IsNestedUnderFailedCreation", "level1/level2/level3/").Return(true).Once()
+	progress.On("IsNestedUnderFailedCreation", "level1/level2/level3/file.json").Return(true).Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "level1/" && r.Error != nil
+	})).Return().Once()
+
+	for _, path := range []string{"level1/level2/", "level1/level2/level3/", "level1/level2/level3/file.json"} {
+		progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+			return r.Path == path && r.Action == repository.FileActionIgnored
+		})).Return().Once()
+	}
+
+	compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, len(changes)).Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	err := FullSync(context.Background(), repo, compareFn.Execute, clients, "ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+}
+
+// TestFullSync_HierarchicalErrorHandling_MultipleFolderDeletionFailures tests multiple folder deletion failures
+func TestFullSync_HierarchicalErrorHandling_MultipleFolderDeletionFailures(t *testing.T) {
+	scheme := runtime.NewScheme()
+	dynamicClient := dynamicfake.NewSimpleDynamicClient(scheme)
+
+	repo := repository.NewMockRepository(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	clients := resources.NewMockResourceClients(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+	compareFn := NewMockCompareFn(t)
+
+	repo.On("Config").Return(&provisioning.Repository{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-repo"},
+		Spec:       provisioning.RepositorySpec{Title: "Test Repo"},
+	})
+
+	changes := []ResourceFileChange{
+		{Path: "folder1/file1.json", Action: repository.FileActionDeleted, Existing: &provisioning.ResourceListItem{Name: "file1", Group: "dashboard.grafana.app", Resource: "dashboards"}},
+		{Path: "folder1/", Action: repository.FileActionDeleted},
+		{Path: "folder2/file2.json", Action: repository.FileActionDeleted, Existing: &provisioning.ResourceListItem{Name: "file2", Group: "dashboard.grafana.app", Resource: "dashboards"}},
+		{Path: "folder2/", Action: repository.FileActionDeleted},
+	}
+
+	gvk := schema.GroupVersionKind{Group: "dashboard.grafana.app", Kind: "Dashboard", Version: "v1"}
+	gvr := schema.GroupVersionResource{Group: "dashboard.grafana.app", Resource: "dashboards", Version: "v1"}
+	clients.On("ForResource", mock.Anything, mock.MatchedBy(func(gvr schema.GroupVersionResource) bool {
+		return gvr.Group == "dashboard.grafana.app"
+	})).Return(dynamicClient.Resource(gvr), gvk, nil)
+
+	dynamicClient.PrependReactor("delete", "dashboards", func(action k8testing.Action) (bool, runtime.Object, error) {
+		return true, nil, fmt.Errorf("permission denied")
+	})
+
+	for _, path := range []string{"folder1/file1.json", "folder2/file2.json"} {
+		progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+			return r.Path == path && r.Error != nil
+		})).Return().Once()
+	}
+
+	progress.On("HasFailedDeletionsUnder", "folder1/").Return(true).Once()
+	progress.On("HasFailedDeletionsUnder", "folder2/").Return(true).Once()
+
+	for _, path := range []string{"folder1/", "folder2/"} {
+		progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+			return r.Path == path && r.Action == repository.FileActionIgnored
+		})).Return().Once()
+	}
+
+	compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, len(changes)).Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	err := FullSync(context.Background(), repo, compareFn.Execute, clients, "ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+}
+
+// TestFullSync_HierarchicalErrorHandling_MixedSuccessAndFailure tests mixed scenarios
+func TestFullSync_HierarchicalErrorHandling_MixedSuccessAndFailure(t *testing.T) {
+	repo := repository.NewMockRepository(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	clients := resources.NewMockResourceClients(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+	compareFn := NewMockCompareFn(t)
+
+	repo.On("Config").Return(&provisioning.Repository{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-repo"},
+		Spec:       provisioning.RepositorySpec{Title: "Test Repo"},
+	})
+
+	changes := []ResourceFileChange{
+		{Path: "success/", Action: repository.FileActionCreated},
+		{Path: "success/file1.json", Action: repository.FileActionCreated},
+		{Path: "failure/", Action: repository.FileActionCreated},
+		{Path: "failure/file2.json", Action: repository.FileActionCreated},
+	}
+
+	repoResources.On("EnsureFolderPathExist", mock.Anything, "success/").Return("success-folder", nil).Once()
+	repoResources.On("WriteResourceFromFile", mock.Anything, "success/file1.json", "").
+		Return("resource1", schema.GroupVersionKind{Kind: "Dashboard"}, nil).Once()
+
+	folderErr := &resources.PathCreationError{Path: "failure/", Err: fmt.Errorf("disk full")}
+	repoResources.On("EnsureFolderPathExist", mock.Anything, "failure/").Return("", folderErr).Once()
+
+	progress.On("IsNestedUnderFailedCreation", "success/file1.json").Return(false).Once()
+	progress.On("IsNestedUnderFailedCreation", "failure/file2.json").Return(true).Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "success/" && r.Error == nil
+	})).Return().Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "success/file1.json" && r.Error == nil
+	})).Return().Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "failure/" && r.Error != nil
+	})).Return().Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "failure/file2.json" && r.Action == repository.FileActionIgnored
+	})).Return().Once()
+
+	compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, len(changes)).Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	err := FullSync(context.Background(), repo, compareFn.Execute, clients, "ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+	repoResources.AssertExpectations(t)
+}
+
+// TestFullSync_HierarchicalErrorHandling_NestedSubfolderDeletionFailure tests nested folder deletion failure
+func TestFullSync_HierarchicalErrorHandling_NestedSubfolderDeletionFailure(t *testing.T) {
+	scheme := runtime.NewScheme()
+	dynamicClient := dynamicfake.NewSimpleDynamicClient(scheme)
+
+	repo := repository.NewMockRepository(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	clients := resources.NewMockResourceClients(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+	compareFn := NewMockCompareFn(t)
+
+	repo.On("Config").Return(&provisioning.Repository{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-repo"},
+		Spec:       provisioning.RepositorySpec{Title: "Test Repo"},
+	})
+
+	changes := []ResourceFileChange{
+		{Path: "parent/subfolder/file.json", Action: repository.FileActionDeleted, Existing: &provisioning.ResourceListItem{Name: "file1", Group: "dashboard.grafana.app", Resource: "dashboards"}},
+		{Path: "parent/subfolder/", Action: repository.FileActionDeleted},
+		{Path: "parent/", Action: repository.FileActionDeleted},
+	}
+
+	gvk := schema.GroupVersionKind{Group: "dashboard.grafana.app", Kind: "Dashboard", Version: "v1"}
+	gvr := schema.GroupVersionResource{Group: "dashboard.grafana.app", Resource: "dashboards", Version: "v1"}
+	clients.On("ForResource", mock.Anything, mock.MatchedBy(func(gvr schema.GroupVersionResource) bool {
+		return gvr.Group == "dashboard.grafana.app"
+	})).Return(dynamicClient.Resource(gvr), gvk, nil)
+
+	dynamicClient.PrependReactor("delete", "dashboards", func(action k8testing.Action) (bool, runtime.Object, error) {
+		return true, nil, fmt.Errorf("permission denied")
+	})
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "parent/subfolder/file.json" && r.Error != nil
+	})).Return().Once()
+
+	progress.On("HasFailedDeletionsUnder", "parent/subfolder/").Return(true).Once()
+	progress.On("HasFailedDeletionsUnder", "parent/").Return(true).Once()
+
+	for _, path := range []string{"parent/subfolder/", "parent/"} {
+		progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+			return r.Path == path && r.Action == repository.FileActionIgnored
+		})).Return().Once()
+	}
+
+	compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, len(changes)).Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	err := FullSync(context.Background(), repo, compareFn.Execute, clients, "ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+}

pkg/registry/apis/provisioning/jobs/sync/incremental.go

@@ -60,7 +60,7 @@ func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef
 	if len(affectedFolders) > 0 {
 		cleanupStart := time.Now()
 		span.AddEvent("checking if impacted folders should be deleted", trace.WithAttributes(attribute.Int("affected_folders", len(affectedFolders))))
-		err := cleanupOrphanedFolders(ctx, repo, affectedFolders, repositoryResources, tracer)
+		err := cleanupOrphanedFolders(ctx, repo, affectedFolders, repositoryResources, tracer, progress)
 		metrics.RecordIncrementalSyncPhase(jobs.IncrementalSyncPhaseCleanup, time.Since(cleanupStart))
 		if err != nil {
 			return tracing.Error(span, fmt.Errorf("cleanup orphaned folders: %w", err))
@@ -85,6 +85,20 @@ func applyIncrementalChanges(ctx context.Context, diff []repository.VersionedFil
 			return nil, tracing.Error(span, err)
 		}
 
+		// Check if this resource is nested under a failed folder creation
+		// This only applies to creation/update/rename operations, not deletions
+		if change.Action != repository.FileActionDeleted && progress.IsNestedUnderFailedCreation(change.Path) {
+			// Skip this resource since its parent folder failed to be created
+			skipCtx, skipSpan := tracer.Start(ctx, "provisioning.sync.incremental.skip_nested_resource")
+			progress.Record(skipCtx, jobs.JobResourceResult{
+				Path:   change.Path,
+				Action: repository.FileActionIgnored,
+				Error:  fmt.Errorf("skipped: parent folder creation failed"),
+			})
+			skipSpan.End()
+			continue
+		}
+
 		if err := resources.IsPathSupported(change.Path); err != nil {
 			ensureFolderCtx, ensureFolderSpan := tracer.Start(ctx, "provisioning.sync.incremental.ensure_folder_path_exist")
 			// Maintain the safe segment for empty folders
@@ -98,6 +112,7 @@ func applyIncrementalChanges(ctx context.Context, diff []repository.VersionedFil
 				if err != nil {
 					ensureFolderSpan.RecordError(err)
 					ensureFolderSpan.End()
+
 					return nil, tracing.Error(span, fmt.Errorf("unable to create empty file folder: %w", err))
 				}
 
@@ -132,6 +147,7 @@ func applyIncrementalChanges(ctx context.Context, diff []repository.VersionedFil
 			if err != nil {
 				writeSpan.RecordError(err)
 				result.Error = fmt.Errorf("writing resource from file %s: %w", change.Path, err)
+
 			}
 			result.Name = name
 			result.Kind = gvk.Kind
@@ -185,6 +201,7 @@ func cleanupOrphanedFolders(
 	affectedFolders map[string]string,
 	repositoryResources resources.RepositoryResources,
 	tracer tracing.Tracer,
+	progress jobs.JobProgressRecorder,
 ) error {
 	ctx, span := tracer.Start(ctx, "provisioning.sync.incremental.cleanup_orphaned_folders")
 	defer span.End()
@@ -198,6 +215,12 @@ func cleanupOrphanedFolders(
 	for path, folderName := range affectedFolders {
 		span.SetAttributes(attribute.String("folder", folderName))
 
+		// Check if any resources under this folder failed to delete
+		if progress.HasFailedDeletionsUnder(path) {
+			span.AddEvent("skipping folder deletion: child resource deletions failed")
+			continue
+		}
+
 		// if we can no longer find the folder in git, then we can delete it from grafana
 		_, err := readerRepo.Read(ctx, path, "")
 		if err != nil && (errors.Is(err, repository.ErrFileNotFound) || apierrors.IsNotFound(err)) {

pkg/registry/apis/provisioning/jobs/sync/incremental_hierarchical_test.go

@@ -0,0 +1,342 @@
+package sync
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"github.com/grafana/grafana/apps/provisioning/pkg/repository"
+	"github.com/grafana/grafana/pkg/infra/tracing"
+	"github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs"
+	"github.com/grafana/grafana/pkg/registry/apis/provisioning/resources"
+)
+
+/*
+TestIncrementalSync_HierarchicalErrorHandling tests the hierarchical error handling behavior:
+
+FOLDER CREATION FAILURES:
+- When EnsureFolderPathExist fails with PathCreationError, the path is tracked
+- Subsequent resources under that path are skipped with FileActionIgnored
+- Only the initial folder creation error counts toward error limits
+- WriteResourceFromFile can also return PathCreationError for implicit folder creation
+
+FOLDER DELETION FAILURES (cleanupOrphanedFolders):
+- When RemoveResourceFromFile fails, path is tracked in failedDeletions
+- In cleanupOrphanedFolders, HasFailedDeletionsUnder() is checked before RemoveFolder
+- If children failed to delete, folder cleanup is skipped with a span event
+
+DELETIONS NOT AFFECTED BY CREATION FAILURES:
+- IsNestedUnderFailedCreation is NOT checked for FileActionDeleted
+- Deletions proceed even if their parent folder failed to be created
+- This handles cleanup of resources that exist from previous syncs
+
+RENAME OPERATIONS:
+- RenameResourceFile can return PathCreationError for the destination folder
+- Renames are affected by failed destination folder creation
+- Renames are NOT skipped due to source folder creation failures
+
+AUTOMATIC TRACKING:
+- Record() automatically detects PathCreationError via errors.As() and adds to failedCreations
+- Record() automatically detects FileActionDeleted with error and adds to failedDeletions
+- No manual tracking calls needed
+*/
+func TestIncrementalSync_HierarchicalErrorHandling(t *testing.T) {
+	tests := []struct {
+		name          string
+		setupMocks    func(*repository.MockVersioned, *resources.MockRepositoryResources, *jobs.MockJobProgressRecorder)
+		changes       []repository.VersionedFileChange
+		previousRef   string
+		currentRef    string
+		description   string
+		expectError   bool
+		errorContains string
+	}{
+		{
+			name:        "folder creation fails, nested file skipped",
+			description: "When unsupported/ fails to create via EnsureFolderPathExist, nested file is skipped",
+			previousRef: "old-ref",
+			currentRef:  "new-ref",
+			changes: []repository.VersionedFileChange{
+				{Action: repository.FileActionCreated, Path: "unsupported/file.txt", Ref: "new-ref"},
+				{Action: repository.FileActionCreated, Path: "unsupported/nested/file2.txt", Ref: "new-ref"},
+			},
+			setupMocks: func(repo *repository.MockVersioned, repoResources *resources.MockRepositoryResources, progress *jobs.MockJobProgressRecorder) {
+				// First file triggers folder creation which fails
+				folderErr := &resources.PathCreationError{Path: "unsupported/", Err: fmt.Errorf("permission denied")}
+				repoResources.On("EnsureFolderPathExist", mock.Anything, "unsupported/").Return("", folderErr).Once()
+
+				// First file recorded with error (note: error is from folder creation, but recorded against file)
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "unsupported/file.txt" &&
+						r.Action == repository.FileActionIgnored &&
+						r.Error != nil
+				})).Return().Once()
+
+				// Second file is skipped because parent folder failed
+				progress.On("IsNestedUnderFailedCreation", "unsupported/nested/file2.txt").Return(true).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "unsupported/nested/file2.txt" &&
+						r.Action == repository.FileActionIgnored &&
+						r.Error != nil &&
+						r.Error.Error() == "skipped: parent folder creation failed"
+				})).Return().Once()
+			},
+		},
+		{
+			name:        "WriteResourceFromFile returns PathCreationError, nested resources skipped",
+			description: "When WriteResourceFromFile implicitly creates a folder and fails, nested resources are skipped",
+			previousRef: "old-ref",
+			currentRef:  "new-ref",
+			changes: []repository.VersionedFileChange{
+				{Action: repository.FileActionCreated, Path: "folder1/file1.json", Ref: "new-ref"},
+				{Action: repository.FileActionCreated, Path: "folder1/file2.json", Ref: "new-ref"},
+				{Action: repository.FileActionCreated, Path: "folder1/nested/file3.json", Ref: "new-ref"},
+			},
+			setupMocks: func(repo *repository.MockVersioned, repoResources *resources.MockRepositoryResources, progress *jobs.MockJobProgressRecorder) {
+				// First file write fails with PathCreationError
+				folderErr := &resources.PathCreationError{Path: "folder1/", Err: fmt.Errorf("permission denied")}
+				repoResources.On("WriteResourceFromFile", mock.Anything, "folder1/file1.json", "new-ref").
+					Return("", schema.GroupVersionKind{}, folderErr).Once()
+
+				// First file recorded with error, automatically tracked
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/file1.json" &&
+						r.Action == repository.FileActionCreated &&
+						r.Error != nil
+				})).Return().Once()
+
+				// Subsequent files are skipped
+				progress.On("IsNestedUnderFailedCreation", "folder1/file2.json").Return(true).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/file2.json" && r.Action == repository.FileActionIgnored
+				})).Return().Once()
+
+				progress.On("IsNestedUnderFailedCreation", "folder1/nested/file3.json").Return(true).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/nested/file3.json" && r.Action == repository.FileActionIgnored
+				})).Return().Once()
+			},
+		},
+		{
+			name:        "file deletion fails, folder cleanup skipped",
+			description: "When RemoveResourceFromFile fails, cleanupOrphanedFolders skips folder removal",
+			previousRef: "old-ref",
+			currentRef:  "new-ref",
+			changes: []repository.VersionedFileChange{
+				{Action: repository.FileActionDeleted, Path: "dashboards/file1.json", PreviousRef: "old-ref"},
+			},
+			setupMocks: func(repo *repository.MockVersioned, repoResources *resources.MockRepositoryResources, progress *jobs.MockJobProgressRecorder) {
+				// File deletion fails
+				progress.On("IsNestedUnderFailedCreation", "dashboards/file1.json").Return(false).Once()
+				repoResources.On("RemoveResourceFromFile", mock.Anything, "dashboards/file1.json", "old-ref").
+					Return("dashboard-1", "folder-uid", schema.GroupVersionKind{Kind: "Dashboard"}, fmt.Errorf("permission denied")).Once()
+
+				// Error recorded, automatically tracked in failedDeletions
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "dashboards/file1.json" &&
+						r.Action == repository.FileActionDeleted &&
+						r.Error != nil
+				})).Return().Once()
+
+				// During cleanup, folder deletion is skipped
+				progress.On("HasFailedDeletionsUnder", "dashboards/").Return(true).Once()
+
+				// Note: RemoveFolder should NOT be called (verified via AssertNotCalled in test)
+			},
+		},
+		{
+			name:        "deletion proceeds despite creation failure",
+			description: "When folder1/ creation fails, deletion of folder1/old.json still proceeds",
+			previousRef: "old-ref",
+			currentRef:  "new-ref",
+			changes: []repository.VersionedFileChange{
+				{Action: repository.FileActionCreated, Path: "folder1/new.json", Ref: "new-ref"},
+				{Action: repository.FileActionDeleted, Path: "folder1/old.json", PreviousRef: "old-ref"},
+			},
+			setupMocks: func(repo *repository.MockVersioned, repoResources *resources.MockRepositoryResources, progress *jobs.MockJobProgressRecorder) {
+				// Creation fails
+				progress.On("IsNestedUnderFailedCreation", "folder1/new.json").Return(false).Once()
+				folderErr := &resources.PathCreationError{Path: "folder1/", Err: fmt.Errorf("permission denied")}
+				repoResources.On("WriteResourceFromFile", mock.Anything, "folder1/new.json", "new-ref").
+					Return("", schema.GroupVersionKind{}, folderErr).Once()
+
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/new.json" && r.Error != nil
+				})).Return().Once()
+
+				// Deletion proceeds (NOT checking IsNestedUnderFailedCreation for deletions)
+				progress.On("IsNestedUnderFailedCreation", "folder1/old.json").Return(false).Once()
+				repoResources.On("RemoveResourceFromFile", mock.Anything, "folder1/old.json", "old-ref").
+					Return("old-resource", "", schema.GroupVersionKind{Kind: "Dashboard"}, nil).Once()
+
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/old.json" &&
+						r.Action == repository.FileActionDeleted &&
+						r.Error == nil // Deletion succeeds!
+				})).Return().Once()
+			},
+		},
+		{
+			name:        "multi-level nesting cascade",
+			description: "When level1/ fails, level1/level2/level3/file.json is also skipped",
+			previousRef: "old-ref",
+			currentRef:  "new-ref",
+			changes: []repository.VersionedFileChange{
+				{Action: repository.FileActionCreated, Path: "level1/file.txt", Ref: "new-ref"},
+				{Action: repository.FileActionCreated, Path: "level1/level2/file.txt", Ref: "new-ref"},
+				{Action: repository.FileActionCreated, Path: "level1/level2/level3/file.txt", Ref: "new-ref"},
+			},
+			setupMocks: func(repo *repository.MockVersioned, repoResources *resources.MockRepositoryResources, progress *jobs.MockJobProgressRecorder) {
+				// First file triggers level1/ failure
+				folderErr := &resources.PathCreationError{Path: "level1/", Err: fmt.Errorf("permission denied")}
+				repoResources.On("EnsureFolderPathExist", mock.Anything, "level1/").Return("", folderErr).Once()
+
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "level1/file.txt" && r.Action == repository.FileActionIgnored
+				})).Return().Once()
+
+				// All nested files are skipped
+				for _, path := range []string{"level1/level2/file.txt", "level1/level2/level3/file.txt"} {
+					progress.On("IsNestedUnderFailedCreation", path).Return(true).Once()
+					progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+						return r.Path == path && r.Action == repository.FileActionIgnored
+					})).Return().Once()
+				}
+			},
+		},
+		{
+			name:        "mixed success and failure",
+			description: "When success/ works and failure/ fails, only failure/* are skipped",
+			previousRef: "old-ref",
+			currentRef:  "new-ref",
+			changes: []repository.VersionedFileChange{
+				{Action: repository.FileActionCreated, Path: "success/file1.json", Ref: "new-ref"},
+				{Action: repository.FileActionCreated, Path: "success/nested/file2.json", Ref: "new-ref"},
+				{Action: repository.FileActionCreated, Path: "failure/file3.txt", Ref: "new-ref"},
+				{Action: repository.FileActionCreated, Path: "failure/nested/file4.txt", Ref: "new-ref"},
+			},
+			setupMocks: func(repo *repository.MockVersioned, repoResources *resources.MockRepositoryResources, progress *jobs.MockJobProgressRecorder) {
+				// Success path works
+				progress.On("IsNestedUnderFailedCreation", "success/file1.json").Return(false).Once()
+				repoResources.On("WriteResourceFromFile", mock.Anything, "success/file1.json", "new-ref").
+					Return("resource-1", schema.GroupVersionKind{Kind: "Dashboard"}, nil).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "success/file1.json" && r.Error == nil
+				})).Return().Once()
+
+				progress.On("IsNestedUnderFailedCreation", "success/nested/file2.json").Return(false).Once()
+				repoResources.On("WriteResourceFromFile", mock.Anything, "success/nested/file2.json", "new-ref").
+					Return("resource-2", schema.GroupVersionKind{Kind: "Dashboard"}, nil).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "success/nested/file2.json" && r.Error == nil
+				})).Return().Once()
+
+				// Failure path fails
+				progress.On("IsNestedUnderFailedCreation", "failure/file3.txt").Return(false).Once()
+				folderErr := &resources.PathCreationError{Path: "failure/", Err: fmt.Errorf("disk full")}
+				repoResources.On("EnsureFolderPathExist", mock.Anything, "failure/").Return("", folderErr).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "failure/file3.txt" && r.Action == repository.FileActionIgnored
+				})).Return().Once()
+
+				// Nested file in failure path is skipped
+				progress.On("IsNestedUnderFailedCreation", "failure/nested/file4.txt").Return(true).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "failure/nested/file4.txt" && r.Action == repository.FileActionIgnored
+				})).Return().Once()
+			},
+		},
+		{
+			name:        "rename with failed destination folder",
+			description: "When RenameResourceFile fails with PathCreationError for destination, rename is skipped",
+			previousRef: "old-ref",
+			currentRef:  "new-ref",
+			changes: []repository.VersionedFileChange{
+				{
+					Action:       repository.FileActionRenamed,
+					Path:         "newfolder/file.json",
+					PreviousPath: "oldfolder/file.json",
+					Ref:          "new-ref",
+					PreviousRef:  "old-ref",
+				},
+			},
+			setupMocks: func(repo *repository.MockVersioned, repoResources *resources.MockRepositoryResources, progress *jobs.MockJobProgressRecorder) {
+				// Rename fails with PathCreationError for destination folder
+				progress.On("IsNestedUnderFailedCreation", "newfolder/file.json").Return(false).Once()
+				folderErr := &resources.PathCreationError{Path: "newfolder/", Err: fmt.Errorf("permission denied")}
+				repoResources.On("RenameResourceFile", mock.Anything, "oldfolder/file.json", "old-ref", "newfolder/file.json", "new-ref").
+					Return("", "", schema.GroupVersionKind{}, folderErr).Once()
+
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "newfolder/file.json" &&
+						r.Action == repository.FileActionRenamed &&
+						r.Error != nil
+				})).Return().Once()
+			},
+		},
+		{
+			name:        "renamed file still checked, subsequent nested resources skipped",
+			description: "After rename fails for folder1/file.json, other folder1/* files are skipped",
+			previousRef: "old-ref",
+			currentRef:  "new-ref",
+			changes: []repository.VersionedFileChange{
+				{Action: repository.FileActionRenamed, Path: "folder1/file1.json", PreviousPath: "old/file1.json", Ref: "new-ref", PreviousRef: "old-ref"},
+				{Action: repository.FileActionCreated, Path: "folder1/file2.json", Ref: "new-ref"},
+			},
+			setupMocks: func(repo *repository.MockVersioned, repoResources *resources.MockRepositoryResources, progress *jobs.MockJobProgressRecorder) {
+				// Rename is NOT skipped for creation failures (it's checking the destination path)
+				progress.On("IsNestedUnderFailedCreation", "folder1/file1.json").Return(true).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/file1.json" &&
+						r.Action == repository.FileActionIgnored &&
+						r.Error.Error() == "skipped: parent folder creation failed"
+				})).Return().Once()
+
+				// Second file also skipped
+				progress.On("IsNestedUnderFailedCreation", "folder1/file2.json").Return(true).Once()
+				progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+					return r.Path == "folder1/file2.json" && r.Action == repository.FileActionIgnored
+				})).Return().Once()
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			repo := repository.NewMockVersioned(t)
+			repoResources := resources.NewMockRepositoryResources(t)
+			progress := jobs.NewMockJobProgressRecorder(t)
+
+			repo.On("CompareFiles", mock.Anything, tt.previousRef, tt.currentRef).Return(tt.changes, nil)
+			progress.On("SetTotal", mock.Anything, len(tt.changes)).Return()
+			progress.On("SetMessage", mock.Anything, "replicating versioned changes").Return()
+			progress.On("SetMessage", mock.Anything, "versioned changes replicated").Return()
+			progress.On("TooManyErrors").Return(nil).Maybe()
+
+			tt.setupMocks(repo, repoResources, progress)
+
+			err := IncrementalSync(context.Background(), repo, tt.previousRef, tt.currentRef, repoResources, progress, tracing.NewNoopTracerService(), jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+
+			if tt.expectError {
+				require.Error(t, err)
+				if tt.errorContains != "" {
+					require.Contains(t, err.Error(), tt.errorContains)
+				}
+			} else {
+				require.NoError(t, err)
+			}
+
+			progress.AssertExpectations(t)
+			repoResources.AssertExpectations(t)
+			// For deletion tests, verify RemoveFolder was NOT called
+			if tt.name == "file deletion fails, folder cleanup skipped" {
+				repoResources.AssertNotCalled(t, "RemoveFolder", mock.Anything, mock.Anything)
+			}
+		})
+	}
+}

pkg/registry/apis/provisioning/jobs/sync/incremental_test.go

@@ -522,3 +522,249 @@ func TestIncrementalSync_CleanupOrphanedFolders(t *testing.T) {
 		})
 	}
 }
+
+// TestIncrementalSync_HierarchicalErrorHandling_FailedFolderCreation tests nested resource skipping
+func TestIncrementalSync_HierarchicalErrorHandling_FailedFolderCreation(t *testing.T) {
+	repo := repository.NewMockVersioned(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+
+	changes := []repository.VersionedFileChange{
+		{Action: repository.FileActionCreated, Path: "unsupported/file.txt", Ref: "new-ref"},
+		{Action: repository.FileActionCreated, Path: "unsupported/subfolder/file2.txt", Ref: "new-ref"},
+		{Action: repository.FileActionCreated, Path: "unsupported/file3.json", Ref: "new-ref"},
+		{Action: repository.FileActionCreated, Path: "other/file.json", Ref: "new-ref"},
+	}
+
+	repo.On("CompareFiles", mock.Anything, "old-ref", "new-ref").Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, 4).Return()
+	progress.On("SetMessage", mock.Anything, "replicating versioned changes").Return()
+	progress.On("SetMessage", mock.Anything, "versioned changes replicated").Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	folderErr := &resources.PathCreationError{Path: "unsupported/", Err: fmt.Errorf("permission denied")}
+	repoResources.On("EnsureFolderPathExist", mock.Anything, "unsupported/").Return("", folderErr).Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "unsupported/file.txt" && r.Action == repository.FileActionIgnored && r.Error != nil
+	})).Return().Once()
+
+	progress.On("IsNestedUnderFailedCreation", "unsupported/subfolder/file2.txt").Return(true).Once()
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "unsupported/subfolder/file2.txt" && r.Action == repository.FileActionIgnored &&
+			r.Error != nil && r.Error.Error() == "skipped: parent folder creation failed"
+	})).Return().Once()
+
+	progress.On("IsNestedUnderFailedCreation", "unsupported/file3.json").Return(true).Once()
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "unsupported/file3.json" && r.Action == repository.FileActionIgnored &&
+			r.Error != nil && r.Error.Error() == "skipped: parent folder creation failed"
+	})).Return().Once()
+
+	progress.On("IsNestedUnderFailedCreation", "other/file.json").Return(false).Once()
+	repoResources.On("WriteResourceFromFile", mock.Anything, "other/file.json", "new-ref").
+		Return("test-resource", schema.GroupVersionKind{Kind: "Dashboard"}, nil).Once()
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "other/file.json" && r.Action == repository.FileActionCreated && r.Error == nil
+	})).Return().Once()
+
+	err := IncrementalSync(context.Background(), repo, "old-ref", "new-ref", repoResources, progress, tracing.NewNoopTracerService(), jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+}
+
+// TestIncrementalSync_HierarchicalErrorHandling_FailedFileDeletion tests folder cleanup prevention
+func TestIncrementalSync_HierarchicalErrorHandling_FailedFileDeletion(t *testing.T) {
+	mockVersioned := repository.NewMockVersioned(t)
+	mockReader := repository.NewMockReader(t)
+	repo := &compositeRepo{MockVersioned: mockVersioned, MockReader: mockReader}
+	repoResources := resources.NewMockRepositoryResources(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+
+	changes := []repository.VersionedFileChange{
+		{Action: repository.FileActionDeleted, Path: "dashboards/file1.json", PreviousRef: "old-ref"},
+	}
+
+	mockVersioned.On("CompareFiles", mock.Anything, "old-ref", "new-ref").Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, 1).Return()
+	progress.On("SetMessage", mock.Anything, "replicating versioned changes").Return()
+	progress.On("SetMessage", mock.Anything, "versioned changes replicated").Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	progress.On("IsNestedUnderFailedCreation", "dashboards/file1.json").Return(false).Once()
+	repoResources.On("RemoveResourceFromFile", mock.Anything, "dashboards/file1.json", "old-ref").
+		Return("dashboard-1", "folder-uid", schema.GroupVersionKind{Kind: "Dashboard"}, fmt.Errorf("permission denied")).Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "dashboards/file1.json" && r.Action == repository.FileActionDeleted &&
+			r.Error != nil && r.Error.Error() == "removing resource from file dashboards/file1.json: permission denied"
+	})).Return().Once()
+
+	progress.On("HasFailedDeletionsUnder", "dashboards/").Return(true).Once()
+
+	err := IncrementalSync(context.Background(), repo, "old-ref", "new-ref", repoResources, progress, tracing.NewNoopTracerService(), jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+	repoResources.AssertNotCalled(t, "RemoveFolder", mock.Anything, mock.Anything)
+}
+
+// TestIncrementalSync_HierarchicalErrorHandling_DeletionNotAffectedByCreationFailure tests deletions proceed despite creation failures
+func TestIncrementalSync_HierarchicalErrorHandling_DeletionNotAffectedByCreationFailure(t *testing.T) {
+	repo := repository.NewMockVersioned(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+
+	changes := []repository.VersionedFileChange{
+		{Action: repository.FileActionCreated, Path: "folder1/file.json", Ref: "new-ref"},
+		{Action: repository.FileActionDeleted, Path: "folder1/old.json", PreviousRef: "old-ref"},
+	}
+
+	repo.On("CompareFiles", mock.Anything, "old-ref", "new-ref").Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, 2).Return()
+	progress.On("SetMessage", mock.Anything, "replicating versioned changes").Return()
+	progress.On("SetMessage", mock.Anything, "versioned changes replicated").Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	// Creation fails
+	progress.On("IsNestedUnderFailedCreation", "folder1/file.json").Return(false).Once()
+	repoResources.On("WriteResourceFromFile", mock.Anything, "folder1/file.json", "new-ref").
+		Return("", schema.GroupVersionKind{}, &resources.PathCreationError{Path: "folder1/", Err: fmt.Errorf("permission denied")}).Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "folder1/file.json" && r.Error != nil
+	})).Return().Once()
+
+	// Deletion should NOT be skipped (not checking IsNestedUnderFailedCreation for deletions)
+	progress.On("IsNestedUnderFailedCreation", "folder1/old.json").Return(false).Once()
+	repoResources.On("RemoveResourceFromFile", mock.Anything, "folder1/old.json", "old-ref").
+		Return("old-resource", "", schema.GroupVersionKind{Kind: "Dashboard"}, nil).Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "folder1/old.json" && r.Action == repository.FileActionDeleted && r.Error == nil
+	})).Return().Once()
+
+	err := IncrementalSync(context.Background(), repo, "old-ref", "new-ref", repoResources, progress, tracing.NewNoopTracerService(), jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+}
+
+// TestIncrementalSync_HierarchicalErrorHandling_MultiLevelNesting tests multi-level cascade
+func TestIncrementalSync_HierarchicalErrorHandling_MultiLevelNesting(t *testing.T) {
+	repo := repository.NewMockVersioned(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+
+	changes := []repository.VersionedFileChange{
+		{Action: repository.FileActionCreated, Path: "level1/file.txt", Ref: "new-ref"},
+		{Action: repository.FileActionCreated, Path: "level1/level2/file.txt", Ref: "new-ref"},
+		{Action: repository.FileActionCreated, Path: "level1/level2/level3/file.txt", Ref: "new-ref"},
+	}
+
+	repo.On("CompareFiles", mock.Anything, "old-ref", "new-ref").Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, 3).Return()
+	progress.On("SetMessage", mock.Anything, "replicating versioned changes").Return()
+	progress.On("SetMessage", mock.Anything, "versioned changes replicated").Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	folderErr := &resources.PathCreationError{Path: "level1/", Err: fmt.Errorf("permission denied")}
+	repoResources.On("EnsureFolderPathExist", mock.Anything, "level1/").Return("", folderErr).Once()
+
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "level1/file.txt" && r.Action == repository.FileActionIgnored
+	})).Return().Once()
+
+	progress.On("IsNestedUnderFailedCreation", "level1/level2/file.txt").Return(true).Once()
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "level1/level2/file.txt" && r.Action == repository.FileActionIgnored &&
+			r.Error != nil && r.Error.Error() == "skipped: parent folder creation failed"
+	})).Return().Once()
+
+	progress.On("IsNestedUnderFailedCreation", "level1/level2/level3/file.txt").Return(true).Once()
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "level1/level2/level3/file.txt" && r.Action == repository.FileActionIgnored &&
+			r.Error != nil && r.Error.Error() == "skipped: parent folder creation failed"
+	})).Return().Once()
+
+	err := IncrementalSync(context.Background(), repo, "old-ref", "new-ref", repoResources, progress, tracing.NewNoopTracerService(), jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+}
+
+// TestIncrementalSync_HierarchicalErrorHandling_MixedSuccessAndFailure tests partial failures
+func TestIncrementalSync_HierarchicalErrorHandling_MixedSuccessAndFailure(t *testing.T) {
+	repo := repository.NewMockVersioned(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+
+	changes := []repository.VersionedFileChange{
+		{Action: repository.FileActionCreated, Path: "success/file1.json", Ref: "new-ref"},
+		{Action: repository.FileActionCreated, Path: "success/nested/file2.json", Ref: "new-ref"},
+		{Action: repository.FileActionCreated, Path: "failure/file3.txt", Ref: "new-ref"},
+		{Action: repository.FileActionCreated, Path: "failure/nested/file4.txt", Ref: "new-ref"},
+	}
+
+	repo.On("CompareFiles", mock.Anything, "old-ref", "new-ref").Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, 4).Return()
+	progress.On("SetMessage", mock.Anything, "replicating versioned changes").Return()
+	progress.On("SetMessage", mock.Anything, "versioned changes replicated").Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	progress.On("IsNestedUnderFailedCreation", "success/file1.json").Return(false).Once()
+	repoResources.On("WriteResourceFromFile", mock.Anything, "success/file1.json", "new-ref").
+		Return("resource-1", schema.GroupVersionKind{Kind: "Dashboard"}, nil).Once()
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "success/file1.json" && r.Action == repository.FileActionCreated && r.Error == nil
+	})).Return().Once()
+
+	progress.On("IsNestedUnderFailedCreation", "success/nested/file2.json").Return(false).Once()
+	repoResources.On("WriteResourceFromFile", mock.Anything, "success/nested/file2.json", "new-ref").
+		Return("resource-2", schema.GroupVersionKind{Kind: "Dashboard"}, nil).Once()
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "success/nested/file2.json" && r.Action == repository.FileActionCreated && r.Error == nil
+	})).Return().Once()
+
+	folderErr := &resources.PathCreationError{Path: "failure/", Err: fmt.Errorf("disk full")}
+	progress.On("IsNestedUnderFailedCreation", "failure/file3.txt").Return(false).Once()
+	repoResources.On("EnsureFolderPathExist", mock.Anything, "failure/").Return("", folderErr).Once()
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "failure/file3.txt" && r.Action == repository.FileActionIgnored
+	})).Return().Once()
+
+	progress.On("IsNestedUnderFailedCreation", "failure/nested/file4.txt").Return(true).Once()
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "failure/nested/file4.txt" && r.Action == repository.FileActionIgnored &&
+			r.Error != nil && r.Error.Error() == "skipped: parent folder creation failed"
+	})).Return().Once()
+
+	err := IncrementalSync(context.Background(), repo, "old-ref", "new-ref", repoResources, progress, tracing.NewNoopTracerService(), jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+	repoResources.AssertExpectations(t)
+}
+
+// TestIncrementalSync_HierarchicalErrorHandling_RenameWithFailedFolderCreation tests rename operations affected by folder failures
+func TestIncrementalSync_HierarchicalErrorHandling_RenameWithFailedFolderCreation(t *testing.T) {
+	repo := repository.NewMockVersioned(t)
+	repoResources := resources.NewMockRepositoryResources(t)
+	progress := jobs.NewMockJobProgressRecorder(t)
+
+	changes := []repository.VersionedFileChange{
+		{Action: repository.FileActionRenamed, Path: "newfolder/file.json", PreviousPath: "oldfolder/file.json", Ref: "new-ref", PreviousRef: "old-ref"},
+	}
+
+	repo.On("CompareFiles", mock.Anything, "old-ref", "new-ref").Return(changes, nil)
+	progress.On("SetTotal", mock.Anything, 1).Return()
+	progress.On("SetMessage", mock.Anything, "replicating versioned changes").Return()
+	progress.On("SetMessage", mock.Anything, "versioned changes replicated").Return()
+	progress.On("TooManyErrors").Return(nil).Maybe()
+
+	progress.On("IsNestedUnderFailedCreation", "newfolder/file.json").Return(true).Once()
+	progress.On("Record", mock.Anything, mock.MatchedBy(func(r jobs.JobResourceResult) bool {
+		return r.Path == "newfolder/file.json" && r.Action == repository.FileActionIgnored &&
+			r.Error != nil && r.Error.Error() == "skipped: parent folder creation failed"
+	})).Return().Once()
+
+	err := IncrementalSync(context.Background(), repo, "old-ref", "new-ref", repoResources, progress, tracing.NewNoopTracerService(), jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()))
+	require.NoError(t, err)
+	progress.AssertExpectations(t)
+}

pkg/registry/apis/provisioning/jobs/worker_mock.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v2.52.4. DO NOT EDIT.
+// Code generated by mockery v2.53.4. DO NOT EDIT.
 
 package jobs
 

pkg/registry/apis/provisioning/resources/folders.go

@@ -73,7 +73,11 @@ func (fm *FolderManager) EnsureFolderPathExist(ctx context.Context, filePath str
 		}
 
 		if err := fm.EnsureFolderExists(ctx, f, parent); err != nil {
-			return fmt.Errorf("ensure folder exists: %w", err)
+			// Wrap in PathCreationError to indicate which path failed
+			return &PathCreationError{
+				Path: f.Path,
+				Err:  fmt.Errorf("ensure folder exists: %w", err),
+			}
 		}
 
 		fm.tree.Add(f, parent)

pkg/registry/apis/provisioning/resources/resources.go

@@ -27,6 +27,21 @@ var (
 	ErrMissingName         = field.Required(field.NewPath("name", "metadata", "name"), "missing name in resource")
 )
 
+// PathCreationError represents an error that occurred while creating a folder path.
+// It contains the path that failed and the underlying error.
+type PathCreationError struct {
+	Path string
+	Err  error
+}
+
+func (e *PathCreationError) Unwrap() error {
+	return e.Err
+}
+
+func (e *PathCreationError) Error() string {
+	return fmt.Sprintf("failed to create path %s: %v", e.Path, e.Err)
+}
+
 // NewResourceOwnershipConflictError creates a BadRequest error for when a resource
 // is owned by a different repository or manager and cannot be modified
 func NewResourceOwnershipConflictError(resourceName string, currentManager utils.ManagerProperties, requestingManager utils.ManagerProperties) error {