Function testPersistenceAndRecovery

modules/backendscheduler/integration_test.go:160–203

(ctx context.Context, t *testing.T, originalScheduler *BackendScheduler, cfg Config, store storage.Store, limits overrides.Interface, rr backend.RawReader, ww backend.RawWriter)

}

func testPersistenceAndRecovery(ctx context.Context, t *testing.T, originalScheduler *BackendScheduler, cfg Config, store storage.Store, limits overrides.Interface, rr backend.RawReader, ww backend.RawWriter) {
	// Get initial jobs
	initialJobs := originalScheduler.ListJobs()
	require.NotEmpty(t, initialJobs, "Should have jobs to test persistence")

	// Force a flush to disk
	err := originalScheduler.work.FlushToLocal(ctx, originalScheduler.cfg.LocalWorkPath, nil) // Flush all
	require.NoError(t, err)

	// Create a new scheduler instance (simulating restart)
	newScheduler, err := New(cfg, store, limits, rr, ww)
	require.NoError(t, err)

	err = newScheduler.starting(ctx)
	require.NoError(t, err)

	// Verify jobs were recovered
	recoveredJobs := newScheduler.ListJobs()
	require.Len(t, recoveredJobs, len(initialJobs))

	// Verify job details match
	initialJobMap := make(map[string]*work.Job)
	for _, job := range initialJobs {
		initialJobMap[job.ID] = job
	}

	for _, recoveredJob := range recoveredJobs {
		originalJob, exists := initialJobMap[recoveredJob.ID]
		require.True(t, exists, "Recovered job should exist in original jobs")
		require.Equal(t, originalJob.GetStatus(), recoveredJob.GetStatus())
		require.Equal(t, originalJob.Tenant(), recoveredJob.Tenant())
	}

	// Should have shard files
	foundShardFiles := 0
	for i := range work.ShardCount {
		shardPath := newScheduler.filepathForShard(uint8(i))
		if _, err := os.Stat(shardPath); err == nil {
			foundShardFiles++
		}
	}
	require.Greater(t, foundShardFiles, 0, "Should have at least some shard files")
	t.Logf("Found %d shard files", foundShardFiles)
}