diff --git a/client/internal/engine.go b/client/internal/engine.go index 692ad1e50..970e3447a 100644 --- a/client/internal/engine.go +++ b/client/internal/engine.go @@ -210,6 +210,12 @@ type Engine struct { // networkSerial is the latest CurrentSerial (state ID) of the network sent by the Management service networkSerial uint64 + // forwardingRules holds the ingress forward rules applied for the current target. + // Wholesale sections (incl. forward rules) run only on the first pass of a target; + // it is stashed here so the final, peer-converged pass can build the lazy-connection + // exclude list without recomputing them on every bounded peer pass. + forwardingRules []firewallManager.ForwardRule + networkMonitor *networkmonitor.NetworkMonitor sshServer sshServer @@ -918,7 +924,7 @@ func (e *Engine) handleAutoUpdateVersion(autoUpdateSettings *mgmProto.AutoUpdate // returns true if more peers remained than the per-pass cap. It is driven by the // mapStateManager, which re-invokes it (releasing the lock between passes) until // the update is fully applied. -func (e *Engine) applySyncPass(update *mgmProto.SyncResponse) (bool, error) { +func (e *Engine) applySyncPass(update *mgmProto.SyncResponse, firstPass bool) (bool, error) { e.syncMsgMux.Lock() defer e.syncMsgMux.Unlock() @@ -952,7 +958,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, maxPeersPerSyncPass) + more, err := e.updateNetworkMap(nm, maxPeersPerSyncPass, firstPass) if err != nil { return false, err } @@ -1362,7 +1368,7 @@ func (e *Engine) updateTURNs(turns []*mgmProto.ProtectedHostConfig) error { // updateNetworkMap applies the wholesale parts (config, routes, ACL, DNS) in full // 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) { +func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap, maxBatch int, firstPass bool) (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()) @@ -1377,6 +1383,86 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap, maxBatch int) return false, nil } + // Wholesale sections (firewall/ACL, DNS, routes, forward rules) are applied + // up-front and only once per target: they are cheap, local, idempotent and must + // be in place before peers come up (fail-closed). On the bounded re-runs that only + // drain the remaining peer batches they are skipped — the applied forward rules are + // reused from e.forwardingRules for the lazy-exclude finalize. + if firstPass { + e.applyWholesale(networkMap, serial) + } + + log.Debugf("got peers update from Management Service, total peers to connect to = %d", len(networkMap.GetRemotePeers())) + + e.updateOfflinePeers(networkMap.GetOfflinePeers()) + + // Filter out own peer from the remote peers list + localPubKey := e.config.WgPrivateKey.PublicKey().String() + remotePeers := make([]*mgmProto.RemotePeerConfig, 0, len(networkMap.GetRemotePeers())) + for _, p := range networkMap.GetRemotePeers() { + if p.GetWgPubKey() != localPubKey { + remotePeers = append(remotePeers, p) + } + } + + // needMore signals the caller to re-run when a peer phase hit its per-pass cap. + needMore := false + + // cleanup request, most likely our peer has been deleted + if networkMap.GetRemotePeersIsEmpty() { + err := e.removeAllPeers() + e.statusRecorder.FinishPeerListModifications() + if err != nil { + return false, err + } + } else { + removeMore, err := e.removePeers(remotePeers, maxBatch) + if err != nil { + return false, err + } + + modifyMore, err := e.modifyPeers(remotePeers, maxBatch) + if err != nil { + return false, err + } + + addMore, err := e.addNewPeers(remotePeers, maxBatch) + if err != nil { + return false, err + } + + needMore = removeMore || modifyMore || addMore + + e.statusRecorder.FinishPeerListModifications() + + e.updatePeerSSHHostKeys(remotePeers) + + if err := e.updateSSHClientConfig(remotePeers); err != nil { + log.Warnf("failed to update SSH client config: %v", err) + } + + e.updateSSHServerAuth(networkMap.GetSshAuth()) + } + + // Set the exclude list only once peers have fully converged (this pass added + // the last batch). It needs all target peers present in the store, and + // ExcludePeer has replace-semantics — a partial set mid-convergence would be wrong. + if !needMore { + excludedLazyPeers := e.toExcludedLazyPeers(e.forwardingRules, remotePeers) + e.connMgr.SetExcludeList(e.ctx, excludedLazyPeers) + } + + e.networkSerial = serial + + return needMore, nil +} + +// applyWholesale applies the cheap, local, idempotent map sections — lazy feature +// flag, firewall/legacy management, DNS, routes, ACL filtering, DNS forwarder and +// ingress forward rules — that must be in place before peers come up. It runs once +// per target (first pass only); the resulting forward rules are stashed in +// e.forwardingRules for the lazy-exclude finalize on the peer-converged pass. +func (e *Engine) applyWholesale(networkMap *mgmProto.NetworkMap, serial uint64) { if err := e.connMgr.UpdatedRemoteFeatureFlag(e.ctx, networkMap.GetPeerConfig().GetLazyConnectionEnabled()); err != nil { log.Errorf("failed to update lazy connection feature flag: %v", err) } @@ -1437,70 +1523,7 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap, maxBatch int) if err != nil { log.Errorf("failed to update forward rules, err: %v", err) } - - log.Debugf("got peers update from Management Service, total peers to connect to = %d", len(networkMap.GetRemotePeers())) - - e.updateOfflinePeers(networkMap.GetOfflinePeers()) - - // Filter out own peer from the remote peers list - localPubKey := e.config.WgPrivateKey.PublicKey().String() - remotePeers := make([]*mgmProto.RemotePeerConfig, 0, len(networkMap.GetRemotePeers())) - for _, p := range networkMap.GetRemotePeers() { - if p.GetWgPubKey() != localPubKey { - remotePeers = append(remotePeers, p) - } - } - - // needMore signals the caller to re-run when a peer phase hit its per-pass cap. - needMore := false - - // cleanup request, most likely our peer has been deleted - if networkMap.GetRemotePeersIsEmpty() { - err := e.removeAllPeers() - e.statusRecorder.FinishPeerListModifications() - if err != nil { - return false, err - } - } else { - removeMore, err := e.removePeers(remotePeers, maxBatch) - if err != nil { - return false, err - } - - modifyMore, err := e.modifyPeers(remotePeers, maxBatch) - if err != nil { - return false, err - } - - addMore, err := e.addNewPeers(remotePeers, maxBatch) - if err != nil { - return false, err - } - - needMore = removeMore || modifyMore || addMore - - e.statusRecorder.FinishPeerListModifications() - - e.updatePeerSSHHostKeys(remotePeers) - - if err := e.updateSSHClientConfig(remotePeers); err != nil { - log.Warnf("failed to update SSH client config: %v", err) - } - - e.updateSSHServerAuth(networkMap.GetSshAuth()) - } - - // Set the exclude list only once peers have fully converged (this pass added - // the last batch). It needs all target peers present in the store, and - // ExcludePeer has replace-semantics — a partial set mid-convergence would be wrong. - if !needMore { - excludedLazyPeers := e.toExcludedLazyPeers(forwardingRules, remotePeers) - e.connMgr.SetExcludeList(e.ctx, excludedLazyPeers) - } - - e.networkSerial = serial - - return needMore, nil + e.forwardingRules = forwardingRules } func toDNSFeatureFlag(networkMap *mgmProto.NetworkMap) bool { diff --git a/client/internal/engine_test.go b/client/internal/engine_test.go index 55263fa60..0543870e7 100644 --- a/client/internal/engine_test.go +++ b/client/internal/engine_test.go @@ -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, maxPeersPerSyncPass) + _, err = engine.updateNetworkMap(c.networkMap, maxPeersPerSyncPass, true) if err != nil { t.Fatal(err) return @@ -470,17 +470,17 @@ func TestEngine_UpdateNetworkMap(t *testing.T) { RemotePeers: []*mgmtProto.RemotePeerConfig{peer1, peer2, peer3}, } - more, err := engine.updateNetworkMap(nm, 1) + more, err := engine.updateNetworkMap(nm, 1, true) 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) + more, err = engine.updateNetworkMap(nm, 1, false) 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) + more, err = engine.updateNetworkMap(nm, 1, false) require.NoError(t, err) require.False(t, more, "pass 3 should converge") require.Len(t, engine.peerStore.PeersPubKey(), 3) @@ -492,11 +492,11 @@ func TestEngine_UpdateNetworkMap(t *testing.T) { RemotePeers: []*mgmtProto.RemotePeerConfig{peer1}, // remove peer2, peer3 } - more, err := engine.updateNetworkMap(nm, 1) + more, err := engine.updateNetworkMap(nm, 1, true) require.NoError(t, err) require.True(t, more, "pass 1 should signal more (2 to remove, cap 1)") - more, err = engine.updateNetworkMap(nm, 1) + more, err = engine.updateNetworkMap(nm, 1, false) require.NoError(t, err) require.False(t, more, "pass 2 should converge") require.Len(t, engine.peerStore.PeersPubKey(), 1) @@ -671,7 +671,7 @@ func TestEngine_UpdateNetworkMapWithRoutes(t *testing.T) { } }() - _, err = engine.updateNetworkMap(testCase.networkMap, maxPeersPerSyncPass) + _, err = engine.updateNetworkMap(testCase.networkMap, maxPeersPerSyncPass, true) 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") @@ -875,7 +875,7 @@ func TestEngine_UpdateNetworkMapWithDNSUpdate(t *testing.T) { } }() - _, err = engine.updateNetworkMap(testCase.networkMap, maxPeersPerSyncPass) + _, err = engine.updateNetworkMap(testCase.networkMap, maxPeersPerSyncPass, true) 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") diff --git a/client/internal/mapsync.go b/client/internal/mapsync.go index e7fed33f2..e30e3c1da 100644 --- a/client/internal/mapsync.go +++ b/client/internal/mapsync.go @@ -30,7 +30,11 @@ import ( // to a real, completed alignment. type mapStateManager struct { // apply performs one bounded apply pass and reports whether more passes are needed. - apply func(*mgmProto.SyncResponse) (bool, error) + // firstPass is true on the first pass of a given target, so the caller can run + // wholesale (firewall/routes/DNS/forward-rules) once per target and skip it on the + // re-runs that only drain the bounded peer batches. The manager owns this signal + // because it owns the convergence boundary; the engine need not track serials for it. + apply func(update *mgmProto.SyncResponse, firstPass bool) (bool, error) // onConverged is called once per processed map, with the elapsed time since that // map was received (for the sync-duration metric / "sync finished" log). onConverged func(time.Duration) @@ -44,7 +48,7 @@ type mapStateManager struct { wake chan struct{} } -func newMapStateManager(apply func(*mgmProto.SyncResponse) (bool, error), onConverged func(time.Duration)) *mapStateManager { +func newMapStateManager(apply func(update *mgmProto.SyncResponse, firstPass bool) (bool, error), onConverged func(time.Duration)) *mapStateManager { return &mapStateManager{ apply: apply, onConverged: onConverged, @@ -95,6 +99,10 @@ func (m *mapStateManager) mergeTarget(prev, update *mgmProto.SyncResponse) *mgmP // run drives convergence until ctx is done. It is meant to run in its own goroutine. func (m *mapStateManager) run(ctx context.Context) { + // passGen is the generation of the most recent apply() call (0 = none). A pass is + // the first for its target when its generation differs from the previous one — + // true on a fresh target and on a coalesced switch to a newer target mid-flight. + var passGen uint64 for { m.mu.Lock() target, tg, ag := m.target, m.targetGen, m.appliedGen @@ -110,7 +118,9 @@ func (m *mapStateManager) run(ctx context.Context) { } } - more, err := m.apply(target) + firstPass := tg != passGen + passGen = tg + more, err := m.apply(target, firstPass) if err != nil { if ctx.Err() != nil { return diff --git a/client/internal/mapsync_test.go b/client/internal/mapsync_test.go index 600499dd9..ba2736d26 100644 --- a/client/internal/mapsync_test.go +++ b/client/internal/mapsync_test.go @@ -16,10 +16,14 @@ import ( // fires onConverged exactly once, then blocks (no further apply) until a new target. func TestMapStateManager_ConvergesThenStops(t *testing.T) { var passes int32 + var firstPasses int32 converged := make(chan struct{}, 1) - apply := func(*mgmProto.SyncResponse) (bool, error) { + apply := func(_ *mgmProto.SyncResponse, firstPass bool) (bool, error) { n := atomic.AddInt32(&passes, 1) + if firstPass { + atomic.AddInt32(&firstPasses, 1) + } return n < 3, nil // more on pass 1 and 2, converge on pass 3 } m := newMapStateManager(apply, func(time.Duration) { converged <- struct{}{} }) @@ -36,6 +40,7 @@ func TestMapStateManager_ConvergesThenStops(t *testing.T) { t.Fatal("manager did not converge") } require.EqualValues(t, 3, atomic.LoadInt32(&passes)) + require.EqualValues(t, 1, atomic.LoadInt32(&firstPasses), "firstPass true only on pass 1, false on re-runs of the same target") // once converged the loop blocks: no further apply calls time.Sleep(100 * time.Millisecond) @@ -46,7 +51,7 @@ func TestMapStateManager_ConvergesThenStops(t *testing.T) { // onConverged exactly once — mirroring the legacy per-message handleSync timing. func TestMapStateManager_SignalsEachProcessedMap(t *testing.T) { converged := make(chan struct{}, 8) - apply := func(*mgmProto.SyncResponse) (bool, error) { + apply := func(_ *mgmProto.SyncResponse, _ bool) (bool, error) { return false, nil // converge in one pass } m := newMapStateManager(apply, func(time.Duration) { converged <- struct{}{} }) @@ -78,7 +83,7 @@ func TestMapStateManager_SignalsEachProcessedMap(t *testing.T) { func TestMapStateManager_SkippedMapNotSignaled(t *testing.T) { release := make(chan struct{}) var applies, converged atomic.Int32 - apply := func(*mgmProto.SyncResponse) (bool, error) { + apply := func(_ *mgmProto.SyncResponse, _ bool) (bool, error) { applies.Add(1) <-release // hold the first apply in-flight so we can queue a newer target return false, nil @@ -111,7 +116,7 @@ func TestMapStateManager_DropsTargetOnError(t *testing.T) { var failNext atomic.Bool failNext.Store(true) - apply := func(*mgmProto.SyncResponse) (bool, error) { + apply := func(_ *mgmProto.SyncResponse, _ bool) (bool, error) { applied <- struct{}{} if failNext.Load() { return false, errors.New("boom") @@ -154,7 +159,7 @@ func TestMapStateManager_DropsTargetOnError(t *testing.T) { // manager does not apply on its own. func TestMapStateManager_ReappliesOnNewTarget(t *testing.T) { applied := make(chan struct{}, 8) - apply := func(*mgmProto.SyncResponse) (bool, error) { + apply := func(_ *mgmProto.SyncResponse, _ bool) (bool, error) { applied <- struct{}{} return false, nil // converge in one pass }