Provisioning: Fix history write for expired jobs (#113764)

* refactor: Move job cleanup to separate controller and fix history write

- Created JobCleanupController in apps/provisioning/pkg/controller
- Separated cleanup logic from ConcurrentJobDriver
- Fixed bug where expired jobs were not written to history
- Added comprehensive tests with 93.8% coverage
- Removed cleanup interval parameter from ConcurrentJobDriver
- Cleanup now properly follows Complete + WriteJob pattern

Fixes expired jobs being lost instead of archived

* refactor: Update lease renewal interval to use jobExpiry variable

- Changed the lease renewal interval in the GetPostStartHooks method to utilize the jobExpiry variable for improved clarity and maintainability.

* Format code

* Fix Unix milliseconds

* fix: correct Unix timestamp assertions and remove duplicate test expectations

- Changed Unix() to UnixMilli() for correct millisecond timestamp validation
- Removed duplicate store.AssertExpectations(t) calls throughout tests

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

@@ -14,7 +14,6 @@ import (
 type ConcurrentJobDriver struct {
 	numDrivers           int
 	jobTimeout           time.Duration
-	cleanupInterval      time.Duration
 	jobInterval          time.Duration
 	leaseRenewalInterval time.Duration
 	store                Store
@@ -27,7 +26,7 @@ type ConcurrentJobDriver struct {
 // NewConcurrentJobDriver creates a new concurrent job driver that spawns multiple job drivers.
 func NewConcurrentJobDriver(
 	numDrivers int,
-	jobTimeout, cleanupInterval, jobInterval, leaseRenewalInterval time.Duration,
+	jobTimeout, jobInterval, leaseRenewalInterval time.Duration,
 	store Store,
 	repoGetter RepoGetter,
 	historicJobs HistoryWriter,
@@ -45,24 +44,12 @@ func NewConcurrentJobDriver(
 	if leaseRenewalInterval < 5*time.Second {
 		leaseRenewalInterval = 5 * time.Second
 	}
-	// For lease-based cleanup, run at most every 3-4 lease renewal intervals
-	// to detect expired leases promptly but not too aggressively
-	if cleanupInterval <= 0 {
-		cleanupInterval = leaseRenewalInterval * 3
-	}
-	if cleanupInterval < 30*time.Second {
-		cleanupInterval = 30 * time.Second // Minimum cleanup interval
-	}
-	if cleanupInterval > 5*time.Minute {
-		cleanupInterval = 5 * time.Minute // Maximum cleanup interval
-	}
 
 	recordConcurrentDriverMetric(registry, numDrivers)
 
 	return &ConcurrentJobDriver{
 		numDrivers:           numDrivers,
 		jobTimeout:           jobTimeout,
-		cleanupInterval:      cleanupInterval,
 		jobInterval:          jobInterval,
 		leaseRenewalInterval: leaseRenewalInterval,
 		store:                store,
@@ -73,43 +60,17 @@ func NewConcurrentJobDriver(
 	}, nil
 }
 
-// Run starts multiple job drivers concurrently and handles cleanup coordination.
+// Run starts multiple job drivers concurrently.
 // This is a blocking function that will run until the context is canceled or an error occurs.
 //
 // Note: This function intentionally does NOT create a tracing span because it runs indefinitely
-// until shutdown. Individual job processing and cleanup operations already have their own spans.
+// until shutdown. Individual job processing operations already have their own spans.
 func (c *ConcurrentJobDriver) Run(ctx context.Context) error {
 	logger := logging.FromContext(ctx).With("logger", "concurrent-job-driver", "num_drivers", c.numDrivers)
-	logger.Info("start concurrent job driver", "num_drivers", c.numDrivers, "cleanup_interval", c.cleanupInterval)
-
-	// Set up cleanup ticker - runs more frequently with lease-based approach
-	cleanupTicker := time.NewTicker(c.cleanupInterval)
-	defer cleanupTicker.Stop()
-
-	// Initial cleanup
-	if err := c.store.Cleanup(ctx); err != nil {
-		logger.Error("failed initial cleanup", "error", err)
-	}
+	logger.Info("start concurrent job driver", "num_drivers", c.numDrivers)
 
 	var wg sync.WaitGroup
-	errChan := make(chan error, c.numDrivers+1) // +1 for cleanup goroutine
-
-	// Start cleanup goroutine
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		for {
-			select {
-			case <-cleanupTicker.C:
-				if err := c.store.Cleanup(ctx); err != nil {
-					logger.Error("failed cleanup", "error", err)
-				}
-			case <-ctx.Done():
-				logger.Debug("cleanup routine stopped")
-				return
-			}
-		}
-	}()
+	errChan := make(chan error, c.numDrivers)
 
 	// Start driver goroutines
 	for i := 0; i < c.numDrivers; i++ {

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

@@ -31,14 +31,10 @@ type Store interface {
 	// The err may be ErrNoJobs if there are no jobs to claim.
 	Claim(ctx context.Context) (job *provisioning.Job, rollback func(), err error)
 
-	// Complete marks a job as completed and moves it to the historic job store.
-	// When in the historic store, there is no more claim on the job.
+	// Complete marks a job as completed and removes it from the active job store.
+	// Callers are responsible for writing the job to history after calling this.
 	Complete(ctx context.Context, job *provisioning.Job) error
 
-	// Cleanup should be called periodically to clean up abandoned jobs.
-	// An abandoned job is one that has been claimed by a worker, but the worker has not updated the job in a while.
-	Cleanup(ctx context.Context) error
-
 	// Update saves the job back to the store.
 	Update(ctx context.Context, job *provisioning.Job) (*provisioning.Job, error)
 
@@ -48,6 +44,10 @@ type Store interface {
 
 	// Get retrieves a job by name for conflict resolution.
 	Get(ctx context.Context, namespace, name string) (*provisioning.Job, error)
+
+	// ListExpiredJobs lists jobs with expired leases (claim timestamp older than the given time).
+	// Returns jobs in batches up to the specified limit.
+	ListExpiredJobs(ctx context.Context, expiredBefore time.Time, limit int) ([]*provisioning.Job, error)
 }
 
 // jobDriver drives jobs to completion and manages the job queue.

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

@@ -0,0 +1,177 @@
+package jobs
+
+import (
+	"context"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+
+	"github.com/grafana/grafana-app-sdk/logging"
+	"github.com/grafana/grafana/apps/provisioning/pkg/apifmt"
+	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/infra/tracing"
+)
+
+// JobCleanupController handles cleanup of expired/abandoned jobs.
+type JobCleanupController struct {
+	store           Store
+	historicJobs    HistoryWriter
+	clock           func() time.Time
+	expiry          time.Duration
+	cleanupInterval time.Duration
+}
+
+// NewJobCleanupController creates a new job cleanup controller.
+func NewJobCleanupController(
+	store Store,
+	historicJobs HistoryWriter,
+	expiry time.Duration,
+) *JobCleanupController {
+	// Calculate cleanup interval based on expiry duration
+	// Run cleanup every 3-4 expiry intervals to detect expired leases promptly but not too aggressively
+	cleanupInterval := expiry * 3
+
+	// Enforce minimum and maximum bounds
+	if cleanupInterval < 30*time.Second {
+		cleanupInterval = 30 * time.Second
+	}
+	if cleanupInterval > 5*time.Minute {
+		cleanupInterval = 5 * time.Minute
+	}
+
+	return &JobCleanupController{
+		store:           store,
+		historicJobs:    historicJobs,
+		clock:           time.Now,
+		expiry:          expiry,
+		cleanupInterval: cleanupInterval,
+	}
+}
+
+// Run starts the cleanup loop that runs at an appropriate interval.
+// This is a blocking function that runs until the context is canceled.
+func (c *JobCleanupController) Run(ctx context.Context) error {
+	logger := logging.FromContext(ctx).With("logger", "job-cleanup-controller")
+	ctx = logging.Context(ctx, logger)
+
+	// Set up provisioning identity to access jobs across all namespaces
+	ctx, _, err := identity.WithProvisioningIdentity(ctx, "*")
+	if err != nil {
+		return apifmt.Errorf("failed to grant provisioning identity for cleanup: %w", err)
+	}
+
+	logger.Info("starting job cleanup controller", "cleanup_interval", c.cleanupInterval, "expiry", c.expiry)
+
+	// Initial cleanup
+	if err := c.Cleanup(ctx); err != nil {
+		logger.Error("failed to clean up jobs at start", "error", err)
+	}
+
+	ticker := time.NewTicker(c.cleanupInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			if err := c.Cleanup(ctx); err != nil {
+				logger.Error("failed to cleanup jobs", "error", err)
+			}
+		case <-ctx.Done():
+			logger.Info("job cleanup controller stopping")
+			return ctx.Err()
+		}
+	}
+}
+
+// Cleanup finds jobs with expired leases and marks them as failed.
+// This should be called periodically to clean up jobs from crashed workers.
+func (c *JobCleanupController) Cleanup(ctx context.Context) error {
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.cleanup")
+	defer span.End()
+
+	startTime := c.clock()
+	logger := logging.FromContext(ctx)
+
+	// Find jobs with expired leases
+	expiredBefore := c.clock().Add(-c.expiry)
+
+	// Process in batches of 100 to avoid overwhelming the system
+	const batchSize = 100
+	jobs, err := c.store.ListExpiredJobs(ctx, expiredBefore, batchSize)
+	if err != nil {
+		span.RecordError(err)
+		return apifmt.Errorf("failed to list jobs with expired leases: %w", err)
+	}
+
+	// If no jobs found, cleanup is complete
+	if len(jobs) == 0 {
+		duration := c.clock().Sub(startTime)
+		span.SetAttributes(
+			attribute.Int("count", 0),
+			attribute.Int64("duration_ms", duration.Milliseconds()),
+		)
+		return nil
+	}
+
+	logger.Info("cleaning up expired jobs", "count", len(jobs))
+
+	for _, job := range jobs {
+		if err := c.cleanUpExpiredJob(ctx, job); err != nil {
+			// Log error but continue processing other jobs
+			logger.Error("failed to clean up expired job", "error", err, "job", job.GetName(), "namespace", job.GetNamespace())
+		}
+	}
+
+	duration := c.clock().Sub(startTime)
+	logger.Info("cleanup complete", "duration", duration, "count", len(jobs))
+
+	span.SetAttributes(
+		attribute.Int("count", len(jobs)),
+		attribute.Int64("duration_ms", duration.Milliseconds()),
+	)
+
+	return nil
+}
+
+// cleanUpExpiredJob marks a single expired job as failed and archives it.
+func (c *JobCleanupController) cleanUpExpiredJob(ctx context.Context, job *provisioning.Job) error {
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.cleanup.complete_expired_job")
+	defer span.End()
+
+	// Mark job as failed due to lease expiry
+	jobCopy := job.DeepCopy()
+	jobCopy.Status.State = provisioning.JobStateError
+	jobCopy.Status.Message = "Job failed due to lease expiry - worker may have crashed or lost connection"
+	jobCopy.Status.Finished = c.clock().UnixMilli()
+
+	span.SetAttributes(
+		attribute.String("job.name", jobCopy.GetName()),
+		attribute.String("job.namespace", jobCopy.GetNamespace()),
+		attribute.String("job.repository", jobCopy.Spec.Repository),
+		attribute.String("job.action", string(jobCopy.Spec.Action)),
+	)
+
+	jobLogger := logging.FromContext(ctx).With("namespace", jobCopy.GetNamespace(), "job", jobCopy.GetName(), "action", jobCopy.Spec.Action)
+
+	// Delete from active job store first
+	if err := c.store.Complete(ctx, jobCopy); err != nil {
+		span.RecordError(err)
+		return apifmt.Errorf("failed to complete expired job: %w", err)
+	}
+
+	// Remove the claim label before archiving
+	if jobCopy.Labels != nil {
+		delete(jobCopy.Labels, LabelJobClaim)
+	}
+
+	// Write to history after deleting from active store (matching driver.go pattern)
+	if err := c.historicJobs.WriteJob(ctx, jobCopy); err != nil {
+		span.RecordError(err)
+		jobLogger.Warn("failed to write expired job to history", "error", err)
+		// Job was already deleted, so we can't recover from this
+	}
+
+	jobLogger.Debug("cleaned up expired job")
+	return nil
+}

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

@@ -0,0 +1,465 @@
+package jobs
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestNewJobCleanupController(t *testing.T) {
+	store := &MockStore{}
+	historyWriter := &MockHistoryWriter{}
+
+	t.Run("creates controller with default cleanup interval", func(t *testing.T) {
+		expiry := 30 * time.Second
+		controller := NewJobCleanupController(store, historyWriter, expiry)
+
+		assert.NotNil(t, controller)
+		assert.Equal(t, expiry, controller.expiry)
+		// Cleanup interval should be 3x expiry = 90 seconds
+		assert.Equal(t, 90*time.Second, controller.cleanupInterval)
+	})
+
+	t.Run("enforces minimum cleanup interval", func(t *testing.T) {
+		// With expiry of 5 seconds, 3x = 15 seconds, but minimum is 30 seconds
+		expiry := 5 * time.Second
+		controller := NewJobCleanupController(store, historyWriter, expiry)
+
+		assert.Equal(t, 30*time.Second, controller.cleanupInterval)
+	})
+
+	t.Run("enforces maximum cleanup interval", func(t *testing.T) {
+		// With expiry of 5 minutes, 3x = 15 minutes, but maximum is 5 minutes
+		expiry := 5 * time.Minute
+		controller := NewJobCleanupController(store, historyWriter, expiry)
+
+		assert.Equal(t, 5*time.Minute, controller.cleanupInterval)
+	})
+}
+
+func TestJobCleanupController_Cleanup(t *testing.T) {
+	t.Run("no expired jobs returns nil", func(t *testing.T) {
+		store := &MockStore{}
+		historyWriter := &MockHistoryWriter{}
+
+		controller := NewJobCleanupController(store, historyWriter, 30*time.Second)
+		ctx := context.Background()
+
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{}, nil)
+
+		err := controller.Cleanup(ctx)
+
+		assert.NoError(t, err)
+		store.AssertExpectations(t)
+		// store.AssertNotCalled(t, "Complete") - not needed with combined Store mock
+		historyWriter.AssertNotCalled(t, "WriteJob")
+	})
+
+	t.Run("error listing expired jobs returns error", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		controller := NewJobCleanupController(store, historyWriter, 30*time.Second)
+		ctx := context.Background()
+
+		expectedErr := errors.New("list failed")
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return(nil, expectedErr)
+
+		err := controller.Cleanup(ctx)
+
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "failed to list jobs with expired leases")
+		store.AssertExpectations(t)
+	})
+
+	t.Run("successfully cleans up expired job", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		controller := NewJobCleanupController(store, historyWriter, 30*time.Second)
+		ctx := context.Background()
+
+		job := &provisioning.Job{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-job",
+				Namespace: "test-ns",
+				Labels: map[string]string{
+					LabelJobClaim: "123456789",
+				},
+			},
+			Spec: provisioning.JobSpec{
+				Repository: "test-repo",
+				Action:     provisioning.JobActionPull,
+			},
+		}
+
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{job}, nil)
+		store.On("Complete", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool {
+			return j.Status.State == provisioning.JobStateError &&
+				j.Status.Message == "Job failed due to lease expiry - worker may have crashed or lost connection"
+		})).Return(nil)
+		historyWriter.On("WriteJob", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool {
+			// Verify claim label was removed before writing to history
+			_, hasLabel := j.Labels[LabelJobClaim]
+			return !hasLabel && j.Status.State == provisioning.JobStateError
+		})).Return(nil)
+
+		err := controller.Cleanup(ctx)
+
+		assert.NoError(t, err)
+		store.AssertExpectations(t)
+		historyWriter.AssertExpectations(t)
+	})
+
+	t.Run("continues on complete error", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		controller := NewJobCleanupController(store, historyWriter, 30*time.Second)
+		ctx := context.Background()
+
+		job1 := &provisioning.Job{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "job-1",
+				Namespace: "test-ns",
+				Labels:    map[string]string{LabelJobClaim: "123"},
+			},
+			Spec: provisioning.JobSpec{
+				Repository: "repo-1",
+				Action:     provisioning.JobActionPull,
+			},
+		}
+		job2 := &provisioning.Job{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "job-2",
+				Namespace: "test-ns",
+				Labels:    map[string]string{LabelJobClaim: "456"},
+			},
+			Spec: provisioning.JobSpec{
+				Repository: "repo-2",
+				Action:     provisioning.JobActionPull,
+			},
+		}
+
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{job1, job2}, nil)
+
+		// First job fails to complete
+		store.On("Complete", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool {
+			return j.Name == "job-1"
+		})).Return(errors.New("complete failed"))
+
+		// Second job succeeds
+		store.On("Complete", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool {
+			return j.Name == "job-2"
+		})).Return(nil)
+		historyWriter.On("WriteJob", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool {
+			return j.Name == "job-2"
+		})).Return(nil)
+
+		err := controller.Cleanup(ctx)
+
+		// Should not return error, continues processing
+		assert.NoError(t, err)
+		store.AssertExpectations(t)
+		historyWriter.AssertExpectations(t)
+	})
+
+	t.Run("continues on history write error", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		controller := NewJobCleanupController(store, historyWriter, 30*time.Second)
+		ctx := context.Background()
+
+		job := &provisioning.Job{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-job",
+				Namespace: "test-ns",
+				Labels:    map[string]string{LabelJobClaim: "123"},
+			},
+			Spec: provisioning.JobSpec{
+				Repository: "test-repo",
+				Action:     provisioning.JobActionPull,
+			},
+		}
+
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{job}, nil)
+		store.On("Complete", mock.Anything, mock.Anything).Return(nil)
+		historyWriter.On("WriteJob", mock.Anything, mock.Anything).Return(errors.New("write failed"))
+
+		err := controller.Cleanup(ctx)
+
+		// Should not return error, just log warning
+		assert.NoError(t, err)
+		store.AssertExpectations(t)
+		historyWriter.AssertExpectations(t)
+	})
+
+	t.Run("sets job status correctly", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		fixedTime := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
+		controller := NewJobCleanupController(store, historyWriter, 30*time.Second)
+		controller.clock = func() time.Time { return fixedTime }
+		ctx := context.Background()
+
+		job := &provisioning.Job{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-job",
+				Namespace: "test-ns",
+				Labels:    map[string]string{LabelJobClaim: "123"},
+			},
+			Spec: provisioning.JobSpec{
+				Repository: "test-repo",
+				Action:     provisioning.JobActionPull,
+			},
+		}
+
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{job}, nil)
+		store.On("Complete", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool {
+			assert.Equal(t, provisioning.JobStateError, j.Status.State)
+			assert.Equal(t, "Job failed due to lease expiry - worker may have crashed or lost connection", j.Status.Message)
+			assert.Equal(t, fixedTime.UnixMilli(), j.Status.Finished)
+			return true
+		})).Return(nil)
+		historyWriter.On("WriteJob", mock.Anything, mock.Anything).Return(nil)
+
+		err := controller.Cleanup(ctx)
+
+		assert.NoError(t, err)
+		store.AssertExpectations(t)
+	})
+
+	t.Run("removes claim label before writing to history", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		controller := NewJobCleanupController(store, historyWriter, 30*time.Second)
+		ctx := context.Background()
+
+		job := &provisioning.Job{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-job",
+				Namespace: "test-ns",
+				Labels: map[string]string{
+					LabelJobClaim: "123456789",
+					"other-label": "value",
+				},
+			},
+			Spec: provisioning.JobSpec{
+				Repository: "test-repo",
+				Action:     provisioning.JobActionPull,
+			},
+		}
+
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{job}, nil)
+		store.On("Complete", mock.Anything, mock.Anything).Return(nil)
+		historyWriter.On("WriteJob", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool {
+			_, hasClaim := j.Labels[LabelJobClaim]
+			_, hasOther := j.Labels["other-label"]
+			assert.False(t, hasClaim, "claim label should be removed")
+			assert.True(t, hasOther, "other labels should be preserved")
+			return !hasClaim && hasOther
+		})).Return(nil)
+
+		err := controller.Cleanup(ctx)
+
+		assert.NoError(t, err)
+		historyWriter.AssertExpectations(t)
+	})
+
+	t.Run("processes multiple expired jobs", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		controller := NewJobCleanupController(store, historyWriter, 30*time.Second)
+		ctx := context.Background()
+
+		jobs := []*provisioning.Job{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "job-1",
+					Namespace: "ns-1",
+					Labels:    map[string]string{LabelJobClaim: "111"},
+				},
+				Spec: provisioning.JobSpec{Repository: "repo-1", Action: provisioning.JobActionPull},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "job-2",
+					Namespace: "ns-2",
+					Labels:    map[string]string{LabelJobClaim: "222"},
+				},
+				Spec: provisioning.JobSpec{Repository: "repo-2", Action: provisioning.JobActionPush},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "job-3",
+					Namespace: "ns-3",
+					Labels:    map[string]string{LabelJobClaim: "333"},
+				},
+				Spec: provisioning.JobSpec{Repository: "repo-3", Action: provisioning.JobActionMigrate},
+			},
+		}
+
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return(jobs, nil)
+		for _, job := range jobs {
+			store.On("Complete", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool {
+				return j.Name == job.Name
+			})).Return(nil)
+			historyWriter.On("WriteJob", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool {
+				return j.Name == job.Name
+			})).Return(nil)
+		}
+
+		err := controller.Cleanup(ctx)
+
+		assert.NoError(t, err)
+		store.AssertExpectations(t)
+		historyWriter.AssertExpectations(t)
+		// Verify all 3 jobs were processed
+		store.AssertNumberOfCalls(t, "Complete", 3)
+		historyWriter.AssertNumberOfCalls(t, "WriteJob", 3)
+	})
+}
+
+func TestJobCleanupController_Run(t *testing.T) {
+	t.Run("runs cleanup on start and periodically", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		// Use short expiry to get short cleanup interval for testing
+		controller := NewJobCleanupController(store, historyWriter, 10*time.Second)
+		// Override to even shorter for test
+		controller.cleanupInterval = 50 * time.Millisecond
+
+		ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+		defer cancel()
+
+		// Expect initial cleanup + periodic cleanups (at least 2, maybe more depending on timing)
+		callCount := 0
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).
+			Return([]*provisioning.Job{}, nil).
+			Run(func(args mock.Arguments) {
+				callCount++
+			}).
+			Maybe() // Allow variable number of calls due to timing
+
+		err := controller.Run(ctx)
+
+		// Should return context.DeadlineExceeded when context times out
+		require.Error(t, err)
+		assert.Equal(t, context.DeadlineExceeded, err)
+
+		// Verify cleanup was called at least 3 times (initial + 2 periodic)
+		assert.GreaterOrEqual(t, callCount, 3, "should have run cleanup at least 3 times")
+	})
+
+	t.Run("stops when context is cancelled", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		controller := NewJobCleanupController(store, historyWriter, 30*time.Second)
+		controller.cleanupInterval = 1 * time.Second
+
+		ctx, cancel := context.WithCancel(context.Background())
+
+		// Expect initial cleanup
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{}, nil).Once()
+
+		// Cancel after initial cleanup
+		go func() {
+			time.Sleep(50 * time.Millisecond)
+			cancel()
+		}()
+
+		err := controller.Run(ctx)
+
+		assert.Error(t, err)
+		assert.Equal(t, context.Canceled, err)
+		store.AssertExpectations(t)
+	})
+
+	t.Run("continues running after cleanup error", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		controller := NewJobCleanupController(store, historyWriter, 10*time.Second)
+		controller.cleanupInterval = 50 * time.Millisecond
+
+		ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond)
+		defer cancel()
+
+		// Track successful calls after initial failure
+		successCount := 0
+		// First cleanup fails
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).
+			Return(nil, errors.New("first failure")).Once()
+		// Subsequent cleanups succeed
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).
+			Run(func(args mock.Arguments) {
+				successCount++
+			}).
+			Return([]*provisioning.Job{}, nil).
+			Maybe()
+
+		err := controller.Run(ctx)
+
+		// Should still run and return context error
+		require.Error(t, err)
+		assert.Equal(t, context.DeadlineExceeded, err)
+		// Verify it was called successfully at least once after first failure
+		assert.GreaterOrEqual(t, successCount, 1, "should have retried after first failure")
+	})
+
+	t.Run("logs error when periodic cleanup fails", func(t *testing.T) {
+		store := &MockStore{}
+
+		historyWriter := &MockHistoryWriter{}
+
+		controller := NewJobCleanupController(store, historyWriter, 10*time.Second)
+		controller.cleanupInterval = 50 * time.Millisecond
+
+		ctx, cancel := context.WithTimeout(context.Background(), 125*time.Millisecond)
+		defer cancel()
+
+		// Initial cleanup succeeds
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).
+			Return([]*provisioning.Job{}, nil).Once()
+
+		// First periodic cleanup fails (this tests the error logging in ticker case)
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).
+			Return(nil, errors.New("periodic failure")).Once()
+
+		// Subsequent cleanups succeed
+		store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).
+			Return([]*provisioning.Job{}, nil).
+			Maybe()
+
+		err := controller.Run(ctx)
+
+		// Should still run and return context error, not the cleanup error
+		require.Error(t, err)
+		assert.Equal(t, context.DeadlineExceeded, err)
+		store.AssertExpectations(t)
+	})
+}

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

