diff --git a/weed/admin/topology/active_topology_test.go b/weed/admin/topology/active_topology_test.go index 9b0990f21..a8b3f8085 100644 --- a/weed/admin/topology/active_topology_test.go +++ b/weed/admin/topology/active_topology_test.go @@ -261,11 +261,11 @@ func TestTargetSelectionScenarios(t *testing.T) { expectedTargets: 4, // All 4 disks available }, { - name: "Vacuum task - avoid conflicting disks", + name: "Vacuum task - cross-type tasks do not block per disk", topology: createTopologyWithConflicts(), taskType: TaskTypeVacuum, excludeNode: "", - expectedTargets: 1, // Only 1 disk without conflicts (conflicts exclude more disks) + expectedTargets: 4, // All 4 disks available; per-volume safety is enforced by HasAnyTask }, } @@ -279,8 +279,6 @@ func TestTargetSelectionScenarios(t *testing.T) { for _, disk := range availableDisks { assert.NotEqual(t, tt.excludeNode, disk.NodeID, "Available disk should not be on excluded node") - - assert.Less(t, disk.LoadCount, 2, "Disk load should be less than 2") } }) } @@ -353,12 +351,10 @@ func TestDiskLoadCalculation(t *testing.T) { assert.Equal(t, 1, targetDisk.LoadCount) } -// TestTaskConflictDetection tests task conflict detection -func TestTaskConflictDetection(t *testing.T) { +func TestCrossTypeTasksDoNotBlockPerDisk(t *testing.T) { topology := NewActiveTopology(10) topology.UpdateTopology(createSampleTopology()) - // Add a balance task err := topology.AddPendingTask(TaskSpec{ TaskID: "balance1", TaskType: TaskTypeBalance, @@ -371,13 +367,10 @@ func TestTaskConflictDetection(t *testing.T) { {ServerID: "10.0.0.2:8080", DiskID: 1}, }, }) - assert.NoError(t, err, "Should add balance task successfully") - topology.AssignTask("balance1") + require.NoError(t, err) + require.NoError(t, topology.AssignTask("balance1")) - // Try to get available disks for vacuum (conflicts with balance) availableDisks := topology.GetAvailableDisks(TaskTypeVacuum, "") - - // Source disk should not be available due to conflict sourceDiskAvailable := false for _, disk := range availableDisks { if disk.NodeID == "10.0.0.1:8080" && disk.DiskID == 0 { @@ -385,7 +378,34 @@ func TestTaskConflictDetection(t *testing.T) { break } } - assert.False(t, sourceDiskAvailable, "Source disk should not be available due to task conflict") + assert.True(t, sourceDiskAvailable, + "Source disk should remain available for an unrelated task type") +} + +// Regression for #9147: a 4-disk cluster with one in-flight balance task must +// still expose all 4 disks to EC placement so MinTotalDisks can be satisfied. +func TestECPlanningNotBlockedByUnrelatedBalance(t *testing.T) { + topology := NewActiveTopology(10) + topology.UpdateTopology(createSampleTopology()) // 2 nodes x 2 disks + + err := topology.AddPendingTask(TaskSpec{ + TaskID: "balance1", + TaskType: TaskTypeBalance, + VolumeID: 42, + VolumeSize: 1024 * 1024 * 1024, + Sources: []TaskSourceSpec{ + {ServerID: "10.0.0.1:8080", DiskID: 0}, + }, + Destinations: []TaskDestinationSpec{ + {ServerID: "10.0.0.2:8080", DiskID: 0}, + }, + }) + require.NoError(t, err) + require.NoError(t, topology.AssignTask("balance1")) + + ecCandidates := topology.GetDisksWithEffectiveCapacity(TaskTypeErasureCoding, "", 0) + assert.Equal(t, 4, len(ecCandidates), + "EC must still see all 4 disks even with an unrelated in-flight balance") } // TestPublicInterfaces tests the public interface methods diff --git a/weed/admin/topology/capacity.go b/weed/admin/topology/capacity.go index 504990d58..e59494974 100644 --- a/weed/admin/topology/capacity.go +++ b/weed/admin/topology/capacity.go @@ -236,21 +236,13 @@ func (at *ActiveTopology) getPlanningCapacityUnsafe(disk *activeDisk) StorageSlo } } -// isDiskAvailableForPlanning checks if disk can accept new tasks considering pending load +// isDiskAvailableForPlanning checks if disk can accept new tasks considering +// pending load. See isDiskAvailable for the cross-type policy. func (at *ActiveTopology) isDiskAvailableForPlanning(disk *activeDisk, taskType TaskType) bool { - // Check total load including pending tasks totalLoad := len(disk.pendingTasks) + len(disk.assignedTasks) if MaxTotalTaskLoadPerDisk > 0 && totalLoad >= MaxTotalTaskLoadPerDisk { return false } - - // Check for conflicting task types in active tasks only - for _, task := range disk.assignedTasks { - if at.areTaskTypesConflicting(task.TaskType, taskType) { - return false - } - } - return true } diff --git a/weed/admin/topology/internal.go b/weed/admin/topology/internal.go index 96ef0ed20..733b2526e 100644 --- a/weed/admin/topology/internal.go +++ b/weed/admin/topology/internal.go @@ -64,44 +64,17 @@ func (at *ActiveTopology) assignTaskToDisk(task *taskState) { } } -// isDiskAvailable checks if a disk can accept new tasks +// isDiskAvailable checks if a disk can accept new tasks. Per-volume safety is +// enforced by HasAnyTask at detection time, so cross-type tasks on the same +// disk are intentionally not considered conflicting (see #9147). func (at *ActiveTopology) isDiskAvailable(disk *activeDisk, taskType TaskType) bool { - // Check if disk has too many pending and active tasks activeLoad := len(disk.pendingTasks) + len(disk.assignedTasks) if MaxConcurrentTasksPerDisk > 0 && activeLoad >= MaxConcurrentTasksPerDisk { return false } - - // Check for conflicting task types - for _, task := range disk.assignedTasks { - if at.areTaskTypesConflicting(task.TaskType, taskType) { - return false - } - } - return true } -// areTaskTypesConflicting checks if two task types conflict -func (at *ActiveTopology) areTaskTypesConflicting(existing, new TaskType) bool { - // Examples of conflicting task types - conflictMap := map[TaskType][]TaskType{ - TaskTypeVacuum: {TaskTypeBalance, TaskTypeErasureCoding}, - TaskTypeBalance: {TaskTypeVacuum, TaskTypeErasureCoding}, - TaskTypeErasureCoding: {TaskTypeVacuum, TaskTypeBalance}, - } - - if conflicts, exists := conflictMap[existing]; exists { - for _, conflictType := range conflicts { - if conflictType == new { - return true - } - } - } - - return false -} - // cleanupRecentTasks removes old recent tasks func (at *ActiveTopology) cleanupRecentTasks() { cutoff := time.Now().Add(-time.Duration(at.recentTaskWindowSeconds) * time.Second)