diff --git a/CHANGELOG.md b/CHANGELOG.md index a9ef68e9e..5e2e4b629 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fleet node heartbeats no longer report unbounded capacity as `load: 0`; `load` is omitted when `max_agents` is `0` and remains a measured `[0,1]` utilization ratio when a finite agent limit is configured. - `agent-relay node up` resolves its installed broker through canonical package-manager links and Relay's user install directories, so mise-managed and minimal-`PATH` launches no longer fail when the broker binary is already installed. - `agent-relay node up` warns instead of silently ignoring stored Cloud fleet enrollments when the project workspace pin has no enrolled node id. That combination started the broker in the pinned workspace while the node never heartbeat, leaving the Cloud dashboard and `agent-relay fleet nodes` showing different rosters with no error from either. - `agent-relay cloud enroll` records the enrolled node on the project workspace pin, so `node up` in that repo serves the node it just enrolled. A pin that already names a different node is reported and left untouched rather than repointed. diff --git a/crates/broker/src/fleet_wire.rs b/crates/broker/src/fleet_wire.rs index 9ba34c870..b44b42e15 100644 --- a/crates/broker/src/fleet_wire.rs +++ b/crates/broker/src/fleet_wire.rs @@ -168,11 +168,17 @@ pub struct NodeHeartbeat { pub capabilities: Vec, pub max_agents: u32, pub version: String, + // Capacity utilization is undefined for an unbounded provider + // (`max_agents == 0`). Omit it instead of reporting a false idle `0`. + // Requires the Relaycast engine to accept an omitted/null `load` before + // this broker version is deployed (relaycast#307). #[serde( - deserialize_with = "deserialize_finite_nonnegative_f64", - serialize_with = "serialize_finite_nonnegative_f64" + default, + deserialize_with = "deserialize_optional_finite_nonnegative_f64", + serialize_with = "serialize_optional_finite_nonnegative_f64", + skip_serializing_if = "Option::is_none" )] - pub load: f64, + pub load: Option, pub active_agents: u32, pub handlers_live: bool, } @@ -322,12 +328,16 @@ where T::deserialize(deserializer).map(Some) } -fn deserialize_finite_nonnegative_f64<'de, D>(deserializer: D) -> Result +fn deserialize_optional_finite_nonnegative_f64<'de, D>( + deserializer: D, +) -> Result, D::Error> where D: Deserializer<'de>, { - let value = f64::deserialize(deserializer)?; - validate_finite_nonnegative_f64(value).map_err(de::Error::custom) + Option::::deserialize(deserializer)? + .map(validate_finite_nonnegative_f64) + .transpose() + .map_err(de::Error::custom) } fn serialize_finite_nonnegative_f64(value: &f64, serializer: S) -> Result @@ -338,6 +348,19 @@ where serializer.serialize_f64(*value) } +fn serialize_optional_finite_nonnegative_f64( + value: &Option, + serializer: S, +) -> Result +where + S: Serializer, +{ + match value { + Some(value) => serialize_finite_nonnegative_f64(value, serializer), + None => serializer.serialize_none(), + } +} + fn validate_finite_nonnegative_f64(value: f64) -> Result { if !value.is_finite() { return Err("load must be finite"); @@ -345,6 +368,9 @@ fn validate_finite_nonnegative_f64(value: f64) -> Result { if value < 0.0 { return Err("load must be nonnegative"); } + if value > 1.0 { + return Err("load must be at most 1"); + } Ok(value) } @@ -633,11 +659,10 @@ pub type RelaycastToBroker = ServerToNode; #[cfg(test)] mod tests { - use serde::de::{value::Error as DeError, IntoDeserializer}; use serde_json::{json, Value}; use super::{ - deserialize_finite_nonnegative_f64, validate_agent_register_reply_data, ActionResult, + validate_agent_register_reply_data, validate_finite_nonnegative_f64, ActionResult, ActionResultError, ActionResultPayload, AgentRegister, BrokerToRelaycast, Deliver, DeliveryMode, Error, FleetCapability, NodeHeartbeat, RelaycastToBroker, Reply, FLEET_WIRE_VERSION, @@ -716,10 +741,51 @@ mod tests { }); assert!(serde_json::from_value::(negative).is_err()); + let over_capacity = json!({ + "type": "node.heartbeat", + "v": 1, + "name": "builder-1", + "node_id": "node_1", + "capabilities": [], + "max_agents": 1, + "version": "relay-broker/test", + "load": 1.1, + "active_agents": 2, + "handlers_live": true + }); + assert!(serde_json::from_value::(over_capacity).is_err()); + for load in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] { - let result: Result = - deserialize_finite_nonnegative_f64(load.into_deserializer()); - assert!(result.is_err(), "expected load {load:?} to be rejected"); + assert!( + validate_finite_nonnegative_f64(load).is_err(), + "expected load {load:?} to be rejected" + ); + } + + for load in [0.0, 1.0] { + assert_eq!( + validate_finite_nonnegative_f64(load), + Ok(load), + "expected load {load:?} to be accepted" + ); + let boundary = json!({ + "type": "node.heartbeat", + "v": 1, + "name": "builder-1", + "node_id": "node_1", + "capabilities": [], + "max_agents": 1, + "version": "relay-broker/test", + "load": load, + "active_agents": 1, + "handlers_live": true + }); + let decoded: BrokerToRelaycast = serde_json::from_value(boundary) + .unwrap_or_else(|err| panic!("expected load {load:?} to decode: {err}")); + match decoded { + BrokerToRelaycast::NodeHeartbeat(hb) => assert_eq!(hb.load, Some(load)), + other => panic!("expected NodeHeartbeat, got {other:?}"), + } } let invalid = BrokerToRelaycast::NodeHeartbeat(NodeHeartbeat { @@ -731,7 +797,7 @@ mod tests { capabilities: vec![], max_agents: 1, version: "relay-broker/test".to_string(), - load: f64::INFINITY, + load: Some(f64::INFINITY), active_agents: 0, handlers_live: true, }); @@ -765,7 +831,7 @@ mod tests { }], max_agents: 4, version: "relay-broker/test".to_string(), - load: 0.25, + load: Some(0.25), active_agents: 1, handlers_live: true, }); @@ -799,6 +865,70 @@ mod tests { ); } + #[test] + fn node_heartbeat_omits_unreported_load() { + let msg = BrokerToRelaycast::NodeHeartbeat(NodeHeartbeat { + v: FLEET_WIRE_VERSION, + id: None, + provider: None, + name: "unbounded-builder".to_string(), + node_id: "node_unbounded".to_string(), + capabilities: vec![], + max_agents: 0, + version: "relay-broker/test".to_string(), + load: None, + active_agents: 25, + handlers_live: true, + }); + + let value = serde_json::to_value(msg).unwrap(); + assert_eq!(value.get("load"), None); + assert_eq!(value["active_agents"], 25); + assert_eq!(value["max_agents"], 0); + + // Decode side: a heartbeat with the `load` field absent entirely, and + // one with an explicit `"load": null` (e.g. from a relay that + // round-trips the omitted value), must both decode to `load: None`. + let missing_field = json!({ + "type": "node.heartbeat", + "v": 1, + "name": "unbounded-builder", + "node_id": "node_unbounded", + "capabilities": [], + "max_agents": 0, + "version": "relay-broker/test", + "active_agents": 25, + "handlers_live": true + }); + let decoded: BrokerToRelaycast = serde_json::from_value(missing_field).unwrap(); + match decoded { + BrokerToRelaycast::NodeHeartbeat(hb) => { + assert_eq!(hb.load, None); + assert_eq!(hb.active_agents, 25); + assert_eq!(hb.max_agents, 0); + } + other => panic!("expected NodeHeartbeat, got {other:?}"), + } + + let explicit_null = json!({ + "type": "node.heartbeat", + "v": 1, + "name": "unbounded-builder", + "node_id": "node_unbounded", + "capabilities": [], + "max_agents": 0, + "version": "relay-broker/test", + "load": null, + "active_agents": 25, + "handlers_live": true + }); + let decoded: BrokerToRelaycast = serde_json::from_value(explicit_null).unwrap(); + match decoded { + BrokerToRelaycast::NodeHeartbeat(hb) => assert_eq!(hb.load, None), + other => panic!("expected NodeHeartbeat, got {other:?}"), + } + } + #[test] fn node_register_absent_resume_cursor_serializes_as_null() { let decoded: BrokerToRelaycast = serde_json::from_value(json!({ diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index 501e49abb..64faee49b 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -464,9 +464,9 @@ impl FleetLoadSnapshot { /// time server-side as the single source of truth for liveness. fn heartbeat(&self, node: &NodeRegister) -> NodeHeartbeat { let load = if self.max_agents == 0 { - 0.0 + None } else { - (self.active_agents as f64 / self.max_agents as f64).clamp(0.0, 1.0) + Some((self.active_agents as f64 / self.max_agents as f64).clamp(0.0, 1.0)) }; NodeHeartbeat { v: FLEET_WIRE_VERSION, @@ -3545,6 +3545,33 @@ mod tests { } } + #[test] + fn heartbeat_reports_only_measured_capacity_load() { + let register = build_node_register( + &test_manifest(), + "node-default", + "host-default", + "broker/test", + None, + ); + + let measured = FleetLoadSnapshot { + active_agents: 3, + max_agents: 4, + handlers_live: true, + } + .heartbeat(®ister); + assert_eq!(measured.load, Some(0.75)); + + let unbounded = FleetLoadSnapshot { + active_agents: 25, + max_agents: 0, + handlers_live: true, + } + .heartbeat(®ister); + assert_eq!(unbounded.load, None); + } + #[test] fn load_node_token_round_trips_when_node_and_workspace_match() { let dir = tempfile::tempdir().unwrap(); diff --git a/packages/sdk/src/__tests__/messaging.test.ts b/packages/sdk/src/__tests__/messaging.test.ts index 286146657..f7f01a884 100644 --- a/packages/sdk/src/__tests__/messaging.test.ts +++ b/packages/sdk/src/__tests__/messaging.test.ts @@ -485,6 +485,7 @@ describe('RelaycastMessagingClient', () => { }).repoKeys ).toEqual(['explicit']); expect(toRelayNode({ name: 'builder-6', tags: ['factory'] }).repoKeys).toBeUndefined(); + expect(toRelayNode({ name: 'builder-unbounded', max_agents: 0, load: null }).load).toBeUndefined(); await expect(client.nodes.get('builder-2')).resolves.toMatchObject({ name: 'builder-2',