@@ -0,0 +1,158 @@
+// Code generated by mockery v2.53.4. DO NOT EDIT.
+
+package jobs
+
+import (
+	context "context"
+
+	v0alpha1 "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+	mock "github.com/stretchr/testify/mock"
+)
+
+// MockHistoryReader is an autogenerated mock type for the HistoryReader type
+type MockHistoryReader struct {
+	mock.Mock
+}
+
+type MockHistoryReader_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *MockHistoryReader) EXPECT() *MockHistoryReader_Expecter {
+	return &MockHistoryReader_Expecter{mock: &_m.Mock}
+}
+
+// GetJob provides a mock function with given fields: ctx, namespace, repo, uid
+func (_m *MockHistoryReader) GetJob(ctx context.Context, namespace string, repo string, uid string) (*v0alpha1.Job, error) {
+	ret := _m.Called(ctx, namespace, repo, uid)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetJob")
+	}
+
+	var r0 *v0alpha1.Job
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, string) (*v0alpha1.Job, error)); ok {
+		return rf(ctx, namespace, repo, uid)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, string) *v0alpha1.Job); ok {
+		r0 = rf(ctx, namespace, repo, uid)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*v0alpha1.Job)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, string, string, string) error); ok {
+		r1 = rf(ctx, namespace, repo, uid)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// MockHistoryReader_GetJob_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetJob'
+type MockHistoryReader_GetJob_Call struct {
+	*mock.Call
+}
+
+// GetJob is a helper method to define mock.On call
+//   - ctx context.Context
+//   - namespace string
+//   - repo string
+//   - uid string
+func (_e *MockHistoryReader_Expecter) GetJob(ctx interface{}, namespace interface{}, repo interface{}, uid interface{}) *MockHistoryReader_GetJob_Call {
+	return &MockHistoryReader_GetJob_Call{Call: _e.mock.On("GetJob", ctx, namespace, repo, uid)}
+}
+
+func (_c *MockHistoryReader_GetJob_Call) Run(run func(ctx context.Context, namespace string, repo string, uid string)) *MockHistoryReader_GetJob_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(string), args[2].(string), args[3].(string))
+	})
+	return _c
+}
+
+func (_c *MockHistoryReader_GetJob_Call) Return(_a0 *v0alpha1.Job, _a1 error) *MockHistoryReader_GetJob_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *MockHistoryReader_GetJob_Call) RunAndReturn(run func(context.Context, string, string, string) (*v0alpha1.Job, error)) *MockHistoryReader_GetJob_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// RecentJobs provides a mock function with given fields: ctx, namespace, repo
+func (_m *MockHistoryReader) RecentJobs(ctx context.Context, namespace string, repo string) (*v0alpha1.JobList, error) {
+	ret := _m.Called(ctx, namespace, repo)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RecentJobs")
+	}
+
+	var r0 *v0alpha1.JobList
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, string, string) (*v0alpha1.JobList, error)); ok {
+		return rf(ctx, namespace, repo)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, string, string) *v0alpha1.JobList); ok {
+		r0 = rf(ctx, namespace, repo)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*v0alpha1.JobList)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
+		r1 = rf(ctx, namespace, repo)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// MockHistoryReader_RecentJobs_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RecentJobs'
+type MockHistoryReader_RecentJobs_Call struct {
+	*mock.Call
+}
+
+// RecentJobs is a helper method to define mock.On call
+//   - ctx context.Context
+//   - namespace string
+//   - repo string
+func (_e *MockHistoryReader_Expecter) RecentJobs(ctx interface{}, namespace interface{}, repo interface{}) *MockHistoryReader_RecentJobs_Call {
+	return &MockHistoryReader_RecentJobs_Call{Call: _e.mock.On("RecentJobs", ctx, namespace, repo)}
+}
+
+func (_c *MockHistoryReader_RecentJobs_Call) Run(run func(ctx context.Context, namespace string, repo string)) *MockHistoryReader_RecentJobs_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(string), args[2].(string))
+	})
+	return _c
+}
+
+func (_c *MockHistoryReader_RecentJobs_Call) Return(_a0 *v0alpha1.JobList, _a1 error) *MockHistoryReader_RecentJobs_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *MockHistoryReader_RecentJobs_Call) RunAndReturn(run func(context.Context, string, string) (*v0alpha1.JobList, error)) *MockHistoryReader_RecentJobs_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// NewMockHistoryReader creates a new instance of MockHistoryReader. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewMockHistoryReader(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *MockHistoryReader {
+	mock := &MockHistoryReader{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

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

@@ -0,0 +1,84 @@
+// Code generated by mockery v2.53.4. DO NOT EDIT.
+
+package jobs
+
+import (
+	context "context"
+
+	v0alpha1 "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
+	mock "github.com/stretchr/testify/mock"
+)
+
+// MockHistoryWriter is an autogenerated mock type for the HistoryWriter type
+type MockHistoryWriter struct {
+	mock.Mock
+}
+
+type MockHistoryWriter_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *MockHistoryWriter) EXPECT() *MockHistoryWriter_Expecter {
+	return &MockHistoryWriter_Expecter{mock: &_m.Mock}
+}
+
+// WriteJob provides a mock function with given fields: ctx, job
+func (_m *MockHistoryWriter) WriteJob(ctx context.Context, job *v0alpha1.Job) error {
+	ret := _m.Called(ctx, job)
+
+	if len(ret) == 0 {
+		panic("no return value specified for WriteJob")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, *v0alpha1.Job) error); ok {
+		r0 = rf(ctx, job)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// MockHistoryWriter_WriteJob_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'WriteJob'
+type MockHistoryWriter_WriteJob_Call struct {
+	*mock.Call
+}
+
+// WriteJob is a helper method to define mock.On call
+//   - ctx context.Context
+//   - job *v0alpha1.Job
+func (_e *MockHistoryWriter_Expecter) WriteJob(ctx interface{}, job interface{}) *MockHistoryWriter_WriteJob_Call {
+	return &MockHistoryWriter_WriteJob_Call{Call: _e.mock.On("WriteJob", ctx, job)}
+}
+
+func (_c *MockHistoryWriter_WriteJob_Call) Run(run func(ctx context.Context, job *v0alpha1.Job)) *MockHistoryWriter_WriteJob_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(*v0alpha1.Job))
+	})
+	return _c
+}
+
+func (_c *MockHistoryWriter_WriteJob_Call) Return(_a0 error) *MockHistoryWriter_WriteJob_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *MockHistoryWriter_WriteJob_Call) RunAndReturn(run func(context.Context, *v0alpha1.Job) error) *MockHistoryWriter_WriteJob_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// NewMockHistoryWriter creates a new instance of MockHistoryWriter. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewMockHistoryWriter(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *MockHistoryWriter {
+	mock := &MockHistoryWriter{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

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

@@ -134,8 +134,7 @@ func (_c *MockLokiClient_RangeQuery_Call) RunAndReturn(run func(context.Context,
 func NewMockLokiClient(t interface {
 	mock.TestingT
 	Cleanup(func())
-},
-) *MockLokiClient {
+}) *MockLokiClient {
 	mock := &MockLokiClient{}
 	mock.Mock.Test(t)
 

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

@@ -306,11 +306,11 @@ func (s *persistentStore) Complete(ctx context.Context, job *provisioning.Job) e
 		return apifmt.Errorf("failed to get provisioning identity for '%s': %w", job.GetNamespace(), err)
 	}
 
-	// We need to delete the job from the job store and create it in the historic job store.
-	// We are fine with the job being lost if the historic job store fails to create it.
+	// Delete the job from the active job store.
+	// Callers are responsible for writing the job to history after calling this.
 	//
 	// We will assume that the caller is the claimant. If this is not true, an error is returned.
-	// This is a best-effort operation; if the job is not in the claimed state, we will still attempt to move it to the historic job store.
+	// This is a best-effort operation; if the job is not in the claimed state, we will still attempt to delete it.
 	err = s.client.Jobs(job.GetNamespace()).Delete(ctx, job.GetName(), metav1.DeleteOptions{})
 	if err != nil {
 		span.RecordError(err)
@@ -329,6 +329,56 @@ func (s *persistentStore) Complete(ctx context.Context, job *provisioning.Job) e
 	return nil
 }
 
+// ListExpiredJobs lists jobs with expired leases (claim timestamp older than the given time).
+// Returns jobs in batches up to the specified limit.
+func (s *persistentStore) ListExpiredJobs(ctx context.Context, expiredBefore time.Time, limit int) ([]*provisioning.Job, error) {
+	ctx, span := tracing.Start(ctx, "provisioning.jobs.list_expired_jobs")
+	defer span.End()
+
+	logger := logging.FromContext(ctx).With("operation", "list_expired_jobs")
+
+	// Set up provisioning identity to access jobs across all namespaces
+	ctx, _, err := identity.WithProvisioningIdentity(ctx, "*")
+	if err != nil {
+		span.RecordError(err)
+		return nil, apifmt.Errorf("failed to grant provisioning identity for listing expired jobs: %w", err)
+	}
+
+	// Find jobs with expired leases (older than expiredBefore)
+	expiry := expiredBefore.UnixMilli()
+	logger.Debug("searching for expired jobs", "expiry_threshold", expiredBefore.Format(time.RFC3339))
+
+	requirement, err := labels.NewRequirement(LabelJobClaim, selection.LessThan, []string{strconv.FormatInt(expiry, 10)})
+	if err != nil {
+		span.RecordError(err)
+		return nil, apifmt.Errorf("could not create requirement: %w", err)
+	}
+
+	span.SetAttributes(
+		attribute.String("expiry_threshold", expiredBefore.Format(time.RFC3339)),
+		attribute.Int("limit", limit),
+	)
+
+	jobList, err := s.client.Jobs("").List(ctx, metav1.ListOptions{
+		LabelSelector: labels.NewSelector().Add(*requirement).String(),
+		Limit:         int64(limit),
+	})
+	if err != nil {
+		span.RecordError(err)
+		return nil, apifmt.Errorf("failed to list jobs with expired leases: %w", err)
+	}
+
+	result := make([]*provisioning.Job, len(jobList.Items))
+	for i := range jobList.Items {
+		result[i] = &jobList.Items[i]
+	}
+
+	span.SetAttributes(attribute.Int("jobs_found", len(result)))
+	logger.Debug("found expired jobs", "count", len(result))
+
+	return result, nil
+}
+
 // RenewLease renews the lease for a claimed job, extending its expiry time.
 // Returns an error if the lease cannot be renewed (e.g., job was completed or lease expired).
 func (s *persistentStore) RenewLease(ctx context.Context, job *provisioning.Job) error {
@@ -405,164 +455,6 @@ func (s *persistentStore) RenewLease(ctx context.Context, job *provisioning.Job)
 	return nil
 }
 
-// Cleanup finds jobs with expired leases and marks them as failed.
-// This replaces the old cleanup mechanism and should be called more frequently.
-func (s *persistentStore) Cleanup(ctx context.Context) error {
-	ctx, span := tracing.Start(ctx, "provisioning.jobs.cleanup")
-	defer span.End()
-
-	startTime := s.clock()
-	logger := logging.FromContext(ctx).With("operation", "cleanup")
-
-	// List expired jobs
-	jobs, err := s.listExpiredJobs(ctx)
-	if err != nil {
-		span.RecordError(err)
-		return err
-	}
-
-	// If no jobs found, cleanup is complete
-	if len(jobs) == 0 {
-		duration := s.clock().Sub(startTime)
-		logger.Info("cleanup complete - no expired jobs found", "duration", duration)
-		span.SetAttributes(
-			attribute.Int("count", 0),
-			attribute.Int64("duration_ms", duration.Milliseconds()),
-		)
-		return nil
-	}
-
-	logger.Info("found expired jobs", "count", len(jobs))
-
-	// Clean up each expired job
-	for _, job := range jobs {
-		if err := s.cleanUpExpiredJob(ctx, job); err != nil {
-			span.RecordError(err)
-			return err
-		}
-	}
-
-	duration := s.clock().Sub(startTime)
-	logger.Info("cleanup complete",
-		"duration", duration,
-		"count", len(jobs),
-	)
-
-	span.SetAttributes(
-		attribute.Int("count", len(jobs)),
-		attribute.Int64("duration_ms", duration.Milliseconds()),
-	)
-
-	return nil
-}
-
-// listExpiredJobs returns jobs with expired leases.
-func (s *persistentStore) listExpiredJobs(ctx context.Context) ([]provisioning.Job, error) {
-	logger := logging.FromContext(ctx)
-
-	// Set up provisioning identity to access jobs across all namespaces
-	ctx, _, err := identity.WithProvisioningIdentity(ctx, "*") // "*" grants access to all namespaces
-	if err != nil {
-		return nil, apifmt.Errorf("failed to grant provisioning identity for cleanup: %w", err)
-	}
-
-	// Find jobs with expired leases (older than expiry time)
-	expiry := s.clock().Add(-s.expiry).UnixMilli()
-	expiryTime := time.UnixMilli(expiry)
-	logger.Debug("search for expired jobs", "expiry_threshold", expiryTime.Format(time.RFC3339))
-
-	requirement, err := labels.NewRequirement(LabelJobClaim, selection.LessThan, []string{strconv.FormatInt(expiry, 10)})
-	if err != nil {
-		return nil, apifmt.Errorf("could not create requirement: %w", err)
-	}
-
-	listCtx, listSpan := tracing.Start(ctx, "provisioning.jobs.cleanup.list_expired_jobs")
-	defer listSpan.End()
-
-	listSpan.SetAttributes(
-		attribute.String("expiry_threshold", expiryTime.Format(time.RFC3339)),
-		attribute.Int64("expiry_duration_seconds", int64(s.expiry.Seconds())),
-	)
-
-	timeoutCtx, cancel := context.WithTimeout(listCtx, 5*time.Second)
-	defer cancel()
-
-	jobList, err := s.client.Jobs("").List(timeoutCtx, metav1.ListOptions{
-		LabelSelector: labels.NewSelector().Add(*requirement).String(),
-		Limit:         100, // Process in batches
-	})
-	if err != nil {
-		listSpan.RecordError(err)
-		return nil, apifmt.Errorf("failed to list jobs with expired leases: %w", err)
-	}
-
-	listSpan.SetAttributes(attribute.Int("jobs_found", len(jobList.Items)))
-	return jobList.Items, nil
-}
-
-// cleanUpExpiredJob marks a single expired job as failed and archives it.
-func (s *persistentStore) cleanUpExpiredJob(ctx context.Context, job provisioning.Job) error {
-	// Calculate how long the job has been expired
-	var expiredFor time.Duration
-	var claimTimestamp time.Time
-	if claimTime, exists := job.Labels[LabelJobClaim]; exists {
-		claimMillis, parseErr := strconv.ParseInt(claimTime, 10, 64)
-		if parseErr == nil {
-			claimTimestamp = time.UnixMilli(claimMillis)
-			expiredFor = s.clock().Sub(claimTimestamp)
-		}
-	}
-
-	logger := logging.FromContext(ctx).With(
-		"job", job.GetName(),
-		"namespace", job.GetNamespace(),
-		"repository", job.Spec.Repository,
-		"action", job.Spec.Action,
-		"expired_for", expiredFor,
-	)
-
-	if !claimTimestamp.IsZero() {
-		logger = logger.With("claim_time", claimTimestamp.Format(time.RFC3339))
-	}
-
-	jobCtx, jobSpan := tracing.Start(ctx, "provisioning.jobs.cleanup.complete_expired_job")
-	defer jobSpan.End()
-
-	jobSpan.SetAttributes(
-		attribute.String("job.name", job.GetName()),
-		attribute.String("job.namespace", job.GetNamespace()),
-		attribute.String("job.repository", job.Spec.Repository),
-		attribute.String("job.action", string(job.Spec.Action)),
-		attribute.String("job.expired_for", expiredFor.String()),
-	)
-
-	// Mark job as failed due to lease expiry and archive it
-	jobCopy := job.DeepCopy()
-	jobCopy.Status.State = provisioning.JobStateError
-	jobCopy.Status.Message = "Job failed due to lease expiry - worker may have crashed or lost connection"
-
-	// Set namespace context for the completion
-	jobCtx, _, err := identity.WithProvisioningIdentity(jobCtx, job.GetNamespace())
-	if err != nil {
-		jobSpan.RecordError(err)
-		return apifmt.Errorf("failed to get provisioning identity for '%s': %w", job.GetNamespace(), err)
-	}
-
-	// Use Complete to properly archive the failed job
-	if err := s.Complete(jobCtx, jobCopy); err != nil {
-		if apierrors.IsNotFound(err) {
-			// Job was already completed/deleted by another process - this is expected
-			logger.Warn("job already completed or deleted by another process")
-			return nil
-		}
-		jobSpan.RecordError(err)
-		return apifmt.Errorf("failed to complete expired job '%s' in '%s': %w", job.GetName(), job.GetNamespace(), err)
-	}
-
-	logger.Info("clean up expired job complete")
-	return nil
-}
-
 func (s *persistentStore) Insert(ctx context.Context, namespace string, spec provisioning.JobSpec) (*provisioning.Job, error) {
 	ctx, span := tracing.Start(ctx, "provisioning.jobs.insert")
 	defer span.End()

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

@@ -1,12 +1,14 @@
-// Code generated by mockery v2.52.4. DO NOT EDIT.
+// Code generated by mockery v2.53.4. DO NOT EDIT.
 
 package jobs
 
 import (
 	context "context"
+	time "time"
+
+	mock "github.com/stretchr/testify/mock"
 
 	v0alpha1 "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1"
-	mock "github.com/stretchr/testify/mock"
 )
 
 // MockStore is an autogenerated mock type for the Store type
@@ -89,52 +91,6 @@ func (_c *MockStore_Claim_Call) RunAndReturn(run func(context.Context) (*v0alpha
 	return _c
 }
 
-// Cleanup provides a mock function with given fields: ctx
-func (_m *MockStore) Cleanup(ctx context.Context) error {
-	ret := _m.Called(ctx)
-
-	if len(ret) == 0 {
-		panic("no return value specified for Cleanup")
-	}
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context) error); ok {
-		r0 = rf(ctx)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// MockStore_Cleanup_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Cleanup'
-type MockStore_Cleanup_Call struct {
-	*mock.Call
-}
-
-// Cleanup is a helper method to define mock.On call
-//   - ctx context.Context
-func (_e *MockStore_Expecter) Cleanup(ctx interface{}) *MockStore_Cleanup_Call {
-	return &MockStore_Cleanup_Call{Call: _e.mock.On("Cleanup", ctx)}
-}
-
-func (_c *MockStore_Cleanup_Call) Run(run func(ctx context.Context)) *MockStore_Cleanup_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context))
-	})
-	return _c
-}
-
-func (_c *MockStore_Cleanup_Call) Return(_a0 error) *MockStore_Cleanup_Call {
-	_c.Call.Return(_a0)
-	return _c
-}
-
-func (_c *MockStore_Cleanup_Call) RunAndReturn(run func(context.Context) error) *MockStore_Cleanup_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
 // Complete provides a mock function with given fields: ctx, job
 func (_m *MockStore) Complete(ctx context.Context, job *v0alpha1.Job) error {
 	ret := _m.Called(ctx, job)
@@ -182,9 +138,9 @@ func (_c *MockStore_Complete_Call) RunAndReturn(run func(context.Context, *v0alp
 	return _c
 }
 
-// Get provides a mock function with given fields: ctx, name
-func (_m *MockStore) Get(ctx context.Context, name string) (*v0alpha1.Job, error) {
-	ret := _m.Called(ctx, name)
+// Get provides a mock function with given fields: ctx, namespace, name
+func (_m *MockStore) Get(ctx context.Context, namespace string, name string) (*v0alpha1.Job, error) {
+	ret := _m.Called(ctx, namespace, name)
 
 	if len(ret) == 0 {
 		panic("no return value specified for Get")
@@ -192,19 +148,19 @@ func (_m *MockStore) Get(ctx context.Context, name string) (*v0alpha1.Job, error
 
 	var r0 *v0alpha1.Job
 	var r1 error
-	if rf, ok := ret.Get(0).(func(context.Context, string) (*v0alpha1.Job, error)); ok {
-		return rf(ctx, name)
+	if rf, ok := ret.Get(0).(func(context.Context, string, string) (*v0alpha1.Job, error)); ok {
+		return rf(ctx, namespace, name)
 	}
-	if rf, ok := ret.Get(0).(func(context.Context, string) *v0alpha1.Job); ok {
-		r0 = rf(ctx, name)
+	if rf, ok := ret.Get(0).(func(context.Context, string, string) *v0alpha1.Job); ok {
+		r0 = rf(ctx, namespace, name)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(*v0alpha1.Job)
 		}
 	}
 
-	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
-		r1 = rf(ctx, name)
+	if rf, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
+		r1 = rf(ctx, namespace, name)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -219,14 +175,15 @@ type MockStore_Get_Call struct {
 
 // Get is a helper method to define mock.On call
 //   - ctx context.Context
+//   - namespace string
 //   - name string
-func (_e *MockStore_Expecter) Get(ctx interface{}, name interface{}) *MockStore_Get_Call {
-	return &MockStore_Get_Call{Call: _e.mock.On("Get", ctx, name)}
+func (_e *MockStore_Expecter) Get(ctx interface{}, namespace interface{}, name interface{}) *MockStore_Get_Call {
+	return &MockStore_Get_Call{Call: _e.mock.On("Get", ctx, namespace, name)}
 }
 
-func (_c *MockStore_Get_Call) Run(run func(ctx context.Context, name string)) *MockStore_Get_Call {
+func (_c *MockStore_Get_Call) Run(run func(ctx context.Context, namespace string, name string)) *MockStore_Get_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(string))
+		run(args[0].(context.Context), args[1].(string), args[2].(string))
 	})
 	return _c
 }
@@ -236,7 +193,67 @@ func (_c *MockStore_Get_Call) Return(_a0 *v0alpha1.Job, _a1 error) *MockStore_Ge
 	return _c
 }
 
-func (_c *MockStore_Get_Call) RunAndReturn(run func(context.Context, string) (*v0alpha1.Job, error)) *MockStore_Get_Call {
+func (_c *MockStore_Get_Call) RunAndReturn(run func(context.Context, string, string) (*v0alpha1.Job, error)) *MockStore_Get_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// ListExpiredJobs provides a mock function with given fields: ctx, expiredBefore, limit
+func (_m *MockStore) ListExpiredJobs(ctx context.Context, expiredBefore time.Time, limit int) ([]*v0alpha1.Job, error) {
+	ret := _m.Called(ctx, expiredBefore, limit)
+
+	if len(ret) == 0 {
+		panic("no return value specified for ListExpiredJobs")
+	}
+
+	var r0 []*v0alpha1.Job
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, time.Time, int) ([]*v0alpha1.Job, error)); ok {
+		return rf(ctx, expiredBefore, limit)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, time.Time, int) []*v0alpha1.Job); ok {
+		r0 = rf(ctx, expiredBefore, limit)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*v0alpha1.Job)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, time.Time, int) error); ok {
+		r1 = rf(ctx, expiredBefore, limit)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// MockStore_ListExpiredJobs_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListExpiredJobs'
+type MockStore_ListExpiredJobs_Call struct {
+	*mock.Call
+}
+
+// ListExpiredJobs is a helper method to define mock.On call
+//   - ctx context.Context
+//   - expiredBefore time.Time
+//   - limit int
+func (_e *MockStore_Expecter) ListExpiredJobs(ctx interface{}, expiredBefore interface{}, limit interface{}) *MockStore_ListExpiredJobs_Call {
+	return &MockStore_ListExpiredJobs_Call{Call: _e.mock.On("ListExpiredJobs", ctx, expiredBefore, limit)}
+}
+
+func (_c *MockStore_ListExpiredJobs_Call) Run(run func(ctx context.Context, expiredBefore time.Time, limit int)) *MockStore_ListExpiredJobs_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(time.Time), args[2].(int))
+	})
+	return _c
+}
+
+func (_c *MockStore_ListExpiredJobs_Call) Return(_a0 []*v0alpha1.Job, _a1 error) *MockStore_ListExpiredJobs_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *MockStore_ListExpiredJobs_Call) RunAndReturn(run func(context.Context, time.Time, int) ([]*v0alpha1.Job, error)) *MockStore_ListExpiredJobs_Call {
 	_c.Call.Return(run)
 	return _c
 }

pkg/registry/apis/provisioning/register.go

@@ -775,13 +775,21 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
 			}
 
 			repoGetter := resources.NewRepositoryGetter(b.repoFactory, b.client)
+
+			// Create job cleanup controller
+			jobExpiry := 30 * time.Second
+			jobCleanupController := jobs.NewJobCleanupController(
+				b.jobs,
+				jobHistoryWriter,
+				jobExpiry,
+			)
+
 			// This is basically our own JobQueue system
 			driver, err := jobs.NewConcurrentJobDriver(
 				3,              // 3 drivers for now
 				20*time.Minute, // Max time for each job
-				time.Minute,    // Cleanup jobs
 				30*time.Second, // Periodically look for new jobs
-				30*time.Second, // Lease renewal interval
+				jobExpiry,      // Lease renewal interval
 				b.jobs, repoGetter, jobHistoryWriter,
 				jobController.InsertNotifications(),
 				b.registry,
@@ -797,6 +805,12 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
 				}
 			}()
 
+			go func() {
+				if err := jobCleanupController.Run(postStartHookCtx.Context); err != nil {
+					logging.FromContext(postStartHookCtx.Context).Error("job cleanup controller failed", "error", err)
+				}
+			}()
+
 			repoController, err := controller.NewRepositoryController(
 				b.GetClient(),
 				repoInformer,
@@ -821,7 +835,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH
 			if b.jobHistoryLoki == nil {
 				// Create HistoryJobController for cleanup of old job history entries
 				// Separate informer factory for HistoryJob cleanup with resync interval
-				historyJobExpiration := 30 * time.Second
+				historyJobExpiration := 10 * time.Minute
 				historyJobInformerFactory := informers.NewSharedInformerFactory(c, historyJobExpiration)
 				historyJobInformer := historyJobInformerFactory.Provisioning().V0alpha1().HistoricJobs()
 				go historyJobInformer.Informer().Run(postStartHookCtx.Done())