diff --git a/src/rust/lqos_topology/src/lib.rs b/src/rust/lqos_topology/src/lib.rs index 884afc81..d483bdd6 100644 --- a/src/rust/lqos_topology/src/lib.rs +++ b/src/rust/lqos_topology/src/lib.rs @@ -2690,6 +2690,22 @@ fn logical_child_branch_counts(ui_state: &TopologyEditorStateFile) -> HashMap HashMap { + let mut counts = HashMap::new(); + for node in &effective.nodes { + let Some(attachment_id) = node.effective_attachment_id.as_deref() else { + continue; + }; + if attachment_id.trim().is_empty() { + continue; + } + *counts.entry(attachment_id.to_string()).or_insert(0) += 1; + } + counts +} + fn read_node_rate_mbps(node: &Map, key: &str) -> Option { node.get(key).and_then(|value| { value @@ -2709,6 +2725,7 @@ fn resolved_queue_visibility_policy( ui_node: &TopologyEditorNode, tree_node: Option<&Value>, child_branch_counts: &HashMap, + attachment_branch_counts: &HashMap, virtualization: &QueueVirtualizationContext, ) -> TopologyQueueVisibilityPolicy { if virtualization @@ -2729,6 +2746,7 @@ fn resolved_queue_visibility_policy( ui_node, tree_node, child_branch_counts, + attachment_branch_counts, virtualization, ), TopologyQueueVisibilityPolicy::QueueHiddenPromoteChildren => { @@ -2739,6 +2757,7 @@ fn resolved_queue_visibility_policy( ui_node, tree_node, child_branch_counts, + attachment_branch_counts, virtualization, ), } @@ -2749,6 +2768,7 @@ fn resolved_auto_queue_visibility_policy( ui_node: &TopologyEditorNode, tree_node: Option<&Value>, child_branch_counts: &HashMap, + attachment_branch_counts: &HashMap, _virtualization: &QueueVirtualizationContext, ) -> TopologyQueueVisibilityPolicy { let Some(tree_node) = tree_node.and_then(Value::as_object) else { @@ -2761,12 +2781,11 @@ fn resolved_auto_queue_visibility_policy( .get(ui_node.node_id.as_str()) .copied() .unwrap_or_default(); - let effective_child_count = tree_node - .get("children") - .and_then(Value::as_object) - .map(Map::len) + let attachment_child_count = attachment_branch_counts + .get(ui_node.node_id.as_str()) + .copied() .unwrap_or_default(); - if logical_child_count == 0 && effective_child_count == 0 { + if logical_child_count == 0 && attachment_child_count == 0 { return TopologyQueueVisibilityPolicy::QueueVisible; } let threshold = config.topology.queue_auto_virtualize_threshold_mbps; @@ -2985,6 +3004,7 @@ fn mark_node_virtual_by_id(map: &mut Map, target_id: &str) -> boo fn apply_queue_hidden_node_virtualization( config: &Config, ui_state: &TopologyEditorStateFile, + effective: &TopologyEffectiveStateFile, root: &mut Map, virtualization: &QueueVirtualizationContext, ) { @@ -2994,6 +3014,7 @@ fn apply_queue_hidden_node_virtualization( .map(|node| (node.node_id.as_str(), node)) .collect::>(); let child_branch_counts = logical_child_branch_counts(ui_state); + let attachment_branch_counts = effective_attachment_branch_counts(effective); let hidden_node_ids = queue_hidden_node_ids_in_promotion_order(ui_state); for hidden_node_id in hidden_node_ids { let Some(ui_node) = ui_by_id.get(hidden_node_id.as_str()).copied() else { @@ -3004,6 +3025,7 @@ fn apply_queue_hidden_node_virtualization( ui_node, find_node_by_id(root, &hidden_node_id), &child_branch_counts, + &attachment_branch_counts, virtualization, ); if resolved_policy != TopologyQueueVisibilityPolicy::QueueHiddenPromoteChildren { @@ -4265,6 +4287,7 @@ fn validate_effective_topology_network_from_canonical( count_node_ids(child, &mut counts); } let child_branch_counts = logical_child_branch_counts(ui_state); + let attachment_branch_counts = effective_attachment_branch_counts(effective); for node in &ui_state.nodes { if !node.node_id.contains(":site:") { @@ -4275,6 +4298,7 @@ fn validate_effective_topology_network_from_canonical( node, queue_policy_root.and_then(|root| find_node_by_id(root, &node.node_id)), &child_branch_counts, + &attachment_branch_counts, virtualization, ) == TopologyQueueVisibilityPolicy::QueueHiddenPromoteChildren { @@ -4340,7 +4364,7 @@ fn apply_effective_topology_to_network_json_from_canonical( let mut out = apply_effective_topology_reparenting_only(canonical_network, ui_state, effective); if let Some(root) = out.as_object_mut() { recompile_effective_network_bandwidths(root, canonical, ui_state, effective); - apply_queue_hidden_node_virtualization(config, ui_state, root, virtualization); + apply_queue_hidden_node_virtualization(config, ui_state, effective, root, virtualization); apply_runtime_squashing(config, ui_state, effective, root); } out @@ -7042,7 +7066,7 @@ mod tests { #[test] fn queue_auto_marks_large_ap_branch_virtual_with_effective_tree_children() { - let (config, canonical, mut editor_state, effective) = ap_branch_fixture(); + let (config, canonical, mut editor_state, mut effective) = ap_branch_fixture(); let access_ap = editor_state .nodes .iter_mut() @@ -7050,6 +7074,22 @@ mod tests { .expect("fixture should include Access AP"); access_ap.current_parent_node_id = Some("site-root".to_string()); access_ap.current_parent_node_name = Some("Core".to_string()); + let access_ap_effective = effective + .nodes + .iter_mut() + .find(|node| node.node_id == "ap-child") + .expect("fixture should include Access AP effective state"); + access_ap_effective.logical_parent_node_id = "site-root".to_string(); + access_ap_effective.effective_attachment_id = Some("ap-agg".to_string()); + access_ap_effective.attachments = vec![TopologyEffectiveAttachmentState { + attachment_id: "ap-agg".to_string(), + effective_selected: true, + health_reason: None, + health_status: TopologyAttachmentHealthStatus::Healthy, + probe_enabled: false, + probeable: false, + suppressed_until_unix: None, + }]; let effective_network = apply_effective_topology_to_network_json( &config, diff --git a/src/scheduler.py b/src/scheduler.py index 9db6b656..bc247d1c 100644 --- a/src/scheduler.py +++ b/src/scheduler.py @@ -60,6 +60,18 @@ class RequiredOverrideReadError(RuntimeError): """Raised when scheduler reloads cannot safely materialize operator overrides.""" +def mark_shaping_runtime_hash_applied(applied_hash): + """ + Record the runtime payload that a completed shaper refresh targeted. + + If topology runtime publishes a newer payload while refreshShapers() is + running, the next topology tick must still see that newer payload as + unapplied and run another refresh. + """ + global shaping_runtime_hash + shaping_runtime_hash = applied_hash + + def configure_scheduler_stdio(): """ Enable line-buffered scheduler logs when running under systemd. @@ -1109,6 +1121,7 @@ def importAndShapeFullReload(): ) return False publish_scheduler_progress(True, "initial_shaping_reload", "Refreshing shaper state", 5, SCHEDULER_STARTUP_STEP_COUNT) + refresh_target_hash = calculate_shaping_runtime_hash() if not enable_insight_topology(): try: refreshShapers() @@ -1116,9 +1129,7 @@ def importAndShapeFullReload(): if _defer_stale_shaping_inputs_failure(e, startup=True): return False raise - shaping_runtime_hash = calculate_shaping_runtime_hash() - else: - shaping_runtime_hash = calculate_shaping_runtime_hash() + mark_shaping_runtime_hash_applied(refresh_target_hash) if shaping_runtime_hash != 0: set_scheduler_status_bus_enabled(True) return True @@ -1166,7 +1177,7 @@ def importAndShapePartialReload(): return report_scheduler_runtime_failure("Scheduled shaping refresh failed", e) return - shaping_runtime_hash = calculate_shaping_runtime_hash() + mark_shaping_runtime_hash_applied(new_hash) if shaping_runtime_hash != 0: set_scheduler_status_bus_enabled(True) publish_ready_progress(False, "ready", "Scheduler ready", SCHEDULER_REFRESH_STEP_COUNT, SCHEDULER_REFRESH_STEP_COUNT, percent=100) @@ -1468,9 +1479,10 @@ def _continue_startup_topology_runtime_wait(): SCHEDULER_STARTUP_STEP_COUNT, ) try: + refresh_target_hash = calculate_shaping_runtime_hash() if not enable_insight_topology(): refreshShapers() - shaping_runtime_hash = calculate_shaping_runtime_hash() + mark_shaping_runtime_hash_applied(refresh_target_hash) except ValidationFailure as e: _reset_startup_topology_runtime_wait() report_scheduler_validation_failure( @@ -1594,7 +1606,7 @@ def _continue_partial_topology_runtime_wait(): _reset_partial_topology_runtime_wait() report_scheduler_runtime_failure("Scheduled shaping refresh failed", e) return - shaping_runtime_hash = calculate_shaping_runtime_hash() + mark_shaping_runtime_hash_applied(new_hash) if shaping_runtime_hash == 0: _reset_partial_topology_runtime_wait() report_scheduler_runtime_failure( @@ -1689,7 +1701,7 @@ def topology_runtime_refresh_tick(): return report_scheduler_runtime_failure("Topology runtime refresh failed", e) return - shaping_runtime_hash = calculate_shaping_runtime_hash() + mark_shaping_runtime_hash_applied(new_hash) if shaping_runtime_hash == 0: report_scheduler_runtime_failure( "Topology runtime refresh failed", diff --git a/src/test_scheduler.py b/src/test_scheduler.py index 11d035c2..3e4528ac 100644 --- a/src/test_scheduler.py +++ b/src/test_scheduler.py @@ -505,11 +505,7 @@ class TestSchedulerLogging(unittest.TestCase): "topology_runtime_readiness_detail", return_value=(True, "", "generation-1"), ): - with patch.object( - scheduler, - "calculate_shaping_runtime_hash", - side_effect=[5, 5], - ): + with patch.object(scheduler, "calculate_shaping_runtime_hash", return_value=5): with patch.object(scheduler, "refreshShapers") as mock_refresh: with patch.object(scheduler, "clear_scheduler_error") as mock_clear_error: with patch.object(scheduler, "publish_scheduler_progress") as mock_progress: @@ -530,28 +526,25 @@ class TestSchedulerLogging(unittest.TestCase): ) ) - def test_topology_runtime_refresh_tick_reports_missing_hash_after_success(self): + def test_topology_runtime_refresh_tick_records_hash_that_triggered_refresh(self): with patch.object(scheduler, "ensure_topology_runtime_process"): with patch.object( scheduler, "topology_runtime_readiness_detail", return_value=(True, "", "generation-1"), ): - with patch.object( - scheduler, - "calculate_shaping_runtime_hash", - side_effect=[5, 0], - ): - with patch.object(scheduler, "refreshShapers"): + with patch.object(scheduler, "calculate_shaping_runtime_hash") as mock_hash: + mock_hash.side_effect = [5] + with patch.object(scheduler, "refreshShapers") as mock_refresh: with patch.object(scheduler, "scheduler_error") as mock_scheduler_error: - with patch.object(scheduler, "clear_scheduler_error") as mock_clear_error: + with patch.object(scheduler, "clear_scheduler_error"): scheduler.shaping_runtime_hash = 1 scheduler.topology_runtime_refresh_tick() - mock_clear_error.assert_not_called() - mock_scheduler_error.assert_called_once_with( - "Topology runtime refresh failed: Shaping runtime hash was not published after topology runtime refresh." - ) + mock_refresh.assert_called_once() + mock_scheduler_error.assert_not_called() + self.assertEqual(scheduler.shaping_runtime_hash, 5) + self.assertEqual(mock_hash.call_count, 1) def test_topology_runtime_refresh_tick_skips_until_initial_shaping_succeeds(self): with patch.object(scheduler, "ensure_topology_runtime_process") as mock_ensure: @@ -719,7 +712,7 @@ class TestSchedulerLogging(unittest.TestCase): return_value=(True, "", "generation-1"), ): with patch.object(scheduler, "refreshShapers") as mock_refresh: - with patch.object(scheduler, "calculate_shaping_runtime_hash", return_value=9): + with patch.object(scheduler, "calculate_shaping_runtime_hash", return_value=9) as mock_hash: with patch.object(scheduler, "publish_scheduler_progress") as mock_progress: with patch.object(scheduler, "clear_scheduler_error") as mock_clear_error: with patch("builtins.print"): @@ -727,6 +720,7 @@ class TestSchedulerLogging(unittest.TestCase): mock_refresh.assert_called_once() mock_clear_error.assert_called_once() + mock_hash.assert_called_once() self.assertEqual(scheduler.shaping_runtime_hash, 9) self.assertFalse(scheduler.startup_topology_runtime_pending) self.assertTrue( @@ -777,7 +771,7 @@ class TestSchedulerLogging(unittest.TestCase): with patch.object( scheduler, "calculate_shaping_runtime_hash", - side_effect=[9, 9], + return_value=9, ): with patch.object(scheduler, "publish_scheduler_progress") as mock_progress: with patch.object(scheduler, "clear_scheduler_error") as mock_clear_error: