scheduler: bound policy backup starts

internal/config/config.go

@@ -50,6 +50,7 @@ type Config struct {
 	AuthBearerTokens       []string
 	MetricsRefreshInterval time.Duration
 	PolicyEvalInterval     time.Duration
+	PolicyBackupsPerCycle  int
 	PolicySecretName       string
 	UsageSecretName        string
 	BackupMaxAge           time.Duration
@@ -133,6 +134,9 @@ func Load() (*Config, error) {
 	if seconds, ok := getenvInt("SOTERIA_POLICY_EVAL_SECONDS"); ok {
 		cfg.PolicyEvalInterval = time.Duration(seconds) * time.Second
 	}
+	if limit, ok := getenvInt("SOTERIA_POLICY_BACKUPS_PER_CYCLE"); ok {
+		cfg.PolicyBackupsPerCycle = limit
+	}
 	if hours, ok := getenvFloat("SOTERIA_BACKUP_MAX_AGE_HOURS"); ok {
 		cfg.BackupMaxAge = time.Duration(hours * float64(time.Hour))
 	}
@@ -187,6 +191,9 @@ func Load() (*Config, error) {
 	if cfg.PolicyEvalInterval <= 0 {
 		return nil, errors.New("SOTERIA_POLICY_EVAL_SECONDS must be greater than zero")
 	}
+	if cfg.PolicyBackupsPerCycle < 0 {
+		return nil, errors.New("SOTERIA_POLICY_BACKUPS_PER_CYCLE must be greater than or equal to zero")
+	}
 	if strings.TrimSpace(cfg.PolicySecretName) == "" {
 		return nil, errors.New("SOTERIA_POLICY_SECRET_NAME must not be empty")
 	}

internal/config/config_test.go

@@ -34,6 +34,7 @@ func clearConfigEnv(t *testing.T) {
 		"SOTERIA_AUTH_BEARER_TOKENS",
 		"SOTERIA_METRICS_REFRESH_SECONDS",
 		"SOTERIA_POLICY_EVAL_SECONDS",
+		"SOTERIA_POLICY_BACKUPS_PER_CYCLE",
 		"SOTERIA_POLICY_SECRET_NAME",
 		"SOTERIA_USAGE_SECRET_NAME",
 		"SOTERIA_BACKUP_MAX_AGE_HOURS",
@@ -117,6 +118,7 @@ func TestLoadSupportsResticAndB2Overrides(t *testing.T) {
 	withEnv(t, "SOTERIA_AUTH_BEARER_TOKENS", "alpha,beta")
 	withEnv(t, "SOTERIA_METRICS_REFRESH_SECONDS", "60")
 	withEnv(t, "SOTERIA_POLICY_EVAL_SECONDS", "120")
+	withEnv(t, "SOTERIA_POLICY_BACKUPS_PER_CYCLE", "3")
 	withEnv(t, "SOTERIA_POLICY_SECRET_NAME", "policy-doc")
 	withEnv(t, "SOTERIA_USAGE_SECRET_NAME", "usage-doc")
 	withEnv(t, "SOTERIA_BACKUP_MAX_AGE_HOURS", "12.5")
@@ -162,6 +164,9 @@ func TestLoadSupportsResticAndB2Overrides(t *testing.T) {
 	if cfg.MetricsRefreshInterval != 60*time.Second || cfg.PolicyEvalInterval != 120*time.Second {
 		t.Fatalf("unexpected intervals: %#v", cfg)
 	}
+	if cfg.PolicyBackupsPerCycle != 3 {
+		t.Fatalf("unexpected policy backup limit: %#v", cfg.PolicyBackupsPerCycle)
+	}
 	if cfg.BackupMaxAge != time.Duration(12.5*float64(time.Hour)) {
 		t.Fatalf("unexpected backup max age: %#v", cfg.BackupMaxAge)
 	}
@@ -285,6 +290,13 @@ func TestLoadRejectsInvalidConfigurations(t *testing.T) {
 			},
 			substr: "SOTERIA_POLICY_EVAL_SECONDS must be greater than zero",
 		},
+		{
+			name: "invalid policy backup limit",
+			env: map[string]string{
+				"SOTERIA_POLICY_BACKUPS_PER_CYCLE": "-1",
+			},
+			substr: "SOTERIA_POLICY_BACKUPS_PER_CYCLE must be greater than or equal to zero",
+		},
 	}
 
 	for _, tc := range testCases {

internal/server/policy_cycle_test.go

@@ -154,3 +154,42 @@ func TestRunPolicyCycleCoversEffectivePoliciesAndResultTracking(t *testing.T) {
 		t.Fatalf("expected one failed executed backup request, got %f", got)
 	}
 }
+
+func TestRunPolicyCycleHonorsBackupLimit(t *testing.T) {
+	client := &policyCycleTestKubeClient{
+		inventoryTestKubeClient: &inventoryTestKubeClient{
+			fakeKubeClient: &fakeKubeClient{
+				pvcs: []k8s.PVCSummary{
+					{Namespace: "apps", Name: "alpha", VolumeName: "vol-alpha", Phase: "Bound"},
+					{Namespace: "apps", Name: "beta", VolumeName: "vol-beta", Phase: "Bound"},
+				},
+			},
+		},
+	}
+	srv := &Server{
+		cfg: &config.Config{
+			BackupDriver:          "restic",
+			BackupMaxAge:          24 * time.Hour,
+			ResticRepository:      "s3:https://repo/root",
+			PolicyBackupsPerCycle: 1,
+		},
+		client:   client,
+		longhorn: &fakeLonghornClient{},
+		metrics:  newTelemetry(),
+		policies: map[string]api.BackupPolicy{
+			"apps__all": {ID: "apps__all", Namespace: "apps", IntervalHours: 1, Enabled: true, Dedupe: true},
+		},
+	}
+
+	srv.runPolicyCycle(context.Background())
+
+	if len(client.backupRequests) != 1 {
+		t.Fatalf("expected one policy backup due to cycle limit, got %#v", client.backupRequests)
+	}
+	if client.backupRequests[0].PVC != "alpha" {
+		t.Fatalf("expected deterministic first backup for alpha, got %#v", client.backupRequests[0])
+	}
+	if got := metricCount(srv.metrics.policyBackups, map[string]string{"result": "cycle_limit"}); got != 1 {
+		t.Fatalf("expected cycle_limit metric, got %f", got)
+	}
+}

internal/server/policy_runtime.go

@@ -72,7 +72,15 @@ func (s *Server) runPolicyCycle(ctx context.Context) {
 		}
 	}
 
-	for key, effective := range effectivePolicies {
+	keys := make([]string, 0, len(effectivePolicies))
+	for key := range effectivePolicies {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+
+	started := 0
+	for _, key := range keys {
+		effective := effectivePolicies[key]
 		pvc, ok := pvcMap[key]
 		if !ok {
 			continue
@@ -95,6 +103,10 @@ func (s *Server) runPolicyCycle(ctx context.Context) {
 			s.metrics.RecordPolicyBackup("not_due")
 			continue
 		}
+		if s.cfg.PolicyBackupsPerCycle > 0 && started >= s.cfg.PolicyBackupsPerCycle {
+			s.metrics.RecordPolicyBackup("cycle_limit")
+			break
+		}
 
 		_, result, err := s.executeBackup(runCtx, api.BackupRequest{
 			Namespace: pvc.Namespace,
@@ -103,6 +115,7 @@ func (s *Server) runPolicyCycle(ctx context.Context) {
 			Dedupe:    boolPtr(effective.Dedupe),
 			KeepLast:  intPtr(effective.KeepLast),
 		}, "policy-scheduler")
+		started++
 		s.metrics.RecordBackupRequest(s.cfg.BackupDriver, result)
 		if err != nil {
 			s.metrics.RecordPolicyBackup(result)