admin: make the maintenance policy actually reach the task detectors

Loading the persisted task configs into the maintenance policy only
matters if something reads that policy, and nothing did.

MaintenanceIntegration pushes the policy into every registered detector
and scheduler through interface{ SetEnabled(bool) } and
interface{ SetMaxConcurrent(int) } type assertions. Every task registered
through base.RegisterTask is backed by base.GenericDetector and
base.GenericScheduler, and neither implemented either method, so all four
assertions failed silently for every task on every startup. The policy's
enabled flag reached nothing: ScanWithTaskDetectors gates on
detector.IsEnabled(), and the queue's policy lookups for max concurrent
and repeat interval are fallbacks that only fire when the scheduler
reports zero, which the generic scheduler never does.

Add the setters, delegating to the TaskConfig.SetEnabled the interface
already declares and to TaskDefinition.MaxConcurrent, which is what
GetMaxConcurrent returns.

Applying the policy required three more fixes, because with the
assertions working the policy could now do damage as well as good:

- IsTaskEnabled reports false for a task type the policy has no entry
  for, so applying it unconditionally would have disabled every task the
  policy does not list. Skip task types with no policy entry: no entry
  means no opinion, not disabled.

- ec_balance was exactly such a task. It is registered like the other
  three but had no entry in the policy builder and no accessor on
  ConfigPersistence at all, so its configuration could never be
  persisted. Add SaveEcBalanceTaskPolicy/LoadEcBalanceTaskPolicy, the
  task_ec_balance.pb file, the SaveTaskPolicy dispatcher case, and the
  policy entry.

- InitMaintenanceManager ran before loadTaskConfigurationsFromPersistence,
  which replaces each task's whole config object, so the policy was
  applied and then immediately thrown away. Swap the order. Both read the
  same files, so the policy is now the last writer and stays
  authoritative.

MaintenanceManager.UpdateConfig also updated the queue's and the
scanner's policy but not the integration's, so a policy changed at
runtime never reached the detectors. Add MaintenanceIntegration.SetPolicy
and call it.

While building the policy, stop hand-copying each task's fields and use
the task's own ToTaskPolicy(). The hand-written version was a second
definition of every task's policy and had already lost the erasure coding
preferred tags and replica placement and the balance IO rate limit. For
the same reason, the "nothing persisted yet" branches of
LoadVacuumTaskPolicy, LoadErasureCodingTaskPolicy and
LoadBalanceTaskPolicy now derive from each task's NewDefaultConfig()
instead of a third hand-written copy. Those copies had drifted, so with a
data directory but no config file on disk the effective defaults differed
from what the task and the admin UI schema both advertise:

  vacuum          scan interval  24h  -> 2h
  balance         scan interval   6h  -> 30m
  balance         imbalance      0.1  -> 0.2
  erasure coding  scan interval 168h  -> 1h
  erasure coding  fullness      0.90  -> 0.95
  erasure coding  min volume   1024MB -> 30MB

Finally, weed/admin/dash and weed/admin/maintenance each carried a copy
of the policy builder and they had already diverged. Export the
maintenance one as BuildPolicyFromTaskConfigs and have dash call it.

Refs #10874

weed/admin/dash/admin_server.go

@@ -278,12 +278,17 @@ func NewAdminServer(masters string, filerGroup string, templateFS http.FileSyste
 		glog.V(1).Infof("No data directory configured, maintenance system will run in memory-only mode (enabled: %v)", maintenanceConfig.Enabled)
 	}
 
+	// Load saved task configurations from persistence. This has to run before the maintenance
+	// manager is created: creating it applies the maintenance policy to the registered
+	// detectors and schedulers, while this call replaces each task's whole config object, so
+	// running it afterwards would discard what the policy just applied. Both read the same
+	// persisted task config files, so the policy ends up as the last writer and stays
+	// authoritative for the task types it covers.
+	server.loadTaskConfigurationsFromPersistence()
+
 	// Always initialize maintenance manager
 	server.InitMaintenanceManager(maintenanceConfig)
 
