@@ -1414,3 +1414,177 @@ func TestEmitOldestExecutorHeartbeatLag(t *testing.T) {
14141414 })
14151415 }
14161416}
1417+
1418+ func TestRunRebalanceTriggeringLoop (t * testing.T ) {
1419+ t .Run ("no events from subscribe, trigger from ticker" , func (t * testing.T ) {
1420+ mocks := setupProcessorTest (t , config .NamespaceTypeFixed )
1421+ defer mocks .ctrl .Finish ()
1422+ processor := mocks .factory .CreateProcessor (mocks .cfg , mocks .store , mocks .election ).(* namespaceProcessor )
1423+
1424+ ctx , cancel := context .WithCancel (context .Background ())
1425+ defer cancel ()
1426+
1427+ updateChan := make (chan int64 )
1428+ triggerChan := make (chan string , 1 )
1429+
1430+ go processor .rebalanceTriggeringLoop (ctx , updateChan , triggerChan )
1431+
1432+ // Wait for ticker to be created
1433+ mocks .timeSource .BlockUntil (1 )
1434+
1435+ // Advance time to trigger the ticker
1436+ mocks .timeSource .Advance (processor .cfg .Period )
1437+
1438+ // Expect trigger from periodic reconciliation
1439+ select {
1440+ case reason := <- triggerChan :
1441+ assert .Equal (t , "Periodic reconciliation triggered" , reason )
1442+ case <- time .After (time .Second ):
1443+ t .Fatal ("expected trigger from ticker, but timed out" )
1444+ }
1445+
1446+ cancel ()
1447+ })
1448+
1449+ t .Run ("events from subscribe before period, trigger from state change" , func (t * testing.T ) {
1450+ mocks := setupProcessorTest (t , config .NamespaceTypeFixed )
1451+ defer mocks .ctrl .Finish ()
1452+ processor := mocks .factory .CreateProcessor (mocks .cfg , mocks .store , mocks .election ).(* namespaceProcessor )
1453+ processor .lastAppliedRevision = 0
1454+
1455+ ctx , cancel := context .WithCancel (context .Background ())
1456+ defer cancel ()
1457+
1458+ updateChan := make (chan int64 , 1 )
1459+ triggerChan := make (chan string , 1 )
1460+
1461+ go processor .rebalanceTriggeringLoop (ctx , updateChan , triggerChan )
1462+
1463+ // Wait for ticker to be created
1464+ mocks .timeSource .BlockUntil (1 )
1465+
1466+ // Send a state change event before the ticker fires
1467+ updateChan <- 1
1468+
1469+ // Expect trigger from state change
1470+ select {
1471+ case reason := <- triggerChan :
1472+ assert .Equal (t , "State change detected" , reason )
1473+ case <- time .After (time .Second ):
1474+ t .Fatal ("expected trigger from state change, but timed out" )
1475+ }
1476+
1477+ cancel ()
1478+ })
1479+
1480+ t .Run ("triggerChan full, multiple subscribe events, loop not stuck" , func (t * testing.T ) {
1481+ mocks := setupProcessorTest (t , config .NamespaceTypeFixed )
1482+ defer mocks .ctrl .Finish ()
1483+ processor := mocks .factory .CreateProcessor (mocks .cfg , mocks .store , mocks .election ).(* namespaceProcessor )
1484+ processor .lastAppliedRevision = 0
1485+
1486+ ctx , cancel := context .WithCancel (context .Background ())
1487+ defer cancel ()
1488+
1489+ // Use unbuffered channel for updates to ensure they are processed one at a time
1490+ updateChan := make (chan int64 )
1491+ triggerChan := make (chan string , 1 )
1492+
1493+ go processor .rebalanceTriggeringLoop (ctx , updateChan , triggerChan )
1494+
1495+ // Wait for ticker to be created
1496+ mocks .timeSource .BlockUntil (1 )
1497+
1498+ // Don't read from triggerChan yet to keep it full
1499+ // Send multiple state change events
1500+ for i := int64 (0 ); i <= 10 ; i ++ {
1501+ select {
1502+ case updateChan <- i :
1503+ case <- time .After (time .Second ):
1504+ // Expect that the loop is not stuck
1505+ t .Fatalf ("failed to send update %d, channel blocked" , i )
1506+ }
1507+ }
1508+
1509+ // Expect trigger from state change
1510+ select {
1511+ case reason := <- triggerChan :
1512+ assert .Equal (t , "State change detected" , reason )
1513+ case <- time .After (time .Second ):
1514+ t .Fatal ("expected trigger from state change, but timed out" )
1515+ }
1516+
1517+ cancel ()
1518+ })
1519+
1520+ t .Run ("stale revision ignored" , func (t * testing.T ) {
1521+ mocks := setupProcessorTest (t , config .NamespaceTypeFixed )
1522+ defer mocks .ctrl .Finish ()
1523+ processor := mocks .factory .CreateProcessor (mocks .cfg , mocks .store , mocks .election ).(* namespaceProcessor )
1524+ processor .lastAppliedRevision = 5
1525+
1526+ ctx , cancel := context .WithCancel (context .Background ())
1527+ defer cancel ()
1528+
1529+ updateChan := make (chan int64 , 1 )
1530+ triggerChan := make (chan string , 1 )
1531+
1532+ go processor .rebalanceTriggeringLoop (ctx , updateChan , triggerChan )
1533+
1534+ // Wait for ticker to be created
1535+ mocks .timeSource .BlockUntil (1 )
1536+
1537+ // Send stale revision (less than or equal to lastAppliedRevision)
1538+ updateChan <- 3
1539+
1540+ // Should not trigger - verify by advancing ticker and getting that trigger instead
1541+ mocks .timeSource .Advance (processor .cfg .Period )
1542+
1543+ select {
1544+ case reason := <- triggerChan :
1545+ assert .Equal (t , "Periodic reconciliation triggered" , reason )
1546+ case <- time .After (time .Second ):
1547+ t .Fatal ("expected trigger from ticker" )
1548+ }
1549+
1550+ cancel ()
1551+ })
1552+
1553+ t .Run ("update channel closed stops loop" , func (t * testing.T ) {
1554+ mocks := setupProcessorTest (t , config .NamespaceTypeFixed )
1555+ defer mocks .ctrl .Finish ()
1556+ processor := mocks .factory .CreateProcessor (mocks .cfg , mocks .store , mocks .election ).(* namespaceProcessor )
1557+
1558+ ctx := context .Background ()
1559+
1560+ updateChan := make (chan int64 )
1561+ triggerChan := make (chan string , 1 )
1562+
1563+ var wg sync.WaitGroup
1564+ wg .Add (1 )
1565+ go func () {
1566+ defer wg .Done ()
1567+ processor .rebalanceTriggeringLoop (ctx , updateChan , triggerChan )
1568+ }()
1569+
1570+ // Wait for ticker to be created
1571+ mocks .timeSource .BlockUntil (1 )
1572+
1573+ // Close update channel
1574+ close (updateChan )
1575+
1576+ // Wait for loop to exit
1577+ done := make (chan struct {})
1578+ go func () {
1579+ wg .Wait ()
1580+ close (done )
1581+ }()
1582+
1583+ select {
1584+ case <- done :
1585+ // Loop exited as expected
1586+ case <- time .After (time .Second ):
1587+ t .Fatal ("loop did not exit after updateChan closed" )
1588+ }
1589+ })
1590+ }
0 commit comments