fix orchestrator data-plan startSeqNum (#4745)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- **New Features**
- Introduced a new field, `StartingOrchestratorSeqNum`, in the handshake
response to enhance information during node registration.
- **Bug Fixes**
- Improved assertions in testing to ensure correct handling of sequence
numbers for both new and reconnecting nodes during handshakes.
- **Documentation**
- Updated comments and logging statements for clarity and consistency
regarding the handshake process.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: wdbaruni

pkg/models/messages/node.go

@@ -15,9 +15,10 @@ type HandshakeRequest struct {
 
 // HandshakeResponse is sent in response to handshake requests
 type HandshakeResponse struct {
-	Accepted          bool   `json:"accepted"`
-	Reason            string `json:"reason,omitempty"`
-	LastComputeSeqNum uint64 `json:"LastComputeSeqNum"` // Last seq received from compute node
+	Accepted                   bool   `json:"accepted"`
+	Reason                     string `json:"reason,omitempty"`
+	LastComputeSeqNum          uint64 `json:"LastComputeSeqNum"`      // Last seq received from compute node
+	StartingOrchestratorSeqNum uint64 `json:"LastOrchestratorSeqNum"` // Seq to start sending to compute node
 }
 
 type HeartbeatRequest struct {

pkg/orchestrator/nodes/manager.go

@@ -492,9 +492,10 @@ func (n *nodesManager) Handshake(
 		reason = "node reconnected"
 	}
 	return messages.HandshakeResponse{
-		Accepted:          true,
-		Reason:            reason,
-		LastComputeSeqNum: state.ConnectionState.LastComputeSeqNum,
+		Accepted:                   true,
+		Reason:                     reason,
+		LastComputeSeqNum:          state.ConnectionState.LastComputeSeqNum,
+		StartingOrchestratorSeqNum: state.ConnectionState.LastOrchestratorSeqNum,
 	}, nil
 }
 

pkg/orchestrator/nodes/manager_test.go

@@ -148,7 +148,6 @@ func (s *NodeManagerTestSuite) TestHeartbeatMaintainsConnection() {
 }
 
 // Edge Cases and Error Scenarios
-
 func (s *NodeManagerTestSuite) TestHandshakeSequenceNumberLogic() {
 	// Test initial handshake with new node
 	nodeInfo := s.createNodeInfo("new-node")
@@ -181,6 +180,8 @@ func (s *NodeManagerTestSuite) TestHandshakeSequenceNumberLogic() {
 	s.Require().NoError(err)
 	s.Assert().Equal(latestSeqNum, state.ConnectionState.LastOrchestratorSeqNum,
 		"New node should be assigned latest sequence number")
+	s.Assert().Equal(latestSeqNum, resp1.StartingOrchestratorSeqNum,
+		"New node should receive latest sequence number as starting point")
 
 	// Update sequence numbers via heartbeat
 	updatedOrchSeqNum := uint64(200)
@@ -210,6 +211,8 @@ func (s *NodeManagerTestSuite) TestHandshakeSequenceNumberLogic() {
 	s.Require().NoError(err)
 	s.Require().True(resp2.Accepted)
 	s.Assert().Contains(resp2.Reason, "reconnected")
+	s.Assert().Equal(updatedOrchSeqNum, resp2.StartingOrchestratorSeqNum,
+		"Reconnecting node should receive its last known sequence number")
 
 	// Verify sequence numbers were preserved from previous state
 	state, err = s.manager.Get(ctx, nodeInfo.ID())
@@ -235,6 +238,8 @@ func (s *NodeManagerTestSuite) TestHandshakeSequenceNumberEdgeCases() {
 	s.Require().NoError(err)
 	s.Assert().Equal(uint64(0), state1.ConnectionState.LastOrchestratorSeqNum,
 		"New node should get zero sequence when event store is empty")
+	s.Assert().Equal(uint64(0), resp1.StartingOrchestratorSeqNum,
+		"New node should receive zero as starting sequence when event store is empty")
 
 	// Test concurrent handshakes with sequence numbers
 	var wg sync.WaitGroup
@@ -271,6 +276,8 @@ func (s *NodeManagerTestSuite) TestHandshakeSequenceNumberEdgeCases() {
 			s.Require().NoError(err)
 			s.Assert().Equal(latestSeqNum, state.ConnectionState.LastOrchestratorSeqNum,
 				"Concurrent new nodes should all get latest sequence number")
+			s.Assert().Equal(latestSeqNum, resp.StartingOrchestratorSeqNum,
+				"Concurrent new nodes should receive latest sequence number as starting point")
 		}(i)
 	}
 

pkg/transport/nclprotocol/orchestrator/manager.go

@@ -186,7 +186,7 @@ func (cm *ComputeManager) handleHandshakeRequest(ctx context.Context, msg *envel
 	}
 
 	// Create data plane for accepted node
-	if err = cm.setupDataPlane(ctx, request.NodeInfo, request.LastOrchestratorSeqNum); err != nil {
+	if err = cm.setupDataPlane(ctx, request.NodeInfo, response.StartingOrchestratorSeqNum); err != nil {
 		return nil, fmt.Errorf("setup data plane failed: %w", err)
 	}