-	// Load saved task configurations from persistence
-	server.loadTaskConfigurationsFromPersistence()
-
 	// Start maintenance manager if enabled
 	if maintenanceConfig.Enabled {
 		go func() {

weed/admin/dash/config_persistence.go

@@ -14,6 +14,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance"
+	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/ec_balance"
 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum"
 	"google.golang.org/protobuf/encoding/protojson"
@@ -29,6 +30,7 @@ const (
 	VacuumTaskConfigFile      = "task_vacuum.pb"
 	ECTaskConfigFile          = "task_erasure_coding.pb"
 	BalanceTaskConfigFile     = "task_balance.pb"
+	EcBalanceTaskConfigFile   = "task_ec_balance.pb"
 	ReplicationTaskConfigFile = "task_replication.pb"
 
 	// JSON reference files
@@ -36,6 +38,7 @@ const (
 	VacuumTaskConfigJSONFile      = "task_vacuum.json"
 	ECTaskConfigJSONFile          = "task_erasure_coding.json"
 	BalanceTaskConfigJSONFile     = "task_balance.json"
+	EcBalanceTaskConfigJSONFile   = "task_ec_balance.json"
 	ReplicationTaskConfigJSONFile = "task_replication.json"
 
 	// Task persistence subdirectories and settings
@@ -53,6 +56,7 @@ type (
 	VacuumTaskConfig        = worker_pb.VacuumTaskConfig
 	ErasureCodingTaskConfig = worker_pb.ErasureCodingTaskConfig
 	BalanceTaskConfig       = worker_pb.BalanceTaskConfig
+	EcBalanceTaskConfig     = worker_pb.EcBalanceTaskConfig
 	ReplicationTaskConfig   = worker_pb.ReplicationTaskConfig
 )
 
@@ -268,6 +272,28 @@ func (cp *ConfigPersistence) RestoreConfig(filename, backupName string) error {
 	return nil
 }
 
+// Default task policies. These derive from each task's own NewDefaultConfig() so that a
+// task type has exactly one definition of its defaults. They used to be hand-written copies
+// here, and had drifted from the values the tasks themselves and the admin UI schema use:
+// vacuum scanned every 24h instead of 2h, balance every 6h instead of 30m with a 0.1 instead
+// of 0.2 imbalance threshold, and erasure coding every 168h instead of 1h with a 0.90 instead
+// of 0.95 fullness ratio and a 1024MB instead of 30MB minimum volume size.
+func defaultVacuumTaskPolicy() *worker_pb.TaskPolicy {
+	return vacuum.NewDefaultConfig().ToTaskPolicy()
+}
+
+func defaultErasureCodingTaskPolicy() *worker_pb.TaskPolicy {
+	return erasure_coding.NewDefaultConfig().ToTaskPolicy()
+}
+
+func defaultBalanceTaskPolicy() *worker_pb.TaskPolicy {
+	return balance.NewDefaultConfig().ToTaskPolicy()
+}
+
+func defaultEcBalanceTaskPolicy() *worker_pb.TaskPolicy {
+	return ec_balance.NewDefaultConfig().ToTaskPolicy()
+}
+
 // SaveVacuumTaskConfig saves vacuum task configuration to protobuf file
 func (cp *ConfigPersistence) SaveVacuumTaskConfig(config *VacuumTaskConfig) error {
 	return cp.saveTaskConfig(VacuumTaskConfigFile, config)
@@ -288,28 +314,14 @@ func (cp *ConfigPersistence) LoadVacuumTaskConfig() (*VacuumTaskConfig, error) {
 	}
 
 	// Return default config if no valid config found
-	return &VacuumTaskConfig{
-		GarbageThreshold:  0.3,
-		MinVolumeAgeHours: 24,
-	}, nil
+	return defaultVacuumTaskPolicy().GetVacuumConfig(), nil
 }
 
 // LoadVacuumTaskPolicy loads complete vacuum task policy from protobuf file
 func (cp *ConfigPersistence) LoadVacuumTaskPolicy() (*worker_pb.TaskPolicy, error) {
 	if cp.dataDir == "" {
 		// Return default policy if no data directory
-		return &worker_pb.TaskPolicy{
-			Enabled:               true,
-			MaxConcurrent:         2,
-			RepeatIntervalSeconds: 24 * 3600, // 24 hours in seconds
-			CheckIntervalSeconds:  6 * 3600,  // 6 hours in seconds
-			TaskConfig: &worker_pb.TaskPolicy_VacuumConfig{
-				VacuumConfig: &worker_pb.VacuumTaskConfig{
-					GarbageThreshold:  0.3,
-					MinVolumeAgeHours: 24,
-				},
-			},
-		}, nil
+		return defaultVacuumTaskPolicy(), nil
 	}
 
 	confDir := filepath.Join(cp.dataDir, ConfigSubdir)
@@ -318,18 +330,7 @@ func (cp *ConfigPersistence) LoadVacuumTaskPolicy() (*worker_pb.TaskPolicy, erro
 	// Check if file exists
 	if _, err := os.Stat(configPath); os.IsNotExist(err) {
 		// Return default policy if file doesn't exist
-		return &worker_pb.TaskPolicy{
-			Enabled:               true,
-			MaxConcurrent:         2,
-			RepeatIntervalSeconds: 24 * 3600, // 24 hours in seconds
-			CheckIntervalSeconds:  6 * 3600,  // 6 hours in seconds
-			TaskConfig: &worker_pb.TaskPolicy_VacuumConfig{
-				VacuumConfig: &worker_pb.VacuumTaskConfig{
-					GarbageThreshold:  0.3,
-					MinVolumeAgeHours: 24,
-				},
-			},
-		}, nil
+		return defaultVacuumTaskPolicy(), nil
 	}
 
 	// Read file
@@ -371,32 +372,14 @@ func (cp *ConfigPersistence) LoadErasureCodingTaskConfig() (*ErasureCodingTaskCo
 	}
 
 	// Return default config if no valid config found
-	return &ErasureCodingTaskConfig{
-		FullnessRatio:    0.9,
-		QuietForSeconds:  3600,
-		MinVolumeSizeMb:  1024,
-		CollectionFilter: "",
-	}, nil
+	return defaultErasureCodingTaskPolicy().GetErasureCodingConfig(), nil
 }
 
 // LoadErasureCodingTaskPolicy loads complete EC task policy from protobuf file
 func (cp *ConfigPersistence) LoadErasureCodingTaskPolicy() (*worker_pb.TaskPolicy, error) {
 	if cp.dataDir == "" {
 		// Return default policy if no data directory
-		return &worker_pb.TaskPolicy{
-			Enabled:               true,
-			MaxConcurrent:         1,
-			RepeatIntervalSeconds: 168 * 3600, // 1 week in seconds
-			CheckIntervalSeconds:  24 * 3600,  // 24 hours in seconds
-			TaskConfig: &worker_pb.TaskPolicy_ErasureCodingConfig{
-				ErasureCodingConfig: &worker_pb.ErasureCodingTaskConfig{
-					FullnessRatio:    0.9,
-					QuietForSeconds:  3600,
-					MinVolumeSizeMb:  1024,
-					CollectionFilter: "",
-				},
-			},
-		}, nil
+		return defaultErasureCodingTaskPolicy(), nil
 	}
 
 	confDir := filepath.Join(cp.dataDir, ConfigSubdir)
@@ -405,20 +388,7 @@ func (cp *ConfigPersistence) LoadErasureCodingTaskPolicy() (*worker_pb.TaskPolic
 	// Check if file exists
 	if _, err := os.Stat(configPath); os.IsNotExist(err) {
 		// Return default policy if file doesn't exist
-		return &worker_pb.TaskPolicy{
-			Enabled:               true,
-			MaxConcurrent:         1,
-			RepeatIntervalSeconds: 168 * 3600, // 1 week in seconds
-			CheckIntervalSeconds:  24 * 3600,  // 24 hours in seconds
-			TaskConfig: &worker_pb.TaskPolicy_ErasureCodingConfig{
-				ErasureCodingConfig: &worker_pb.ErasureCodingTaskConfig{
-					FullnessRatio:    0.9,
-					QuietForSeconds:  3600,
-					MinVolumeSizeMb:  1024,
-					CollectionFilter: "",
-				},
-			},
-		}, nil
+		return defaultErasureCodingTaskPolicy(), nil
 	}
 
 	// Read file
@@ -460,28 +430,14 @@ func (cp *ConfigPersistence) LoadBalanceTaskConfig() (*BalanceTaskConfig, error)
 	}
 
 	// Return default config if no valid config found
-	return &BalanceTaskConfig{
-		ImbalanceThreshold: 0.1,
-		MinServerCount:     2,
-	}, nil
+	return defaultBalanceTaskPolicy().GetBalanceConfig(), nil
 }
 
 // LoadBalanceTaskPolicy loads complete balance task policy from protobuf file
 func (cp *ConfigPersistence) LoadBalanceTaskPolicy() (*worker_pb.TaskPolicy, error) {
 	if cp.dataDir == "" {
 		// Return default policy if no data directory
-		return &worker_pb.TaskPolicy{
-			Enabled:               true,
-			MaxConcurrent:         1,
-			RepeatIntervalSeconds: 6 * 3600,  // 6 hours in seconds
-			CheckIntervalSeconds:  12 * 3600, // 12 hours in seconds
-			TaskConfig: &worker_pb.TaskPolicy_BalanceConfig{
-				BalanceConfig: &worker_pb.BalanceTaskConfig{
-					ImbalanceThreshold: 0.1,
-					MinServerCount:     2,
-				},
-			},
-		}, nil
+		return defaultBalanceTaskPolicy(), nil
 	}
 
 	confDir := filepath.Join(cp.dataDir, ConfigSubdir)
@@ -490,18 +446,7 @@ func (cp *ConfigPersistence) LoadBalanceTaskPolicy() (*worker_pb.TaskPolicy, err
 	// Check if file exists
 	if _, err := os.Stat(configPath); os.IsNotExist(err) {
 		// Return default policy if file doesn't exist
-		return &worker_pb.TaskPolicy{
-			Enabled:               true,
-			MaxConcurrent:         1,
-			RepeatIntervalSeconds: 6 * 3600,  // 6 hours in seconds
-			CheckIntervalSeconds:  12 * 3600, // 12 hours in seconds
-			TaskConfig: &worker_pb.TaskPolicy_BalanceConfig{
-				BalanceConfig: &worker_pb.BalanceTaskConfig{
-					ImbalanceThreshold: 0.1,
-					MinServerCount:     2,
-				},
-			},
-		}, nil
+		return defaultBalanceTaskPolicy(), nil
 	}
 
 	// Read file
@@ -523,6 +468,60 @@ func (cp *ConfigPersistence) LoadBalanceTaskPolicy() (*worker_pb.TaskPolicy, err
 	return nil, fmt.Errorf("failed to unmarshal balance task configuration")
 }
 
+// SaveEcBalanceTaskPolicy saves complete EC balance task policy to protobuf file
+func (cp *ConfigPersistence) SaveEcBalanceTaskPolicy(policy *worker_pb.TaskPolicy) error {
+	return cp.saveTaskConfig(EcBalanceTaskConfigFile, policy)
+}
+
+// LoadEcBalanceTaskConfig loads EC balance task configuration from protobuf file
+func (cp *ConfigPersistence) LoadEcBalanceTaskConfig() (*EcBalanceTaskConfig, error) {
+	if taskPolicy, err := cp.LoadEcBalanceTaskPolicy(); err == nil && taskPolicy != nil {
+		if ecBalanceConfig := taskPolicy.GetEcBalanceConfig(); ecBalanceConfig != nil {
+			return ecBalanceConfig, nil
+		}
+	}
+
+	// Return default config if no valid config found
+	return defaultEcBalanceTaskPolicy().GetEcBalanceConfig(), nil
+}
+
+// LoadEcBalanceTaskPolicy loads complete EC balance task policy from protobuf file.
+// ec_balance is registered like the other maintenance tasks and ec_balance.LoadConfigFromPersistence
+// asserts on this accessor, so without it the task could never be configured at all.
+func (cp *ConfigPersistence) LoadEcBalanceTaskPolicy() (*worker_pb.TaskPolicy, error) {
+	if cp.dataDir == "" {
+		// Return default policy if no data directory
+		return defaultEcBalanceTaskPolicy(), nil
+	}
+
+	confDir := filepath.Join(cp.dataDir, ConfigSubdir)
+	configPath := filepath.Join(confDir, EcBalanceTaskConfigFile)
+
+	// Check if file exists
+	if _, err := os.Stat(configPath); os.IsNotExist(err) {
+		// Return default policy if file doesn't exist
+		return defaultEcBalanceTaskPolicy(), nil
+	}
+
+	// Read file
+	configData, err := os.ReadFile(configPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read EC balance task config file: %w", err)
+	}
+
+	// Try to unmarshal as TaskPolicy
+	var policy worker_pb.TaskPolicy
+	if err := proto.Unmarshal(configData, &policy); err == nil {
+		// Validate that it's actually a TaskPolicy with EC balance config
+		if policy.GetEcBalanceConfig() != nil {
+			glog.V(1).Infof("Loaded EC balance task policy from %s", configPath)
+			return &policy, nil
+		}
+	}
+
+	return nil, fmt.Errorf("failed to unmarshal EC balance task configuration")
+}
+
 // SaveReplicationTaskConfig saves replication task configuration to protobuf file
 func (cp *ConfigPersistence) SaveReplicationTaskConfig(config *ReplicationTaskConfig) error {
 	return cp.saveTaskConfig(ReplicationTaskConfigFile, config)
@@ -632,6 +631,8 @@ func (cp *ConfigPersistence) SaveTaskPolicy(taskType string, policy *worker_pb.T
 		return cp.SaveErasureCodingTaskPolicy(policy)
 	case "balance":
 		return cp.SaveBalanceTaskPolicy(policy)
+	case "ec_balance":
+		return cp.SaveEcBalanceTaskPolicy(policy)
 	case "replication":
 		return cp.SaveReplicationTaskPolicy(policy)
 	}
@@ -687,69 +688,13 @@ func (cp *ConfigPersistence) GetConfigInfo() map[string]interface{} {
 	return info
 }
 
-// buildPolicyFromTaskConfigs loads task configurations from separate files and builds a MaintenancePolicy.
-// cp is passed to each task loader so the persisted configs are actually honoured; the loaders fall
-// back to compiled-in defaults for anything that has never been saved.
+// buildPolicyFromTaskConfigs builds the maintenance policy from the persisted task configs.
+//
+// The body lives in weed/admin/maintenance because the maintenance manager needs the same
+// policy when it has to build one itself, and this package already imports that one. Keeping
+// a second copy here is what let the two drift apart in the first place.
 func (cp *ConfigPersistence) buildPolicyFromTaskConfigs() *worker_pb.MaintenancePolicy {
-	policy := &worker_pb.MaintenancePolicy{
-		GlobalMaxConcurrent:          4,
-		DefaultRepeatIntervalSeconds: 6 * 3600,  // 6 hours in seconds
-		DefaultCheckIntervalSeconds:  12 * 3600, // 12 hours in seconds
-		TaskPolicies:                 make(map[string]*worker_pb.TaskPolicy),
-	}
-
-	// Load vacuum task configuration
-	if vacuumConfig := vacuum.LoadConfigFromPersistence(cp); vacuumConfig != nil {
-		policy.TaskPolicies["vacuum"] = &worker_pb.TaskPolicy{
-			Enabled:               vacuumConfig.Enabled,
-			MaxConcurrent:         int32(vacuumConfig.MaxConcurrent),
-			RepeatIntervalSeconds: int32(vacuumConfig.ScanIntervalSeconds),
-			CheckIntervalSeconds:  int32(vacuumConfig.ScanIntervalSeconds),
-			TaskConfig: &worker_pb.TaskPolicy_VacuumConfig{
-				VacuumConfig: &worker_pb.VacuumTaskConfig{
-					GarbageThreshold:  float64(vacuumConfig.GarbageThreshold),
-					MinVolumeAgeHours: int32((vacuumConfig.MinVolumeAgeSeconds + 3599) / 3600), // round up so sub-hour values don't become 0
-				},
-			},
-		}
-	}
-
-	// Load erasure coding task configuration
-	if ecConfig := erasure_coding.LoadConfigFromPersistence(cp); ecConfig != nil {
-		policy.TaskPolicies["erasure_coding"] = &worker_pb.TaskPolicy{
-			Enabled:               ecConfig.Enabled,
-			MaxConcurrent:         int32(ecConfig.MaxConcurrent),
-			RepeatIntervalSeconds: int32(ecConfig.ScanIntervalSeconds),
-			CheckIntervalSeconds:  int32(ecConfig.ScanIntervalSeconds),
-			TaskConfig: &worker_pb.TaskPolicy_ErasureCodingConfig{
-				ErasureCodingConfig: &worker_pb.ErasureCodingTaskConfig{
-					FullnessRatio:    float64(ecConfig.FullnessRatio),
-					QuietForSeconds:  int32(ecConfig.QuietForSeconds),
-					MinVolumeSizeMb:  int32(ecConfig.MinSizeMB),
-					CollectionFilter: ecConfig.CollectionFilter,
-				},
-			},
-		}
-	}
-
-	// Load balance task configuration
-	if balanceConfig := balance.LoadConfigFromPersistence(cp); balanceConfig != nil {
-		policy.TaskPolicies["balance"] = &worker_pb.TaskPolicy{
-			Enabled:               balanceConfig.Enabled,
-			MaxConcurrent:         int32(balanceConfig.MaxConcurrent),
-			RepeatIntervalSeconds: int32(balanceConfig.ScanIntervalSeconds),
-			CheckIntervalSeconds:  int32(balanceConfig.ScanIntervalSeconds),
-			TaskConfig: &worker_pb.TaskPolicy_BalanceConfig{
-				BalanceConfig: &worker_pb.BalanceTaskConfig{
-					ImbalanceThreshold: float64(balanceConfig.ImbalanceThreshold),
-					MinServerCount:     int32(balanceConfig.MinServerCount),
-				},
-			},
-		}
-	}
-
-	glog.V(1).Infof("Built maintenance policy from separate task configs - %d task policies loaded", len(policy.TaskPolicies))
-	return policy
+	return maintenance.BuildPolicyFromTaskConfigs(cp)
 }
 
 // SaveTaskDetail saves detailed task information to disk

weed/admin/dash/task_policy_defaults_test.go

@@ -0,0 +1,178 @@
+package dash
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
+	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance"
+	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/ec_balance"
+	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum"
+	"google.golang.org/protobuf/proto"
+)
+
+// TestLoadTaskPolicyDefaultsMatchTaskDefaults pins the persistence layer's "nothing saved
+// yet" defaults to each task's own NewDefaultConfig(). They used to be a second, hand-written
+// copy and had drifted: with a data directory but no config file on disk, vacuum ran on a 24h
+// scan interval instead of 2h, balance on 6h with a 0.1 imbalance threshold instead of 30m
+// with 0.2, and erasure coding on 168h with a 0.90 fullness ratio and a 1024MB minimum volume
+// size instead of 1h with 0.95 and 30MB - none of which is what the admin UI shows as the
+// default for those fields.
+func TestLoadTaskPolicyDefaultsMatchTaskDefaults(t *testing.T) {
+	cases := []struct {
+		name string
+		want *worker_pb.TaskPolicy
+		load func(cp *ConfigPersistence) (*worker_pb.TaskPolicy, error)
+	}{
+		{
+			name: "vacuum",
+			want: vacuum.NewDefaultConfig().ToTaskPolicy(),
+			load: func(cp *ConfigPersistence) (*worker_pb.TaskPolicy, error) { return cp.LoadVacuumTaskPolicy() },
+		},
+		{
+			name: "erasure_coding",
+			want: erasure_coding.NewDefaultConfig().ToTaskPolicy(),
+			load: func(cp *ConfigPersistence) (*worker_pb.TaskPolicy, error) {
+				return cp.LoadErasureCodingTaskPolicy()
+			},
+		},
+		{
+			name: "balance",
+			want: balance.NewDefaultConfig().ToTaskPolicy(),
+			load: func(cp *ConfigPersistence) (*worker_pb.TaskPolicy, error) { return cp.LoadBalanceTaskPolicy() },
+		},
+		{
+			name: "ec_balance",
+			want: ec_balance.NewDefaultConfig().ToTaskPolicy(),
+			load: func(cp *ConfigPersistence) (*worker_pb.TaskPolicy, error) { return cp.LoadEcBalanceTaskPolicy() },
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Both no-file branches have to agree with the task's own defaults: no data
+			// directory at all, and a data directory that has never been written to.
+			for _, cp := range []*ConfigPersistence{NewConfigPersistence(""), NewConfigPersistence(t.TempDir())} {
+				got, err := tc.load(cp)
+				if err != nil {
+					t.Fatalf("load %s policy: %v", tc.name, err)
+				}
+				if !proto.Equal(got, tc.want) {
+					t.Errorf("%s default policy (dataDir=%q) =\n  %v\nwant NewDefaultConfig().ToTaskPolicy() =\n  %v",
+						tc.name, cp.GetDataDir(), got, tc.want)
+				}
+			}
+		})
+	}
+}
+
+// TestLoadTaskConfigDefaultsMatchTaskDefaults covers the narrower Load*TaskConfig accessors,
+// which carried a third copy of the same defaults.
+func TestLoadTaskConfigDefaultsMatchTaskDefaults(t *testing.T) {
+	cp := NewConfigPersistence(t.TempDir())
+
+	vacuumConfig, err := cp.LoadVacuumTaskConfig()
+	if err != nil {
+		t.Fatalf("load vacuum config: %v", err)
+	}
+	if want := vacuum.NewDefaultConfig().ToTaskPolicy().GetVacuumConfig(); !proto.Equal(vacuumConfig, want) {
+		t.Errorf("vacuum default config = %v, want %v", vacuumConfig, want)
+	}
+
+	ecConfig, err := cp.LoadErasureCodingTaskConfig()
+	if err != nil {
+		t.Fatalf("load erasure coding config: %v", err)
+	}
+	if want := erasure_coding.NewDefaultConfig().ToTaskPolicy().GetErasureCodingConfig(); !proto.Equal(ecConfig, want) {
+		t.Errorf("erasure coding default config = %v, want %v", ecConfig, want)
+	}
+
+	balanceConfig, err := cp.LoadBalanceTaskConfig()
+	if err != nil {
+		t.Fatalf("load balance config: %v", err)
+	}
+	if want := balance.NewDefaultConfig().ToTaskPolicy().GetBalanceConfig(); !proto.Equal(balanceConfig, want) {
+		t.Errorf("balance default config = %v, want %v", balanceConfig, want)
+	}
+}
+
+// TestEcBalanceTaskPolicyRoundTrip checks the accessor ec_balance.LoadConfigFromPersistence
+// asserts on. Before it existed, ec_balance was the one registered maintenance task whose
+// configuration could not be persisted at all.
+func TestEcBalanceTaskPolicyRoundTrip(t *testing.T) {
+	cp := NewConfigPersistence(t.TempDir())
+
+	saved := ec_balance.NewDefaultConfig()
+	saved.Enabled = false
+	saved.MinServerCount = 9
+	saved.ImbalanceThreshold = 0.42
+	saved.CollectionFilter = "pictures"
+
+	if err := cp.SaveEcBalanceTaskPolicy(saved.ToTaskPolicy()); err != nil {
+		t.Fatalf("save ec_balance policy: %v", err)
+	}
+
+	loaded := ec_balance.LoadConfigFromPersistence(cp)
+	if loaded == nil {
+		t.Fatal("ec_balance.LoadConfigFromPersistence returned nil")
+	}
+	if loaded.Enabled {
+		t.Error("ec_balance enabled = true, want the persisted false")
+	}
+	if loaded.MinServerCount != 9 {
+		t.Errorf("ec_balance min server count = %d, want the persisted 9", loaded.MinServerCount)
+	}
+	if loaded.ImbalanceThreshold != 0.42 {
+		t.Errorf("ec_balance imbalance threshold = %v, want the persisted 0.42", loaded.ImbalanceThreshold)
+	}
+	if loaded.CollectionFilter != "pictures" {
+		t.Errorf("ec_balance collection filter = %q, want the persisted %q", loaded.CollectionFilter, "pictures")
+	}
+
+	// The generic dispatcher the maintenance manager uses has to know the type too.
+	if err := cp.SaveTaskPolicy("ec_balance", saved.ToTaskPolicy()); err != nil {
+		t.Errorf("SaveTaskPolicy(ec_balance): %v", err)
+	}
+}
+
+// TestBuildPolicyKeepsTaskSpecificFields guards the fields the hand-written policy builder
+// used to drop on the floor: the erasure coding preferred tags and replica placement, and
+// the balance IO rate limit. Building each entry from the task's own ToTaskPolicy() keeps
+// them, so a value set in admin.toml survives into the maintenance policy.
+func TestBuildPolicyKeepsTaskSpecificFields(t *testing.T) {
+	cp := NewConfigPersistence(t.TempDir())
+
+	ecConfig := erasure_coding.NewDefaultConfig()
+	ecConfig.PreferredTags = []string{"ssd", "archive"}
+	ecConfig.ReplicaPlacement = "020"
+	if err := cp.SaveErasureCodingTaskPolicy(ecConfig.ToTaskPolicy()); err != nil {
+		t.Fatalf("save erasure coding policy: %v", err)
+	}
+
+	balanceConfig := balance.NewDefaultConfig()
+	balanceConfig.IoBytePerSecond = 5 << 20
+	if err := cp.SaveBalanceTaskPolicy(balanceConfig.ToTaskPolicy()); err != nil {
+		t.Fatalf("save balance policy: %v", err)
+	}
+
+	policy := cp.buildPolicyFromTaskConfigs()
+
+	ecPolicy := policy.TaskPolicies["erasure_coding"].GetErasureCodingConfig()
+	if ecPolicy == nil {
+		t.Fatal("no erasure coding config in the built policy")
+	}
+	if got := ecPolicy.GetReplicaPlacement(); got != "020" {
+		t.Errorf("erasure coding replica placement = %q, want the persisted %q", got, "020")
+	}
+	if got := ecPolicy.GetPreferredTags(); len(got) != 2 {
+		t.Errorf("erasure coding preferred tags = %v, want the persisted 2 entries", got)
+	}
+
+	balancePolicy := policy.TaskPolicies["balance"].GetBalanceConfig()
+	if balancePolicy == nil {
+		t.Fatal("no balance config in the built policy")
+	}
+	if got := balancePolicy.GetIoBytePerSecond(); got != 5<<20 {
+		t.Errorf("balance IO limit = %d, want the persisted %d", got, 5<<20)
+	}
+}

weed/admin/maintenance/maintenance_integration.go

@@ -122,6 +122,15 @@ func (s *MaintenanceIntegration) registerAllTasks() {
 	glog.V(1).Infof("Registered tasks: %v", registeredTaskTypes)
 }
 
+// SetPolicy replaces the maintenance policy the integration configures tasks from and
+// applies it immediately. Without this the integration kept the policy it was built with,
+// so a policy updated at runtime reached the queue but never the detectors that decide
+// which task types are scanned for.
+func (s *MaintenanceIntegration) SetPolicy(policy *MaintenancePolicy) {
+	s.maintenancePolicy = policy
+	s.ConfigureTasksFromPolicy()
+}
+
 // ConfigureTasksFromPolicy dynamically configures all registered tasks based on the maintenance policy
 func (s *MaintenanceIntegration) ConfigureTasksFromPolicy() {
 	if s.maintenancePolicy == nil {
@@ -155,20 +164,32 @@ func (s *MaintenanceIntegration) configureDetectorFromPolicy(taskType types.Task
 		return
 	}
 
-	// Apply basic configuration that all detectors should support
-	if basicDetector, ok := detector.(interface{ SetEnabled(bool) }); ok {
-		// Convert task system type to maintenance task type for policy lookup
-		maintenanceTaskType, exists := s.taskTypeMap[taskType]
-		if exists {
-			enabled := IsTaskEnabled(s.maintenancePolicy, maintenanceTaskType)
-			basicDetector.SetEnabled(enabled)
-			glog.V(3).Infof("Set enabled=%v for detector %s", enabled, taskType)
-		}
+	// Convert task system type to maintenance task type for policy lookup
+	maintenanceTaskType, exists := s.taskTypeMap[taskType]
+	if !exists {
+		glog.V(3).Infof("No maintenance task type mapping for %s, skipping configuration", taskType)
+		return
 	}
 
-	// For detectors that don't implement PolicyConfigurableDetector interface,
-	// they should be updated to implement it for full policy-based configuration
-	glog.V(2).Infof("Detector %s should implement PolicyConfigurableDetector interface for full policy support", taskType)
+	// A task type the policy says nothing about is left alone. IsTaskEnabled reports
+	// false for a missing entry, so applying it unconditionally would silently disable
+	// every task the policy does not list - which is how ec_balance would have been
+	// switched off the moment SetEnabled started working.
+	if GetTaskPolicy(s.maintenancePolicy, maintenanceTaskType) == nil {
+		glog.V(2).Infof("Maintenance policy has no entry for %s, leaving its detector configuration untouched", taskType)
+		return
+	}
+
+	// Apply basic configuration that all detectors should support
+	if basicDetector, ok := detector.(interface{ SetEnabled(bool) }); ok {
+		enabled := IsTaskEnabled(s.maintenancePolicy, maintenanceTaskType)
+		basicDetector.SetEnabled(enabled)
+		glog.V(3).Infof("Set enabled=%v for detector %s", enabled, taskType)
+	} else {
+		// For detectors that don't implement PolicyConfigurableDetector interface,
+		// they should be updated to implement it for full policy-based configuration
+		glog.V(2).Infof("Detector %s supports neither PolicyConfigurableDetector nor SetEnabled, its policy is ignored", taskType)
+	}
 }
 
 // configureSchedulerFromPolicy configures a scheduler using policy-based configuration
@@ -187,11 +208,21 @@ func (s *MaintenanceIntegration) configureSchedulerFromPolicy(taskType types.Tas
 		return
 	}
 
+	// Same guard as on the detector side: no policy entry means no opinion, not disabled.
+	if GetTaskPolicy(s.maintenancePolicy, maintenanceTaskType) == nil {
+		glog.V(2).Infof("Maintenance policy has no entry for %s, leaving its scheduler configuration untouched", taskType)
+		return
+	}
+
 	// Set enabled status if scheduler supports it
 	if enableableScheduler, ok := scheduler.(interface{ SetEnabled(bool) }); ok {
 		enabled := IsTaskEnabled(s.maintenancePolicy, maintenanceTaskType)
 		enableableScheduler.SetEnabled(enabled)
 		glog.V(3).Infof("Set enabled=%v for scheduler %s", enabled, taskType)
+	} else {
+		// For schedulers that don't implement PolicyConfigurableScheduler interface,
+		// they should be updated to implement it for full policy-based configuration
+		glog.V(2).Infof("Scheduler %s supports neither PolicyConfigurableScheduler nor SetEnabled, its policy is ignored", taskType)
 	}
 
 	// Set max concurrent if scheduler supports it
@@ -202,10 +233,6 @@ func (s *MaintenanceIntegration) configureSchedulerFromPolicy(taskType types.Tas
 			glog.V(3).Infof("Set max concurrent=%d for scheduler %s", maxConcurrent, taskType)
 		}
 	}
-
-	// For schedulers that don't implement PolicyConfigurableScheduler interface,
-	// they should be updated to implement it for full policy-based configuration
-	glog.V(2).Infof("Scheduler %s should implement PolicyConfigurableScheduler interface for full policy support", taskType)
 }
 
 // ScanWithTaskDetectors performs a scan using the task system

weed/admin/maintenance/maintenance_manager.go

@@ -10,17 +10,29 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
 	stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/balance"
+	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/ec_balance"
 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum"
+	"github.com/seaweedfs/seaweedfs/weed/worker/types"
 )
 
-// buildPolicyFromTaskConfigs loads task configurations from separate files and builds a MaintenancePolicy.
+// BuildPolicyFromTaskConfigs loads each registered task's configuration and builds the
+// MaintenancePolicy the maintenance system runs on.
+//
+// Every entry is produced by the task's own ToTaskPolicy(), so a policy entry always carries
+// exactly what that task's config holds. Hand-copying the fields here instead made this a
+// second, silently diverging definition of every task's policy: it dropped the erasure
+// coding preferred tags and replica placement and the balance IO rate limit outright.
+//
+// Every task registered through base.RegisterTask needs an entry, because a task type the
+// policy does not list has no enabled flag and no concurrency limit of its own -
+// IsTaskEnabled reports false for a missing entry.
 //
 // configPersistence is duck-typed as interface{} because weed/admin/dash already imports this
 // package, so importing *dash.ConfigPersistence back here would create an import cycle. It must be
 // a value implementing the LoadXTaskPolicy() accessors the task loaders assert on; passing nil (or
 // anything else) makes every task fall back to its compiled-in defaults.
-func buildPolicyFromTaskConfigs(configPersistence interface{}) *worker_pb.MaintenancePolicy {
+func BuildPolicyFromTaskConfigs(configPersistence interface{}) *worker_pb.MaintenancePolicy {
 	policy := &worker_pb.MaintenancePolicy{
 		GlobalMaxConcurrent:          4,
 		DefaultRepeatIntervalSeconds: 6 * 3600,  // 6 hours in seconds
@@ -28,54 +40,20 @@ func buildPolicyFromTaskConfigs(configPersistence interface{}) *worker_pb.Mainte
 		TaskPolicies:                 make(map[string]*worker_pb.TaskPolicy),
 	}
 
-	// Load vacuum task configuration
 	if vacuumConfig := vacuum.LoadConfigFromPersistence(configPersistence); vacuumConfig != nil {
-		policy.TaskPolicies["vacuum"] = &worker_pb.TaskPolicy{
-			Enabled:               vacuumConfig.Enabled,
-			MaxConcurrent:         int32(vacuumConfig.MaxConcurrent),
-			RepeatIntervalSeconds: int32(vacuumConfig.ScanIntervalSeconds),
-			CheckIntervalSeconds:  int32(vacuumConfig.ScanIntervalSeconds),
-			TaskConfig: &worker_pb.TaskPolicy_VacuumConfig{
-				VacuumConfig: &worker_pb.VacuumTaskConfig{
-					GarbageThreshold:  float64(vacuumConfig.GarbageThreshold),
-					MinVolumeAgeHours: int32((vacuumConfig.MinVolumeAgeSeconds + 3599) / 3600), // round up so sub-hour values don't become 0
-				},
-			},
-		}
+		policy.TaskPolicies[string(types.TaskTypeVacuum)] = vacuumConfig.ToTaskPolicy()
 	}
 
-	// Load erasure coding task configuration
 	if ecConfig := erasure_coding.LoadConfigFromPersistence(configPersistence); ecConfig != nil {
-		policy.TaskPolicies["erasure_coding"] = &worker_pb.TaskPolicy{
-			Enabled:               ecConfig.Enabled,
-			MaxConcurrent:         int32(ecConfig.MaxConcurrent),
-			RepeatIntervalSeconds: int32(ecConfig.ScanIntervalSeconds),
-			CheckIntervalSeconds:  int32(ecConfig.ScanIntervalSeconds),
-			TaskConfig: &worker_pb.TaskPolicy_ErasureCodingConfig{
-				ErasureCodingConfig: &worker_pb.ErasureCodingTaskConfig{
-					FullnessRatio:    float64(ecConfig.FullnessRatio),
-					QuietForSeconds:  int32(ecConfig.QuietForSeconds),
-					MinVolumeSizeMb:  int32(ecConfig.MinSizeMB),
-					CollectionFilter: ecConfig.CollectionFilter,
-				},
-			},
-		}
+		policy.TaskPolicies[string(types.TaskTypeErasureCoding)] = ecConfig.ToTaskPolicy()
 	}
 
-	// Load balance task configuration
 	if balanceConfig := balance.LoadConfigFromPersistence(configPersistence); balanceConfig != nil {
-		policy.TaskPolicies["balance"] = &worker_pb.TaskPolicy{
-			Enabled:               balanceConfig.Enabled,
-			MaxConcurrent:         int32(balanceConfig.MaxConcurrent),
-			RepeatIntervalSeconds: int32(balanceConfig.ScanIntervalSeconds),
-			CheckIntervalSeconds:  int32(balanceConfig.ScanIntervalSeconds),
-			TaskConfig: &worker_pb.TaskPolicy_BalanceConfig{
-				BalanceConfig: &worker_pb.BalanceTaskConfig{
-					ImbalanceThreshold: float64(balanceConfig.ImbalanceThreshold),
-					MinServerCount:     int32(balanceConfig.MinServerCount),
-				},
-			},
-		}
+		policy.TaskPolicies[string(types.TaskTypeBalance)] = balanceConfig.ToTaskPolicy()
+	}
+
+	if ecBalanceConfig := ec_balance.LoadConfigFromPersistence(configPersistence); ecBalanceConfig != nil {
+		policy.TaskPolicies[string(types.TaskTypeECBalance)] = ecBalanceConfig.ToTaskPolicy()
 	}
 
 	glog.V(1).Infof("Built maintenance policy from separate task configs - %d task policies loaded", len(policy.TaskPolicies))
@@ -102,7 +80,7 @@ type MaintenanceManager struct {
 // NewMaintenanceManager creates a new maintenance manager.
 //
 // configPersistence is the config store to read persisted task configs from when the policy has to
-// be built here. See buildPolicyFromTaskConfigs for why it is duck-typed; pass nil when no config
+// be built here. See BuildPolicyFromTaskConfigs for why it is duck-typed; pass nil when no config
 // store is available.
 func NewMaintenanceManager(adminClient AdminClient, config *MaintenanceConfig, configPersistence interface{}) *MaintenanceManager {
 	if config == nil {
@@ -113,7 +91,7 @@ func NewMaintenanceManager(adminClient AdminClient, config *MaintenanceConfig, c
 	policy := config.Policy
 	if policy == nil {
 		// Fallback: build policy from separate task configuration files if not already populated
-		policy = buildPolicyFromTaskConfigs(configPersistence)
+		policy = BuildPolicyFromTaskConfigs(configPersistence)
 	}
 
 	queue := NewMaintenanceQueue(policy)
@@ -610,6 +588,14 @@ func (mm *MaintenanceManager) UpdateConfig(config *MaintenanceConfig) error {
 		mm.saveTaskConfigsFromPolicy(config.Policy)
 	}
 
+	// The integration holds its own reference to the policy and is what pushes the
+	// enabled flag and the concurrency limit into the registered detectors and
+	// schedulers. Without this the queue and the scanner saw the new policy while the
+	// detectors kept scanning under the old one until the next restart.
+	if mm.scanner != nil && mm.scanner.integration != nil {
+		mm.scanner.integration.SetPolicy(config.Policy)
+	}
+
 	glog.V(1).Infof("Maintenance configuration updated")
 	return nil
 }

weed/admin/maintenance/maintenance_policy_test.go

@@ -11,9 +11,10 @@ import (
 // import: weed/admin/dash already imports weed/admin/maintenance, so the dependency only runs one
 // way and the persistence argument has to stay duck-typed.
 type stubConfigPersistence struct {
-	vacuum  *worker_pb.TaskPolicy
-	ec      *worker_pb.TaskPolicy
-	balance *worker_pb.TaskPolicy
+	vacuum    *worker_pb.TaskPolicy
+	ec        *worker_pb.TaskPolicy
+	balance   *worker_pb.TaskPolicy
+	ecBalance *worker_pb.TaskPolicy
 }
 
 func (s *stubConfigPersistence) LoadVacuumTaskPolicy() (*worker_pb.TaskPolicy, error) {
@@ -28,6 +29,10 @@ func (s *stubConfigPersistence) LoadBalanceTaskPolicy() (*worker_pb.TaskPolicy,
 	return s.balance, nil
 }
 
+func (s *stubConfigPersistence) LoadEcBalanceTaskPolicy() (*worker_pb.TaskPolicy, error) {
+	return s.ecBalance, nil
+}
+
 func disabledStub() *stubConfigPersistence {
 	return &stubConfigPersistence{
 		vacuum: &worker_pb.TaskPolicy{
@@ -54,6 +59,14 @@ func disabledStub() *stubConfigPersistence {
 				BalanceConfig: &worker_pb.BalanceTaskConfig{ImbalanceThreshold: 0.2, MinServerCount: 7},
 			},
 		},
+		ecBalance: &worker_pb.TaskPolicy{
+			Enabled:               false,
+			MaxConcurrent:         1,
+			RepeatIntervalSeconds: 60 * 60,
+			TaskConfig: &worker_pb.TaskPolicy_EcBalanceConfig{
+				EcBalanceConfig: &worker_pb.EcBalanceTaskConfig{ImbalanceThreshold: 0.2, MinServerCount: 5},
+			},
+		},
 	}
 }
 
@@ -61,9 +74,9 @@ func disabledStub() *stubConfigPersistence {
 // https://github.com/seaweedfs/seaweedfs/issues/10874: the persistence argument used to be a
 // literal nil, which no type assertion can satisfy, so a task disabled on disk came back enabled.
 func TestBuildPolicyFromTaskConfigsUsesPersistence(t *testing.T) {
-	policy := buildPolicyFromTaskConfigs(disabledStub())
+	policy := BuildPolicyFromTaskConfigs(disabledStub())
 
-	for _, taskType := range []string{"vacuum", "erasure_coding", "balance"} {
+	for _, taskType := range []string{"vacuum", "erasure_coding", "balance", "ec_balance"} {
 		taskPolicy := policy.TaskPolicies[taskType]
 		if taskPolicy == nil {
 			t.Fatalf("no %s task policy built", taskType)
@@ -81,9 +94,9 @@ func TestBuildPolicyFromTaskConfigsUsesPersistence(t *testing.T) {
 // TestBuildPolicyFromTaskConfigsWithoutPersistence keeps the documented fallback: with no config
 // store there is nothing to read, so the compiled-in defaults apply.
 func TestBuildPolicyFromTaskConfigsWithoutPersistence(t *testing.T) {
-	policy := buildPolicyFromTaskConfigs(nil)
+	policy := BuildPolicyFromTaskConfigs(nil)
 
-	for _, taskType := range []string{"vacuum", "erasure_coding", "balance"} {
+	for _, taskType := range []string{"vacuum", "erasure_coding", "balance", "ec_balance"} {
 		taskPolicy := policy.TaskPolicies[taskType]
 		if taskPolicy == nil {
 			t.Fatalf("no %s task policy built", taskType)

weed/admin/maintenance/maintenance_policy_wiring_test.go

@@ -0,0 +1,179 @@
+package maintenance
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
+	"github.com/seaweedfs/seaweedfs/weed/worker/tasks"
+	"github.com/seaweedfs/seaweedfs/weed/worker/types"
+)
+
+// The detectors and schedulers live in a process-global registry, so these tests restore
+// whatever they change. Otherwise a test that disables a task leaks that state into every
+// later test in the package.
+func snapshotDetectorState(t *testing.T) {
+	t.Helper()
+
+	registry := tasks.GetGlobalTypesRegistry()
+	enabled := make(map[types.TaskType]bool)
+	maxConcurrent := make(map[types.TaskType]int)
+	for taskType, detector := range registry.GetAllDetectors() {
+		enabled[taskType] = detector.IsEnabled()
+	}
+	for taskType, scheduler := range registry.GetAllSchedulers() {
+		maxConcurrent[taskType] = scheduler.GetMaxConcurrent()
+	}
+
+	t.Cleanup(func() {
+		for taskType, detector := range registry.GetAllDetectors() {
+			if setter, ok := detector.(interface{ SetEnabled(bool) }); ok {
+				setter.SetEnabled(enabled[taskType])
+			}
+		}
+		for taskType, scheduler := range registry.GetAllSchedulers() {
+			if setter, ok := scheduler.(interface{ SetMaxConcurrent(int) }); ok {
+				setter.SetMaxConcurrent(maxConcurrent[taskType])
+			}
+		}
+	})
+}
+
+// TestPolicyReachesRegisteredDetectors is the regression test for the half of issue #10874
+// that a corrected policy alone did not fix: MaintenanceIntegration pushes the policy into
+// detectors and schedulers through interface{ SetEnabled(bool) } and
+// interface{ SetMaxConcurrent(int) } type assertions, but every task is backed by
+// base.GenericDetector/base.GenericScheduler, which implemented neither. The assertions
+// failed silently for every task on every startup, so the policy never reached
+// detector.IsEnabled() - which is what ScanWithTaskDetectors gates scanning on.
+func TestPolicyReachesRegisteredDetectors(t *testing.T) {
+	snapshotDetectorState(t)
+
+	registry := tasks.GetGlobalTypesRegistry()
+	if len(registry.GetAllDetectors()) == 0 {
+		t.Fatal("no detectors registered, the test cannot prove anything")
+	}
+
+	// Enable everything first so the disabling below cannot pass by accident.
+	policy := policyWithAllTasks(t, true)
+	NewMaintenanceIntegration(NewMaintenanceQueue(policy), policy)
+
+	for taskType, detector := range registry.GetAllDetectors() {
+		if !detector.IsEnabled() {
+			t.Fatalf("detector %s is disabled after an all-enabled policy was applied", taskType)
+		}
+	}
+
+	// Now disable everything through the policy and check it lands on the detectors.
+	policy = policyWithAllTasks(t, false)
+	NewMaintenanceIntegration(NewMaintenanceQueue(policy), policy)
+
+	for taskType, detector := range registry.GetAllDetectors() {
+		if detector.IsEnabled() {
+			t.Errorf("detector %s still reports enabled after the policy disabled it; "+
+				"the policy is not reaching the flag ScanWithTaskDetectors gates on", taskType)
+		}
+	}
+	for taskType, scheduler := range registry.GetAllSchedulers() {
+		if scheduler.IsEnabled() {
+			t.Errorf("scheduler %s still reports enabled after the policy disabled it", taskType)
+		}
+	}
+}
+
+// TestPolicyMaxConcurrentReachesSchedulers covers the SetMaxConcurrent half of the same
+// wiring. GetMaxConcurrent is what MaintenanceQueue.getMaxConcurrentForTaskType asks before
+// starting another task of a type.
+func TestPolicyMaxConcurrentReachesSchedulers(t *testing.T) {
+	snapshotDetectorState(t)
+
+	const wantMaxConcurrent = 7
+
+	policy := policyWithAllTasks(t, true)
+	for _, taskPolicy := range policy.TaskPolicies {
+		taskPolicy.MaxConcurrent = wantMaxConcurrent
+	}
+	NewMaintenanceIntegration(NewMaintenanceQueue(policy), policy)
+
+	registry := tasks.GetGlobalTypesRegistry()
+	for taskType, scheduler := range registry.GetAllSchedulers() {
+		if _, covered := policy.TaskPolicies[string(taskType)]; !covered {
+			continue
+		}
+		if got := scheduler.GetMaxConcurrent(); got != wantMaxConcurrent {
+			t.Errorf("scheduler %s max concurrent = %d, want %d from the policy", taskType, got, wantMaxConcurrent)
+		}
+	}
+}
+
+// TestPolicyWithoutEntryLeavesTaskAlone guards the direction that would have been a silent
+// outage: IsTaskEnabled reports false for a task type the policy does not list, so applying
+// it unconditionally would disable every task the policy has no entry for.
+func TestPolicyWithoutEntryLeavesTaskAlone(t *testing.T) {
+	snapshotDetectorState(t)
+
+	registry := tasks.GetGlobalTypesRegistry()
+
+	// Start from a policy that enables everything.
+	enabling := policyWithAllTasks(t, true)
+	NewMaintenanceIntegration(NewMaintenanceQueue(enabling), enabling)
+
+	// An empty policy has an entry for nothing at all.
+	empty := &MaintenancePolicy{TaskPolicies: make(map[string]*worker_pb.TaskPolicy)}
+	NewMaintenanceIntegration(NewMaintenanceQueue(empty), empty)
+
+	for taskType, detector := range registry.GetAllDetectors() {
+		if !detector.IsEnabled() {
+			t.Errorf("detector %s was disabled by a policy that has no entry for it; "+
+				"a missing entry means no opinion, not disabled", taskType)
+		}
+	}
+}
+
+// TestBuildPolicyCoversEveryRegisteredTask keeps BuildPolicyFromTaskConfigs and the task
+// registry in step. A registered task with no policy entry has no enabled flag and no
+// concurrency limit of its own, which is the state ec_balance was in.
+func TestBuildPolicyCoversEveryRegisteredTask(t *testing.T) {
+	policy := BuildPolicyFromTaskConfigs(nil)
+
+	for taskType := range tasks.GetGlobalTypesRegistry().GetAllDetectors() {
+		if _, ok := policy.TaskPolicies[string(taskType)]; !ok {
+			t.Errorf("task %s is registered as a detector but BuildPolicyFromTaskConfigs "+
+				"builds no policy entry for it, so IsTaskEnabled reports false for it", taskType)
+		}
+	}
+}
+
+// TestBuildPolicyIncludesEcBalance pins the specific gap above.
+func TestBuildPolicyIncludesEcBalance(t *testing.T) {
+	policy := BuildPolicyFromTaskConfigs(nil)
+
+	ecBalance := policy.TaskPolicies[string(types.TaskTypeECBalance)]
+	if ecBalance == nil {
+		t.Fatal("no ec_balance entry in the built maintenance policy")
+	}
+	if !IsTaskEnabled(policy, MaintenanceTaskType(types.TaskTypeECBalance)) {
+		t.Error("ec_balance reports disabled with no persisted config, want the compiled-in default of enabled")
+	}
+	if ecBalance.GetEcBalanceConfig() == nil {
+		t.Error("ec_balance policy entry carries no EcBalanceConfig")
+	}
+}
+
+// policyWithAllTasks builds a policy that has an entry for every registered task type,
+// all sharing the same enabled flag.
+func policyWithAllTasks(t *testing.T, enabled bool) *MaintenancePolicy {
+	t.Helper()
+
+	policy := &MaintenancePolicy{
+		GlobalMaxConcurrent: 4,
+		TaskPolicies:        make(map[string]*worker_pb.TaskPolicy),
+	}
+	for taskType := range tasks.GetGlobalTypesRegistry().GetAllDetectors() {
+		policy.TaskPolicies[string(taskType)] = &worker_pb.TaskPolicy{
+			Enabled:               enabled,
+			MaxConcurrent:         1,
+			RepeatIntervalSeconds: 3600,
+		}
+	}
+	return policy
+}

weed/worker/tasks/base/generic_components.go

@@ -42,6 +42,18 @@ func (d *GenericDetector) IsEnabled() bool {
 	return d.taskDef.Config.IsEnabled()
 }
 
+// SetEnabled turns detection for this task type on or off.
+//
+// The admin maintenance policy is applied to detectors through an
+// interface{ SetEnabled(bool) } type assertion (see
+// MaintenanceIntegration.configureDetectorFromPolicy). Every registered task is
+// backed by this generic detector, so without this method that assertion failed
+// for every task and the policy never reached the flag that
+// ScanWithTaskDetectors actually gates on. See issue #10874.
+func (d *GenericDetector) SetEnabled(enabled bool) {
+	d.taskDef.Config.SetEnabled(enabled)
+}
+
 // GenericScheduler implements TaskScheduler using function-based logic
 type GenericScheduler struct {
 	taskDef *TaskDefinition
@@ -127,3 +139,22 @@ func (s *GenericScheduler) GetDefaultRepeatInterval() time.Duration {
 func (s *GenericScheduler) IsEnabled() bool {
 	return s.taskDef.Config.IsEnabled()
 }
+
+// SetEnabled turns scheduling for this task type on or off. Detector and scheduler
+// share one TaskDefinition, so this is the same flag GenericDetector.SetEnabled sets;
+// both setters exist because the maintenance integration configures the two
+// independently. See GenericDetector.SetEnabled and issue #10874.
+func (s *GenericScheduler) SetEnabled(enabled bool) {
+	s.taskDef.Config.SetEnabled(enabled)
+}
+
+// SetMaxConcurrent applies the policy's concurrency limit for this task type. It is
+// the value GetMaxConcurrent returns, which the maintenance queue uses to decide
+// whether another task of this type may start. Non-positive limits are ignored
+// rather than turning into the implicit default of 1.
+func (s *GenericScheduler) SetMaxConcurrent(maxConcurrent int) {
+	if maxConcurrent <= 0 {
+		return
+	}
+	s.taskDef.MaxConcurrent = maxConcurrent
+}