sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-07-19 23:11:29 +02:00
Code Issues Packages Projects Releases Wiki Activity

Allows to specify max batch for tests

Browse Source
This commit is contained in:
riccardom
2026-06-26 16:09:26 +02:00
parent 29dde908ae
commit 74bb5c613e
2 changed files with 69 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
client/internal/engine.go
View File

@@ -762,14 +762,15 @@ func (e *Engine) blockLanAccess() {
// modifyPeers updates peers that have been modified (e.g. IP address has been changed).
// It closes the existing connection, removes it from the peerConns map, and creates a new one.
// maxPeersPerSyncPass bounds how many peers each of removePeers/modifyPeers/
// addNewPeers applies per sync pass, so syncMsgMux is held only for a batch at a
// time and the signal handler can interleave between passes.
// maxPeersPerSyncPass is the default per-pass cap on how many peers each of
// removePeers/modifyPeers/addNewPeers applies, so syncMsgMux is held only for a
// batch at a time and other subsystems can interleave between passes. It is
// passed in (not read globally) so tests can exercise the multi-pass path.
const maxPeersPerSyncPass = 300
// modifyPeers re-applies up to maxPeersPerSyncPass changed peers per call. It
// returns true when more changed peers remained than the cap, so the caller re-runs.
func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig) (bool, error) {
// modifyPeers re-applies up to maxBatch changed peers per call. It returns true
// when more changed peers remained than the cap, so the caller re-runs.
func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig, maxBatch int) (bool, error) {
// first, check if peers have been modified
var modified []*mgmProto.RemotePeerConfig
@@ -800,8 +801,8 @@ func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig) (bool, er
}
more := false
if len(modified) > maxPeersPerSyncPass {
modified = modified[:maxPeersPerSyncPass]
if len(modified) > maxBatch {
modified = modified[:maxBatch]
more = true
}
@@ -822,9 +823,9 @@ func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig) (bool, er
// removePeers finds and removes peers that do not exist anymore in the network map received from the Management Service.
// It also removes peers that have been modified (e.g. change of IP address). They will be added again in addPeers method.
// removePeers removes up to maxPeersPerSyncPass peers per call. It returns true
// when more peers remained to remove than the cap, so the caller re-runs.
func (e *Engine) removePeers(peersUpdate []*mgmProto.RemotePeerConfig) (bool, error) {
// removePeers removes up to maxBatch peers per call. It returns true when more
// peers remained to remove than the cap, so the caller re-runs.
func (e *Engine) removePeers(peersUpdate []*mgmProto.RemotePeerConfig, maxBatch int) (bool, error) {
newPeers := make([]string, 0, len(peersUpdate))
for _, p := range peersUpdate {
newPeers = append(newPeers, p.GetWgPubKey())
@@ -833,8 +834,8 @@ func (e *Engine) removePeers(peersUpdate []*mgmProto.RemotePeerConfig) (bool, er
toRemove := util.SliceDiff(e.peerStore.PeersPubKey(), newPeers)
more := false
if len(toRemove) > maxPeersPerSyncPass {
toRemove = toRemove[:maxPeersPerSyncPass]
if len(toRemove) > maxBatch {
toRemove = toRemove[:maxBatch]
more = true
}
@@ -973,7 +974,7 @@ func (e *Engine) applySyncPass(update *mgmProto.SyncResponse) (bool, error) {
e.persistSyncResponse(update)
// only apply new changes and ignore old ones
more, err := e.updateNetworkMap(nm)
more, err := e.updateNetworkMap(nm, maxPeersPerSyncPass)
if err != nil {
return false, err
}
@@ -1369,9 +1370,9 @@ func (e *Engine) updateTURNs(turns []*mgmProto.ProtectedHostConfig) error {
}
// updateNetworkMap applies the wholesale parts (config, routes, ACL, DNS) in full
// and a bounded batch of peers. It returns true when more peers remained than the
// per-pass cap, so the caller re-runs until convergence.
func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) (bool, error) {
// and up to maxBatch peers per phase. It returns true when more peers remained
// than the cap, so the caller re-runs until convergence.
func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap, maxBatch int) (bool, error) {
// intentionally leave it before checking serial because for now it can happen that peer IP changed but serial didn't
if networkMap.GetPeerConfig() != nil {
err := e.updateConfig(networkMap.GetPeerConfig())
@@ -1471,17 +1472,17 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) (bool, error)
return false, err
}
} else {
removeMore, err := e.removePeers(remotePeers)
removeMore, err := e.removePeers(remotePeers, maxBatch)
if err != nil {
return false, err
}
modifyMore, err := e.modifyPeers(remotePeers)
modifyMore, err := e.modifyPeers(remotePeers, maxBatch)
if err != nil {
return false, err
}
addMore, err := e.addNewPeers(remotePeers)
addMore, err := e.addNewPeers(remotePeers, maxBatch)
if err != nil {
return false, err
}
@@ -1689,15 +1690,15 @@ func addrToString(addr netip.Addr) string {
}
// addNewPeers adds peers that were not know before but arrived from the Management service with the update
// addNewPeers adds up to maxPeersPerSyncPass not-yet-present peers per call. It
// returns true when more new peers remained than the cap, so the caller re-runs.
func (e *Engine) addNewPeers(peersUpdate []*mgmProto.RemotePeerConfig) (bool, error) {
// addNewPeers adds up to maxBatch not-yet-present peers per call. It returns true
// when more new peers remained than the cap, so the caller re-runs.
func (e *Engine) addNewPeers(peersUpdate []*mgmProto.RemotePeerConfig, maxBatch int) (bool, error) {
added := 0
for _, p := range peersUpdate {
if _, ok := e.peerStore.PeerConn(p.GetWgPubKey()); ok {
continue // already present (cheap skip), does not count toward the cap
}
if added >= maxPeersPerSyncPass {
if added >= maxBatch {
return true, nil // at least one more new peer remains
}
if err := e.addNewPeer(p); err != nil {

47
client/internal/engine_test.go
View File

@@ -433,7 +433,7 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
for _, c := range []testCase{case1, case2, case3, case4, case5, case6} {
t.Run(c.name, func(t *testing.T) {
_, err = engine.updateNetworkMap(c.networkMap)
_, err = engine.updateNetworkMap(c.networkMap, maxPeersPerSyncPass)
if err != nil {
t.Fatal(err)
return
@@ -460,6 +460,47 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
}
})
}
// chunked apply: with a per-pass cap smaller than the number of peers, a
// single updateNetworkMap applies one batch and reports more==true; the
// caller re-runs until convergence. (engine currently holds 0 peers.)
t.Run("chunked add converges over multiple passes", func(t *testing.T) {
nm := &mgmtProto.NetworkMap{
Serial: 6,
RemotePeers: []*mgmtProto.RemotePeerConfig{peer1, peer2, peer3},
}
more, err := engine.updateNetworkMap(nm, 1)
require.NoError(t, err)
require.True(t, more, "pass 1 should signal more")
require.Len(t, engine.peerStore.PeersPubKey(), 1)
more, err = engine.updateNetworkMap(nm, 1)
require.NoError(t, err)
require.True(t, more, "pass 2 should signal more")
require.Len(t, engine.peerStore.PeersPubKey(), 2)
more, err = engine.updateNetworkMap(nm, 1)
require.NoError(t, err)
require.False(t, more, "pass 3 should converge")
require.Len(t, engine.peerStore.PeersPubKey(), 3)
})
t.Run("chunked remove converges over multiple passes", func(t *testing.T) {
nm := &mgmtProto.NetworkMap{
Serial: 7,
RemotePeers: []*mgmtProto.RemotePeerConfig{peer1}, // remove peer2, peer3
}
more, err := engine.updateNetworkMap(nm, 1)
require.NoError(t, err)
require.True(t, more, "pass 1 should signal more (2 to remove, cap 1)")
more, err = engine.updateNetworkMap(nm, 1)
require.NoError(t, err)
require.False(t, more, "pass 2 should converge")
require.Len(t, engine.peerStore.PeersPubKey(), 1)
})
}
func TestEngine_UpdateNetworkMapWithRoutes(t *testing.T) {
@@ -630,7 +671,7 @@ func TestEngine_UpdateNetworkMapWithRoutes(t *testing.T) {
}
}()
_, err = engine.updateNetworkMap(testCase.networkMap)
_, err = engine.updateNetworkMap(testCase.networkMap, maxPeersPerSyncPass)
assert.NoError(t, err, "shouldn't return error")
assert.Equal(t, testCase.expectedSerial, input.inputSerial, "serial should match")
assert.Len(t, input.clientRoutes, testCase.expectedLen, "clientRoutes len should match")
@@ -834,7 +875,7 @@ func TestEngine_UpdateNetworkMapWithDNSUpdate(t *testing.T) {
}
}()
_, err = engine.updateNetworkMap(testCase.networkMap)
_, err = engine.updateNetworkMap(testCase.networkMap, maxPeersPerSyncPass)
assert.NoError(t, err, "shouldn't return error")
assert.Equal(t, testCase.expectedSerial, input.inputSerial, "serial should match")
assert.Len(t, input.inputNSGroups, testCase.expectedZonesLen, "zones len should match")