feat(bf-28dv): implement zone-aware rate control (Component 24)
Implement fleet-level sentinel link coordination with zone-aware idle modes: - Zone membership tracking: Track which nodes belong to which zones and update zone state on motion events - Zone idle detection: Detect when all nodes in a zone have been idle and designate one sentinel link per zone at 5 Hz - Sentinel designation: Lexicographically smallest MAC becomes sentinel (5 Hz), others drop to 1 Hz - Fleet idle coordination: When all zones are idle, only sentinel links run at 5 Hz for minimal bandwidth - Adjacent zone ramping: When motion detected in a zone, ramp adjacent zones to 5 Hz for preemptive coverage - RampZone() method: Interface for prediction engine to preemptively ramp zones when P(arrival) > threshold - Zone-aware tests: Comprehensive tests for zone membership, idle detection, sentinel designation, fleet coordination, and backward compatibility Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
jedarden
parent
e13b2e220b
commit
9c36b0dab3
3 changed files with 801 additions and 2 deletions
|
|
@ -11,6 +11,12 @@ const (
|
|||
RateIdle = 2
|
||||
// RateActive is the CSI sampling rate (Hz) when motion is detected.
|
||||
RateActive = 50
|
||||
// RateSentinel is the CSI sampling rate (Hz) for the designated sentinel link in an idle zone.
|
||||
// Runs at 5 Hz to maintain minimal coverage while reducing bandwidth.
|
||||
RateSentinel = 5
|
||||
// RateFleetIdle is the CSI sampling rate (Hz) for non-sentinel links when the entire fleet is idle.
|
||||
// Only one sentinel per zone runs at RateSentinel; all others run at 1 Hz.
|
||||
RateFleetIdle = 1
|
||||
// idleTimeout is how long after the last motion event before dropping back to idle.
|
||||
idleTimeout = 30 * time.Second
|
||||
|
||||
|
|
@ -27,15 +33,34 @@ type nodeRateState struct {
|
|||
lastMotionAt time.Time
|
||||
}
|
||||
|
||||
// zoneRateState tracks the adaptive rate state for a single zone.
|
||||
type zoneRateState struct {
|
||||
zoneID string
|
||||
nodes map[string]bool // nodes in this zone (keyed by MAC)
|
||||
sentinelNodeMAC string // designated sentinel node (may be empty)
|
||||
allNodesIdle bool // true when all nodes in zone are idle
|
||||
lastMotionAt time.Time // last motion in this zone
|
||||
}
|
||||
|
||||
// RateController manages per-node adaptive sensing rates. When motion is detected
|
||||
// on a node's link, it ramps that node to RateActive (50 Hz). When no motion has
|
||||
// been seen for idleTimeout, it drops back to RateIdle (2 Hz). The caller provides
|
||||
// a configSender callback that sends the rate and variance threshold to the node.
|
||||
//
|
||||
// Zone-aware mode: When SetZoneMembershipFn is called, the controller also tracks
|
||||
// zone-level idle states and designates sentinel links per zone at RateSentinel (5 Hz)
|
||||
// when the fleet is idle, with all other links at RateFleetIdle (1 Hz).
|
||||
type RateController struct {
|
||||
mu sync.Mutex
|
||||
nodes map[string]*nodeRateState // keyed by node MAC
|
||||
configSender func(nodeMAC string, rateHz int, varianceThreshold float64)
|
||||
adjacentNodes func(nodeMAC string) []string // returns MACs of adjacent nodes; may be nil
|
||||
|
||||
// Zone-aware fields
|
||||
zones map[string]*zoneRateState // keyed by zone ID
|
||||
zoneMembership func(nodeMAC string) []string // returns zone IDs for a node
|
||||
adjacentZones func(zoneID string) []string // returns zone IDs adjacent to a zone
|
||||
fleetIdle bool // true when all zones are idle
|
||||
}
|
||||
|
||||
// NewRateController creates a RateController. configSender is called whenever a
|
||||
|
|
@ -44,6 +69,7 @@ type RateController struct {
|
|||
func NewRateController(configSender func(nodeMAC string, rateHz int, varianceThreshold float64)) *RateController {
|
||||
return &RateController{
|
||||
nodes: make(map[string]*nodeRateState),
|
||||
zones: make(map[string]*zoneRateState),
|
||||
configSender: configSender,
|
||||
}
|
||||
}
|
||||
|
|
@ -58,6 +84,24 @@ func (rc *RateController) SetAdjacentNodesFn(fn func(nodeMAC string) []string) {
|
|||
rc.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetZoneMembershipFn configures a callback that returns the zone IDs for a given node.
|
||||
// When set, the controller tracks zone-level idle states and designates sentinel links.
|
||||
// The function should return zone IDs based on the node's position (X, Y, Z) within zone bounds.
|
||||
func (rc *RateController) SetZoneMembershipFn(fn func(nodeMAC string) []string) {
|
||||
rc.mu.Lock()
|
||||
rc.zoneMembership = fn
|
||||
rc.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetAdjacentZonesFn configures a callback that returns zone IDs adjacent to a given zone.
|
||||
// When set, OnMotionState preemptively ramps adjacent zones to RateSentinel when motion
|
||||
// is detected in a zone. The function should return zone IDs based on portal connections.
|
||||
func (rc *RateController) SetAdjacentZonesFn(fn func(zoneID string) []string) {
|
||||
rc.mu.Lock()
|
||||
rc.adjacentZones = fn
|
||||
rc.mu.Unlock()
|
||||
}
|
||||
|
||||
// OnMotionState is called after each CSI frame is processed. If the node was idle
|
||||
// and motion is now detected, it ramps up immediately.
|
||||
func (rc *RateController) OnMotionState(nodeMAC string, motionDetected bool) {
|
||||
|
|
@ -75,6 +119,25 @@ func (rc *RateController) OnMotionState(nodeMAC string, motionDetected bool) {
|
|||
ns.active = true
|
||||
rc.configSender(nodeMAC, RateActive, 0) // active: disable on-device hint, server handles it
|
||||
}
|
||||
|
||||
// Zone-aware mode: update zones this node belongs to
|
||||
if rc.zoneMembership != nil {
|
||||
now := time.Now()
|
||||
zoneIDs := rc.zoneMembership(nodeMAC)
|
||||
for _, zoneID := range zoneIDs {
|
||||
zone := rc.getOrCreateZone(zoneID)
|
||||
zoneWasIdle := zone.allNodesIdle
|
||||
zone.allNodesIdle = false
|
||||
zone.lastMotionAt = now
|
||||
zone.nodes[nodeMAC] = true
|
||||
|
||||
// If zone was idle and is now active, ramp all nodes in zone to active
|
||||
// and ramp adjacent zones to sentinel (preemptive coverage)
|
||||
if zoneWasIdle {
|
||||
rc.activateZone(zoneID)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// OnMotionHint is called when the ESP32 sends a motion_hint message (on-device
|
||||
|
|
@ -97,8 +160,57 @@ func (rc *RateController) OnMotionHint(nodeMAC string) {
|
|||
// OnNodeDisconnected removes rate state for a disconnected node.
|
||||
func (rc *RateController) OnNodeDisconnected(nodeMAC string) {
|
||||
rc.mu.Lock()
|
||||
defer rc.mu.Unlock()
|
||||
|
||||
// Remove from nodes
|
||||
delete(rc.nodes, nodeMAC)
|
||||
rc.mu.Unlock()
|
||||
|
||||
// If zone-aware mode is active, remove from all zones and redesignate sentinels if needed
|
||||
if rc.zoneMembership != nil {
|
||||
zoneIDs := rc.zoneMembership(nodeMAC)
|
||||
for _, zoneID := range zoneIDs {
|
||||
if zone, ok := rc.zones[zoneID]; ok {
|
||||
delete(zone.nodes, nodeMAC)
|
||||
// If this was the sentinel, redesignate
|
||||
if zone.sentinelNodeMAC == nodeMAC {
|
||||
zone.sentinelNodeMAC = ""
|
||||
if zone.allNodesIdle && len(zone.nodes) > 0 {
|
||||
rc.designateSentinel(zoneID)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// RampZone preemptively ramps all nodes in a zone to active rate.
|
||||
// Called by the prediction engine when a zone arrival is predicted (P(arrival) > threshold).
|
||||
// Optionally ramps adjacent zones to sentinel rate for preemptive coverage.
|
||||
func (rc *RateController) RampZone(zoneID string, rampAdjacent bool) {
|
||||
rc.mu.Lock()
|
||||
defer rc.mu.Unlock()
|
||||
|
||||
zone := rc.getOrCreateZone(zoneID)
|
||||
now := time.Now()
|
||||
|
||||
// Ramp all nodes in zone to active
|
||||
for nodeMAC := range zone.nodes {
|
||||
ns := rc.getOrCreate(nodeMAC)
|
||||
ns.active = true
|
||||
ns.lastMotionAt = now
|
||||
rc.configSender(nodeMAC, RateActive, 0)
|
||||
}
|
||||
|
||||
// Update zone state
|
||||
zone.allNodesIdle = false
|
||||
zone.lastMotionAt = now
|
||||
|
||||
// Optionally ramp adjacent zones to sentinel
|
||||
if rampAdjacent && rc.adjacentZones != nil {
|
||||
for _, adjZoneID := range rc.adjacentZones(zoneID) {
|
||||
rc.rampZoneToSentinel(adjZoneID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Run starts the background goroutine that enforces idle timeouts.
|
||||
|
|
@ -123,12 +235,87 @@ func (rc *RateController) checkIdleTimeouts() {
|
|||
rc.mu.Lock()
|
||||
defer rc.mu.Unlock()
|
||||
|
||||
// First, handle per-node idle detection
|
||||
for mac, ns := range rc.nodes {
|
||||
if ns.active && now.Sub(ns.lastMotionAt) >= idleTimeout {
|
||||
ns.active = false
|
||||
rc.configSender(mac, RateIdle, DefaultVarianceThreshold) // idle: enable on-device hint
|
||||
// If zone-aware mode is active, don't send config here; let zone-level logic handle it
|
||||
if rc.zoneMembership == nil {
|
||||
rc.configSender(mac, RateIdle, DefaultVarianceThreshold)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Then, handle zone-level idle detection and sentinel designation
|
||||
if rc.zoneMembership != nil {
|
||||
allZonesIdle := true
|
||||
now := time.Now()
|
||||
|
||||
for zoneID, zone := range rc.zones {
|
||||
// Determine if all nodes in this zone are idle
|
||||
allNodesIdle := true
|
||||
oldestMotion := time.Unix(0, 0)
|
||||
|
||||
for nodeMAC := range zone.nodes {
|
||||
ns := rc.nodes[nodeMAC]
|
||||
if ns.active {
|
||||
allNodesIdle = false
|
||||
break
|
||||
}
|
||||
if ns.lastMotionAt.After(oldestMotion) {
|
||||
oldestMotion = ns.lastMotionAt
|
||||
}
|
||||
}
|
||||
|
||||
// Check if zone has been idle for timeout duration
|
||||
zoneIdle := allNodesIdle && (!oldestMotion.IsZero() && now.Sub(oldestMotion) >= idleTimeout)
|
||||
|
||||
// Zone transition: active -> idle
|
||||
if zoneIdle && !zone.allNodesIdle {
|
||||
zone.allNodesIdle = true
|
||||
zone.lastMotionAt = oldestMotion
|
||||
rc.designateSentinel(zoneID)
|
||||
} else if !zoneIdle {
|
||||
// Zone is active (or has no nodes with valid timestamps)
|
||||
allZonesIdle = false
|
||||
}
|
||||
}
|
||||
|
||||
// Update fleet idle state
|
||||
rc.fleetIdle = allZonesIdle && len(rc.zones) > 0
|
||||
}
|
||||
}
|
||||
|
||||
// designateSentinel chooses a sentinel node for an idle zone and adjusts rates.
|
||||
// The sentinel gets RateSentinel (5 Hz); all other nodes get RateFleetIdle (1 Hz).
|
||||
func (rc *RateController) designateSentinel(zoneID string) {
|
||||
zone := rc.zones[zoneID]
|
||||
if len(zone.nodes) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// Find lexicographically smallest MAC as sentinel
|
||||
var sentinelMAC string
|
||||
for nodeMAC := range zone.nodes {
|
||||
if sentinelMAC == "" || nodeMAC < sentinelMAC {
|
||||
sentinelMAC = nodeMAC
|
||||
}
|
||||
}
|
||||
zone.sentinelNodeMAC = sentinelMAC
|
||||
|
||||
// Set rates: sentinel at 5 Hz, others at 1 Hz
|
||||
for nodeMAC := range zone.nodes {
|
||||
rate := RateFleetIdle
|
||||
if nodeMAC == sentinelMAC {
|
||||
rate = RateSentinel
|
||||
}
|
||||
ns := rc.nodes[nodeMAC]
|
||||
if ns.active {
|
||||
// Shouldn't happen if zone is idle, but handle it
|
||||
ns.active = false
|
||||
}
|
||||
rc.configSender(nodeMAC, rate, DefaultVarianceThreshold)
|
||||
}
|
||||
}
|
||||
|
||||
func (rc *RateController) getOrCreate(nodeMAC string) *nodeRateState {
|
||||
|
|
@ -139,3 +326,53 @@ func (rc *RateController) getOrCreate(nodeMAC string) *nodeRateState {
|
|||
rc.nodes[nodeMAC] = ns
|
||||
return ns
|
||||
}
|
||||
|
||||
// getOrCreateZone retrieves a zone's state, creating it if necessary.
|
||||
func (rc *RateController) getOrCreateZone(zoneID string) *zoneRateState {
|
||||
if zone, ok := rc.zones[zoneID]; ok {
|
||||
return zone
|
||||
}
|
||||
zone := &zoneRateState{
|
||||
zoneID: zoneID,
|
||||
nodes: make(map[string]bool),
|
||||
allNodesIdle: true, // start idle; will be marked active on first motion
|
||||
}
|
||||
rc.zones[zoneID] = zone
|
||||
return zone
|
||||
}
|
||||
|
||||
// activateZone ramps all nodes in a zone to active and ramps adjacent zones to sentinel.
|
||||
// Called when a zone transitions from idle to active (motion detected in an idle zone).
|
||||
func (rc *RateController) activateZone(zoneID string) {
|
||||
// Ramp all nodes in this zone to active
|
||||
zone := rc.zones[zoneID]
|
||||
for nodeMAC := range zone.nodes {
|
||||
ns := rc.getOrCreate(nodeMAC)
|
||||
if !ns.active {
|
||||
ns.active = true
|
||||
ns.lastMotionAt = time.Now()
|
||||
rc.configSender(nodeMAC, RateActive, 0)
|
||||
}
|
||||
}
|
||||
|
||||
// Ramp adjacent zones to sentinel (preemptive coverage)
|
||||
if rc.adjacentZones != nil {
|
||||
for _, adjZoneID := range rc.adjacentZones(zoneID) {
|
||||
rc.rampZoneToSentinel(adjZoneID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// rampZoneToSentinel ramps all nodes in a zone to sentinel rate (5 Hz).
|
||||
// Used for preemptive coverage of zones adjacent to active zones.
|
||||
func (rc *RateController) rampZoneToSentinel(zoneID string) {
|
||||
zone := rc.getOrCreateZone(zoneID)
|
||||
for nodeMAC := range zone.nodes {
|
||||
ns := rc.getOrCreate(nodeMAC)
|
||||
// Only ramp if node is idle (no need to affect already-active nodes)
|
||||
if !ns.active {
|
||||
ns.lastMotionAt = time.Now()
|
||||
rc.configSender(nodeMAC, RateSentinel, DefaultVarianceThreshold)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -197,3 +197,357 @@ func TestNodeDisconnectClearsState(t *testing.T) {
|
|||
t.Error("node state should be removed after disconnect")
|
||||
}
|
||||
}
|
||||
|
||||
// Zone-aware tests
|
||||
|
||||
func TestZoneMembershipTracking(t *testing.T) {
|
||||
rc, sent := newTestRC()
|
||||
|
||||
// Set zone membership function
|
||||
rc.SetZoneMembershipFn(func(mac string) []string {
|
||||
if mac == "AA:BB:CC:DD:EE:01" {
|
||||
return []string{"zone-a", "zone-b"} // Node in overlapping zones
|
||||
}
|
||||
if mac == "AA:BB:CC:DD:EE:02" {
|
||||
return []string{"zone-a"}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
// Motion in zone-a node
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:01", true)
|
||||
|
||||
// Should ramp the node to active
|
||||
if len(*sent) != 1 || (*sent)[0].rate != RateActive {
|
||||
t.Errorf("expected node to ramp to active, got %d sends: %v", len(*sent), *sent)
|
||||
}
|
||||
|
||||
// Check zone state was updated
|
||||
rc.mu.Lock()
|
||||
zoneA := rc.zones["zone-a"]
|
||||
zoneB := rc.zones["zone-b"]
|
||||
rc.mu.Unlock()
|
||||
|
||||
if zoneA == nil || zoneB == nil {
|
||||
t.Fatal("zones should be created")
|
||||
}
|
||||
if !zoneA.nodes["AA:BB:CC:DD:EE:01"] || !zoneB.nodes["AA:BB:CC:DD:EE:01"] {
|
||||
t.Error("node should be registered in both zones")
|
||||
}
|
||||
if zoneA.allNodesIdle || zoneB.allNodesIdle {
|
||||
t.Error("zones should be marked active after motion")
|
||||
}
|
||||
}
|
||||
|
||||
func TestZoneIdleDetectionAndSentinelDesignation(t *testing.T) {
|
||||
rc, sent := newTestRC()
|
||||
|
||||
// Set zone membership
|
||||
rc.SetZoneMembershipFn(func(mac string) []string {
|
||||
return []string{"zone-a"}
|
||||
})
|
||||
|
||||
// Add two nodes to zone-a (both active)
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:01", true)
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:02", true)
|
||||
|
||||
// Both should be at active rate
|
||||
if len(*sent) != 2 {
|
||||
t.Fatalf("expected 2 active sends, got %d", len(*sent))
|
||||
}
|
||||
|
||||
// Force both nodes to timeout
|
||||
rc.mu.Lock()
|
||||
rc.nodes["AA:BB:CC:DD:EE:01"].lastMotionAt = time.Now().Add(-idleTimeout - time.Second)
|
||||
rc.nodes["AA:BB:CC:DD:EE:02"].lastMotionAt = time.Now().Add(-idleTimeout - time.Second)
|
||||
rc.mu.Unlock()
|
||||
|
||||
// Run idle timeout check
|
||||
rc.checkIdleTimeouts()
|
||||
|
||||
// Should have sent 2 more configs: sentinel (5 Hz) and non-sentinel (1 Hz)
|
||||
// Sentinel is lexicographically smaller MAC
|
||||
if len(*sent) != 4 {
|
||||
t.Fatalf("expected 4 sends total (2 active + 2 idle), got %d: %v", len(*sent), *sent)
|
||||
}
|
||||
|
||||
// Verify rates
|
||||
sentinelMAC := "AA:BB:CC:DD:EE:01" // Lexicographically smaller
|
||||
otherMAC := "AA:BB:CC:DD:EE:02"
|
||||
|
||||
sentinelRate, otherRate := -1, -1
|
||||
for _, s := range *sent {
|
||||
if s.mac == sentinelMAC && s.rate != RateActive { // Skip initial active send
|
||||
sentinelRate = s.rate
|
||||
}
|
||||
if s.mac == otherMAC && s.rate != RateActive {
|
||||
otherRate = s.rate
|
||||
}
|
||||
}
|
||||
|
||||
if sentinelRate != RateSentinel {
|
||||
t.Errorf("expected sentinel rate %d, got %d", RateSentinel, sentinelRate)
|
||||
}
|
||||
if otherRate != RateFleetIdle {
|
||||
t.Errorf("expected non-sentinel rate %d, got %d", RateFleetIdle, otherRate)
|
||||
}
|
||||
|
||||
// Verify zone state
|
||||
rc.mu.Lock()
|
||||
zone := rc.zones["zone-a"]
|
||||
rc.mu.Unlock()
|
||||
|
||||
if !zone.allNodesIdle {
|
||||
t.Error("zone should be marked idle after timeout")
|
||||
}
|
||||
if zone.sentinelNodeMAC != sentinelMAC {
|
||||
t.Errorf("expected sentinel %s, got %s", sentinelMAC, zone.sentinelNodeMAC)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFleetIdleDetection(t *testing.T) {
|
||||
rc, sent := newTestRC()
|
||||
|
||||
// Set zone membership and adjacent zones
|
||||
rc.SetZoneMembershipFn(func(mac string) []string {
|
||||
if mac == "AA:BB:CC:DD:EE:01" || mac == "AA:BB:CC:DD:EE:02" {
|
||||
return []string{"zone-a"}
|
||||
}
|
||||
if mac == "AA:BB:CC:DD:EE:03" {
|
||||
return []string{"zone-b"}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
rc.SetAdjacentZonesFn(func(zoneID string) []string {
|
||||
if zoneID == "zone-a" {
|
||||
return []string{"zone-b"}
|
||||
}
|
||||
if zoneID == "zone-b" {
|
||||
return []string{"zone-a"}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
// Add nodes to both zones (all active)
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:01", true)
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:02", true)
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:03", true)
|
||||
|
||||
// All should be at active rate
|
||||
if len(*sent) != 3 {
|
||||
t.Fatalf("expected 3 active sends, got %d", len(*sent))
|
||||
}
|
||||
|
||||
// Force all nodes to timeout
|
||||
rc.mu.Lock()
|
||||
for mac := range rc.nodes {
|
||||
rc.nodes[mac].lastMotionAt = time.Now().Add(-idleTimeout - time.Second)
|
||||
}
|
||||
rc.mu.Unlock()
|
||||
|
||||
// Run idle timeout check
|
||||
rc.checkIdleTimeouts()
|
||||
|
||||
// Should have sent configs for sentinel + non-sentinel in each zone
|
||||
// zone-a: 2 nodes (sentinel + non-sentinel), zone-b: 1 node (sentinel only)
|
||||
expectedSends := 3 + 2 + 1 // initial active + idle configs
|
||||
if len(*sent) != expectedSends {
|
||||
t.Fatalf("expected %d sends, got %d: %v", expectedSends, len(*sent), *sent)
|
||||
}
|
||||
|
||||
// Verify fleet is idle
|
||||
rc.mu.Lock()
|
||||
fleetIdle := rc.fleetIdle
|
||||
rc.mu.Unlock()
|
||||
|
||||
if !fleetIdle {
|
||||
t.Error("fleet should be idle when all zones are idle")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdjacentZoneRamping(t *testing.T) {
|
||||
rc, sent := newTestRC()
|
||||
|
||||
// Set zone membership and adjacent zones
|
||||
rc.SetZoneMembershipFn(func(mac string) []string {
|
||||
if mac == "AA:BB:CC:DD:EE:01" {
|
||||
return []string{"zone-a"}
|
||||
}
|
||||
if mac == "AA:BB:CC:DD:EE:02" {
|
||||
return []string{"zone-b"}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
rc.SetAdjacentZonesFn(func(zoneID string) []string {
|
||||
if zoneID == "zone-a" {
|
||||
return []string{"zone-b"}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
// Zone-b is idle (timeout) - start with idle node
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:02", true)
|
||||
rc.mu.Lock()
|
||||
rc.nodes["AA:BB:CC:DD:EE:02"].lastMotionAt = time.Now().Add(-idleTimeout - time.Second)
|
||||
rc.mu.Unlock()
|
||||
|
||||
// Run idle timeout to mark zone-b idle and designate sentinel
|
||||
rc.checkIdleTimeouts()
|
||||
|
||||
// Reset sent tracker to count only zone-a activation sends
|
||||
initialSends := len(*sent)
|
||||
*sent = (*sent)[:initialSends]
|
||||
|
||||
// Zone-a detects motion (should ramp adjacent zone-b to sentinel)
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:01", true)
|
||||
|
||||
// Should have: zone-a node active + zone-b node ramped to sentinel
|
||||
// Note: zone-b node is already at sentinel (5 Hz), but we ramp it again
|
||||
// This is expected behavior for preemptive coverage
|
||||
if len(*sent) != initialSends+2 {
|
||||
t.Fatalf("expected 2 sends (zone-a active + zone-b sentinel), got %d: %v", len(*sent)-initialSends, (*sent)[initialSends:])
|
||||
}
|
||||
|
||||
// Verify zone-a went to active and zone-b went to sentinel
|
||||
zoneAActive := false
|
||||
zoneBSentinel := false
|
||||
for i := initialSends; i < len(*sent); i++ {
|
||||
if (*sent)[i].mac == "AA:BB:CC:DD:EE:01" && (*sent)[i].rate == RateActive {
|
||||
zoneAActive = true
|
||||
}
|
||||
if (*sent)[i].mac == "AA:BB:CC:DD:EE:02" && (*sent)[i].rate == RateSentinel {
|
||||
zoneBSentinel = true
|
||||
}
|
||||
}
|
||||
if !zoneAActive {
|
||||
t.Error("zone-a node should be ramped to active rate")
|
||||
}
|
||||
if !zoneBSentinel {
|
||||
t.Error("adjacent zone-b node should be ramped to sentinel rate")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRampZonePredictionEngine(t *testing.T) {
|
||||
rc, sent := newTestRC()
|
||||
|
||||
// Set zone membership
|
||||
rc.SetZoneMembershipFn(func(mac string) []string {
|
||||
if mac == "AA:BB:CC:DD:EE:01" || mac == "AA:BB:CC:DD:EE:02" {
|
||||
return []string{"zone-a"}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
// Add nodes to zone-a (they'll be idle by default)
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:01", true)
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:02", true)
|
||||
|
||||
// Force idle timeout
|
||||
rc.mu.Lock()
|
||||
for mac := range rc.nodes {
|
||||
rc.nodes[mac].lastMotionAt = time.Now().Add(-idleTimeout - time.Second)
|
||||
}
|
||||
rc.mu.Unlock()
|
||||
rc.checkIdleTimeouts()
|
||||
|
||||
// Reset sent tracker to count only RampZone sends
|
||||
initialSends := len(*sent)
|
||||
*sent = (*sent)[:initialSends]
|
||||
|
||||
// Prediction engine ramps zone-a (with adjacent zone ramping disabled)
|
||||
rc.RampZone("zone-a", false)
|
||||
|
||||
// Should have ramped both nodes to active
|
||||
if len(*sent) != initialSends+2 {
|
||||
t.Fatalf("expected 2 ramp sends, got %d: %v", len(*sent)-initialSends, (*sent)[initialSends:])
|
||||
}
|
||||
|
||||
for i := initialSends; i < len(*sent); i++ {
|
||||
if (*sent)[i].rate != RateActive {
|
||||
t.Errorf("RampZone should set active rate, got %d", (*sent)[i].rate)
|
||||
}
|
||||
}
|
||||
|
||||
// Verify zone state updated
|
||||
rc.mu.Lock()
|
||||
zone := rc.zones["zone-a"]
|
||||
rc.mu.Unlock()
|
||||
|
||||
if zone.allNodesIdle {
|
||||
t.Error("zone should be marked active after RampZone")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackwardCompatibilityNoZones(t *testing.T) {
|
||||
rc, sent := newTestRC()
|
||||
|
||||
// No zone membership function set - should fall back to per-node behavior
|
||||
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:FF", true)
|
||||
|
||||
if len(*sent) != 1 || (*sent)[0].rate != RateActive {
|
||||
t.Errorf("should work without zones (per-node fallback), got %d sends: %v", len(*sent), *sent)
|
||||
}
|
||||
|
||||
// Force timeout
|
||||
rc.mu.Lock()
|
||||
rc.nodes["AA:BB:CC:DD:EE:FF"].lastMotionAt = time.Now().Add(-idleTimeout - time.Second)
|
||||
rc.mu.Unlock()
|
||||
|
||||
rc.checkIdleTimeouts()
|
||||
|
||||
// Should drop to idle rate (2 Hz)
|
||||
if len(*sent) != 2 || (*sent)[1].rate != RateIdle {
|
||||
t.Errorf("should drop to RateIdle (2 Hz) without zones, got %d sends: %v", len(*sent), *sent)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNodeDisconnectWithSentinelRedesignation(t *testing.T) {
|
||||
rc, sent := newTestRC()
|
||||
|
||||
// Set zone membership
|
||||
rc.SetZoneMembershipFn(func(mac string) []string {
|
||||
return []string{"zone-a"}
|
||||
})
|
||||
|
||||
// Add two nodes, let them go idle
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:01", true)
|
||||
rc.OnMotionState("AA:BB:CC:DD:EE:02", true)
|
||||
|
||||
// Force timeout
|
||||
rc.mu.Lock()
|
||||
for mac := range rc.nodes {
|
||||
rc.nodes[mac].lastMotionAt = time.Now().Add(-idleTimeout - time.Second)
|
||||
}
|
||||
rc.mu.Unlock()
|
||||
rc.checkIdleTimeouts()
|
||||
|
||||
// Sentinel should be 01 (lexicographically smaller)
|
||||
rc.mu.Lock()
|
||||
sentinelMAC := rc.zones["zone-a"].sentinelNodeMAC
|
||||
rc.mu.Unlock()
|
||||
|
||||
if sentinelMAC != "AA:BB:CC:DD:EE:01" {
|
||||
t.Errorf("expected sentinel 01, got %s", sentinelMAC)
|
||||
}
|
||||
|
||||
// Disconnect sentinel
|
||||
rc.OnNodeDisconnected("AA:BB:CC:DD:EE:01")
|
||||
|
||||
// New sentinel should be redesignated (02)
|
||||
rc.mu.Lock()
|
||||
newSentinelMAC := rc.zones["zone-a"].sentinelNodeMAC
|
||||
rc.mu.Unlock()
|
||||
|
||||
if newSentinelMAC != "AA:BB:CC:DD:EE:02" {
|
||||
t.Errorf("expected new sentinel 02, got %s", newSentinelMAC)
|
||||
}
|
||||
|
||||
// Should have sent config for new sentinel
|
||||
lastSend := (*sent)[len(*sent)-1]
|
||||
if lastSend.mac != "AA:BB:CC:DD:EE:02" || lastSend.rate != RateSentinel {
|
||||
t.Errorf("expected new sentinel config, got %d Hz for %s", lastSend.rate, lastSend.mac)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
208
mothership/internal/ingestion/ratecontrol_zones_design.md
Normal file
208
mothership/internal/ingestion/ratecontrol_zones_design.md
Normal file
|
|
@ -0,0 +1,208 @@
|
|||
# Zone-Aware Rate Control Design
|
||||
|
||||
## Overview
|
||||
|
||||
Extend the current per-node `RateController` to support fleet-level coordination with zone awareness and sentinel link designation.
|
||||
|
||||
## Current State
|
||||
|
||||
**RateController** (mothership/internal/ingestion/ratecontrol.go):
|
||||
- Manages per-node adaptive rates: 50 Hz active, 2 Hz idle
|
||||
- Per-node state machine: active/idle with 30-second timeout
|
||||
- OnMotionHint ramps adjacent nodes (already has adjacency function)
|
||||
- No zone awareness, no fleet-level coordination
|
||||
|
||||
## New Requirements (Component 24)
|
||||
|
||||
1. **Zone-level idle detection**: When all nodes in a zone have been idle, designate one sentinel link per zone at 5 Hz; all others drop to 1 Hz
|
||||
2. **Fleet-wide idle state**: When all zones are idle, only sentinel links at 5 Hz run
|
||||
3. **Adjacent zone ramping**: When activity detected in one zone, ramp that zone to full rate AND adjacent zones to 5 Hz
|
||||
4. **Prediction engine interface**: RampZone(zoneID) for preemptive zone ramping
|
||||
|
||||
## Design
|
||||
|
||||
### Rate Tiers
|
||||
|
||||
```go
|
||||
const (
|
||||
// RateActive is the CSI sampling rate (Hz) when motion is detected.
|
||||
RateActive = 50
|
||||
|
||||
// RateIdle is the CSI sampling rate (Hz) when a node is idle but fleet is not fully idle.
|
||||
RateIdle = 2
|
||||
|
||||
// RateSentinel is the CSI sampling rate (Hz) for the designated sentinel link in an idle zone.
|
||||
// Runs at 5 Hz to maintain minimal coverage while reducing bandwidth.
|
||||
RateSentinel = 5
|
||||
|
||||
// RateFleetIdle is the CSI sampling rate (Hz) for non-sentinel links when the entire fleet is idle.
|
||||
// Only one sentinel per zone runs at RateSentinel; all others run at 1 Hz.
|
||||
RateFleetIdle = 1
|
||||
)
|
||||
```
|
||||
|
||||
### Data Structures
|
||||
|
||||
```go
|
||||
// zoneRateState tracks the adaptive rate state for a single zone.
|
||||
type zoneRateState struct {
|
||||
zoneID string
|
||||
nodes map[string]bool // nodes in this zone
|
||||
sentinelNodeMAC string // designated sentinel node (may be empty)
|
||||
allNodesIdle bool // true when all nodes in zone are idle
|
||||
lastMotionAt time.Time // last motion in this zone
|
||||
}
|
||||
|
||||
// RateController manages per-node and zone-aware adaptive sensing rates.
|
||||
type RateController struct {
|
||||
// Existing fields
|
||||
mu sync.Mutex
|
||||
nodes map[string]*nodeRateState
|
||||
configSender func(nodeMAC string, rateHz int, varianceThreshold float64)
|
||||
adjacentNodes func(nodeMAC string) []string
|
||||
|
||||
// New zone-aware fields
|
||||
zones map[string]*zoneRateState // keyed by zone ID
|
||||
zoneMembership func(nodeMAC string) []string // returns zone IDs for a node
|
||||
adjacentZones func(zoneID string) []string // returns zone IDs adjacent to a zone
|
||||
fleetIdle bool // true when all zones are idle
|
||||
}
|
||||
```
|
||||
|
||||
### Zone Membership Algorithm
|
||||
|
||||
**Determining which zones a node belongs to:**
|
||||
- A node is in a zone if its position (X, Y, Z) is within the zone's bounds
|
||||
- Zone bounds: (min_x <= node.X <= max_x) && (min_y <= node.Y <= max_y) && (min_z <= node.Z <= max_z)
|
||||
- A node can be in multiple zones (overlapping zones)
|
||||
- Use zone membership provider function from zone manager
|
||||
|
||||
**Zone membership provider signature:**
|
||||
```go
|
||||
// ZoneMembershipFn returns the zone IDs that a node belongs to based on its position.
|
||||
// The caller must provide this function; it typically queries the zone manager.
|
||||
type ZoneMembershipFn func(nodeMAC string) []string
|
||||
```
|
||||
|
||||
### Sentinel Designation Algorithm
|
||||
|
||||
**Choosing the sentinel node for a zone:**
|
||||
- When a zone becomes idle, designate one node as the sentinel
|
||||
- Selection criteria: the node with the lexicographically smallest MAC address (deterministic)
|
||||
- Sentinel gets RateSentinel (5 Hz), all other nodes in zone get RateFleetIdle (1 Hz)
|
||||
- Redesignate only when previous sentinel disconnects
|
||||
|
||||
**Zone idle detection:**
|
||||
- Zone is idle when all its nodes have been idle for the timeout period
|
||||
- Track last motion time per zone (max of last motion times of all nodes in zone)
|
||||
- When zone transitions to idle, designate sentinel and adjust rates
|
||||
|
||||
### Fleet Idle Detection
|
||||
|
||||
**Fleet is idle when:**
|
||||
- All zones are in the idle state (allNodesIdle = true)
|
||||
- OR no zones exist (fallback to per-node behavior)
|
||||
|
||||
**When fleet becomes idle:**
|
||||
- All zones: sentinel runs at 5 Hz, others at 1 Hz
|
||||
- When fleet was idle, one zone becomes active: that zone at 50 Hz, adjacent zones at 5 Hz
|
||||
|
||||
### Adjacent Zone Ramping
|
||||
|
||||
**When motion is detected in a zone:**
|
||||
- Ramp that zone's nodes to RateActive (50 Hz)
|
||||
- Ramp adjacent zones' nodes to RateSentinel (5 Hz) - preemptive coverage
|
||||
- Adjacent zones determined by portals in zone manager
|
||||
|
||||
**Adjacent zones provider signature:**
|
||||
```go
|
||||
// AdjacentZonesFn returns zone IDs that are adjacent to a given zone.
|
||||
// The caller must provide this function; it typically queries the zone manager for portals.
|
||||
type AdjacentZonesFn func(zoneID string) []string
|
||||
```
|
||||
|
||||
### Prediction Engine Interface
|
||||
|
||||
**RampZone(zoneID) method:**
|
||||
- Called by prediction engine when P(arrival in zone) > threshold
|
||||
- Ramps all nodes in the zone to RateActive (50 Hz)
|
||||
- Resets zone's lastMotionAt to now
|
||||
- Optionally ramps adjacent zones to RateSentinel (5 Hz)
|
||||
|
||||
```go
|
||||
// RampZone preemptively ramps all nodes in a zone to active rate.
|
||||
// Called by the prediction engine when a zone arrival is predicted.
|
||||
func (rc *RateController) RampZone(zoneID string) {
|
||||
// Set all nodes in zone to active
|
||||
// Update zone state
|
||||
// Optionally ramp adjacent zones to sentinel
|
||||
}
|
||||
```
|
||||
|
||||
### State Machine
|
||||
|
||||
**Node state:**
|
||||
- Active (50 Hz) → Idle (2 Hz) after 30s timeout
|
||||
- Idle → Active immediately on motion detection
|
||||
- Idle (2 Hz) → Fleet Idle (1 Hz) when fleet becomes idle AND not sentinel
|
||||
- Idle (2 Hz) → Sentinel (5 Hz) when fleet becomes idle AND is sentinel
|
||||
|
||||
**Zone state:**
|
||||
- All nodes idle → Zone idle
|
||||
- Zone idle → Zone active when any node detects motion
|
||||
- Zone active → Zone idle after timeout (same 30s)
|
||||
|
||||
**Fleet state:**
|
||||
- All zones idle → Fleet idle
|
||||
- Fleet idle → Fleet active when any zone becomes active
|
||||
|
||||
## Integration Points
|
||||
|
||||
### Required from callers:
|
||||
|
||||
1. **Node position provider**: Function to get a node's (X, Y, Z) position
|
||||
- Already available from fusion.Engine.NodePositions()
|
||||
- Used to determine zone membership
|
||||
|
||||
2. **Zone membership provider**: Function to get zone IDs for a node
|
||||
- Queries zone manager for zones containing the node's position
|
||||
- Returns []string (may be empty for nodes in no zones)
|
||||
|
||||
3. **Adjacent zones provider**: Function to get adjacent zone IDs
|
||||
- Queries zone manager for portals connected to a zone
|
||||
- Returns []string (may be empty for isolated zones)
|
||||
|
||||
4. **Zone bounds provider**: Function to get zone bounds
|
||||
- Queries zone manager for zone (min_x, min_y, min_z, max_x, max_y, max_z)
|
||||
- Used for spatial containment test
|
||||
|
||||
## Implementation Plan
|
||||
|
||||
1. Add new rate constants (RateSentinel, RateFleetIdle)
|
||||
2. Add zoneRateState struct
|
||||
3. Add zone-related fields to RateController
|
||||
4. Implement SetZoneMembershipFn() and SetAdjacentZonesFn()
|
||||
5. Implement zone membership tracking in OnMotionState()
|
||||
6. Implement zone idle detection in checkIdleTimeouts()
|
||||
7. Implement sentinel designation logic
|
||||
8. Implement fleet idle detection
|
||||
9. Implement RampZone() method
|
||||
10. Implement adjacent zone ramping in OnMotionState()
|
||||
11. Update tests to cover zone-aware behavior
|
||||
|
||||
## Backward Compatibility
|
||||
|
||||
- If no zone membership function is set, fall back to per-node behavior only
|
||||
- Existing tests should continue to pass
|
||||
- Zone-aware features are opt-in via SetZoneMembershipFn()
|
||||
|
||||
## Testing Strategy
|
||||
|
||||
1. Test zone membership determination
|
||||
2. Test zone idle detection
|
||||
3. Test sentinel designation
|
||||
4. Test fleet idle detection
|
||||
5. Test adjacent zone ramping
|
||||
6. Test RampZone() prediction engine interface
|
||||
7. Test fallback to per-node behavior when zones not configured
|
||||
8. Test node disconnection handles sentinel redesignation
|
||||
Loading…
Add table
Reference in a new issue