rdp-proxy/agents/rap-node-agent/internal/mesh/fabric_channel_router_test.go

package mesh

import (
	"testing"
	"time"
)

func TestFabricChannelRouterOpensOnBestRoute(t *testing.T) {
	router := NewFabricChannelRouter(FabricChannelRouterConfig{})
	now := time.Now()
	channel, event, err := router.OpenChannel(testFabricChannelSpec(FabricChannelTargetNode, "node-b"), FabricRouteSet{
		TargetKind: FabricChannelTargetNode,
		TargetID:   "node-b",
		Primary:    testFabricRoute("route-slow", "node-b", 80, 100, 0, true),
		WarmStandby: []FabricRoute{
			testFabricRoute("route-fast", "node-b", 15, 100, 0, true),
		},
	}, now)
	if err != nil {
		t.Fatalf("open channel: %v", err)
	}
	if channel.RouteID != "route-fast" || channel.State != FabricChannelOpen {
		t.Fatalf("channel = %+v, want route-fast open", channel)
	}
	if event.Type != FabricChannelRouteEventOpened || event.NextRoute.RouteID != "route-fast" {
		t.Fatalf("event = %+v", event)
	}
}

func TestFabricChannelRouterReroutesOnSlowAck(t *testing.T) {
	router := NewFabricChannelRouter(FabricChannelRouterConfig{MaxAckLatencyMs: 30})
	now := time.Now()
	routeSet := FabricRouteSet{
		TargetKind: FabricChannelTargetNode,
		TargetID:   "node-b",
		Primary:    testFabricRoute("route-primary", "node-b", 10, 100, 0, true),
		WarmStandby: []FabricRoute{
			testFabricRoute("route-standby", "node-b", 20, 100, 0, true),
		},
	}
	channel := FabricChannel{
		Spec:     testFabricChannelSpec(FabricChannelTargetNode, "node-b"),
		State:    FabricChannelOpen,
		RouteID:  "route-primary",
		OpenedAt: now.Add(-time.Minute),
	}
	updated, event, err := router.ObserveChannel(channel, routeSet, FabricChannelObservation{
		ChannelID:    channel.Spec.ChannelID,
		RouteID:      channel.RouteID,
		AckLatencyMs: 120,
		BytesSent:    4096,
		FramesSent:   4,
	}, now)
	if err != nil {
		t.Fatalf("observe channel: %v", err)
	}
	if event.Type != FabricChannelRouteEventReroute || event.Reason != "ack_latency_threshold" {
		t.Fatalf("event = %+v", event)
	}
	if updated.RouteID != "route-standby" || updated.RerouteCount != 1 || updated.BytesSent != 4096 || updated.FramesSent != 4 {
		t.Fatalf("updated = %+v", updated)
	}
}

func TestFabricChannelRouterReroutesPoolTargetOnFailure(t *testing.T) {
	router := NewFabricChannelRouter(FabricChannelRouterConfig{})
	now := time.Now()
	routeSet := FabricRouteSet{
		TargetKind: FabricChannelTargetPool,
		TargetID:   "pool-egress",
		Primary:    testFabricPoolRoute("route-node-b", "node-b", 10, true),
		WarmStandby: []FabricRoute{
			testFabricPoolRoute("route-node-c", "node-c", 20, true),
		},
	}
	channel := FabricChannel{
		Spec:       testFabricChannelSpec(FabricChannelTargetPool, "pool-egress"),
		State:      FabricChannelOpen,
		RouteID:    "route-node-b",
		TargetNode: "node-b",
		OpenedAt:   now.Add(-time.Minute),
	}
	updated, event, err := router.ObserveChannel(channel, routeSet, FabricChannelObservation{
		ChannelID: channel.Spec.ChannelID,
		RouteID:   channel.RouteID,
		Failed:    true,
		Reason:    "target_failed",
	}, now)
	if err != nil {
		t.Fatalf("observe channel: %v", err)
	}
	if event.Type != FabricChannelRouteEventReroute || event.PreviousRoute.RouteID != "route-node-b" || event.NextRoute.RouteID != "route-node-c" {
		t.Fatalf("event = %+v", event)
	}
	if updated.TargetNode != "node-c" || updated.RouteID != "route-node-c" {
		t.Fatalf("updated = %+v", updated)
	}
}

func TestFabricChannelRouterSuppressesRerouteInsideHysteresis(t *testing.T) {
	router := NewFabricChannelRouter(FabricChannelRouterConfig{MaxAckLatencyMs: 30, MinRerouteInterval: time.Minute})
	now := time.Now()
	channel := FabricChannel{
		Spec:        testFabricChannelSpec(FabricChannelTargetNode, "node-b"),
		State:       FabricChannelOpen,
		RouteID:     "route-primary",
		LastReroute: now.Add(-10 * time.Second),
	}
	updated, event, err := router.ObserveChannel(channel, FabricRouteSet{
		TargetKind:  FabricChannelTargetNode,
		TargetID:    "node-b",
		Primary:     testFabricRoute("route-primary", "node-b", 10, 100, 0, true),
		WarmStandby: []FabricRoute{testFabricRoute("route-standby", "node-b", 20, 100, 0, true)},
	}, FabricChannelObservation{AckLatencyMs: 120}, now)
	if err != nil {
		t.Fatalf("observe channel: %v", err)
	}
	if event.Type != FabricChannelRouteEventNone || updated.RouteID != "route-primary" {
		t.Fatalf("event=%+v updated=%+v", event, updated)
	}
}

func testFabricChannelSpec(kind FabricChannelTargetKind, targetID string) FabricChannelSpec {
	return FabricChannelSpec{
		ChannelID:    "channel-1",
		ClusterID:    "cluster-1",
		SourceNodeID: "node-a",
		TargetKind:   kind,
		TargetID:     targetID,
	}
}

func testFabricRoute(routeID string, destination string, latency int, capacity int, active int, healthy bool) FabricRoute {
	return FabricRoute{
		RouteID:           routeID,
		ClusterID:         "cluster-1",
		SourceNodeID:      "node-a",
		DestinationNodeID: destination,
		Hops:              []FabricRouteHop{{NodeID: "node-a"}, {NodeID: destination}},
		BaseLatencyMs:     latency,
		Capacity:          capacity,
		ActiveChannels:    active,
		Healthy:           healthy,
	}
}

func testFabricPoolRoute(routeID string, destination string, latency int, healthy bool) FabricRoute {
	route := testFabricRoute(routeID, destination, latency, 100, 0, healthy)
	route.PoolID = "pool-egress"
	return route
}