Merge branch 'main' into embedded-vnc

client/internal/engine.go

@@ -22,7 +22,6 @@ import (
 	log "github.com/sirupsen/logrus"
 	"golang.zx2c4.com/wireguard/tun/netstack"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
-	"google.golang.org/protobuf/proto"
 
 	nberrors "github.com/netbirdio/netbird/client/errors"
 	"github.com/netbirdio/netbird/client/firewall"
@@ -57,6 +56,7 @@ import (
 	"github.com/netbirdio/netbird/client/internal/routemanager"
 	"github.com/netbirdio/netbird/client/internal/routemanager/systemops"
 	"github.com/netbirdio/netbird/client/internal/statemanager"
+	"github.com/netbirdio/netbird/client/internal/syncstore"
 	"github.com/netbirdio/netbird/client/internal/updater"
 	"github.com/netbirdio/netbird/client/jobexec"
 	cProto "github.com/netbirdio/netbird/client/proto"
@@ -73,6 +73,7 @@ import (
 	sProto "github.com/netbirdio/netbird/shared/signal/proto"
 	"github.com/netbirdio/netbird/util"
 	"github.com/netbirdio/netbird/util/capture"
+	"github.com/netbirdio/netbird/version"
 )
 
 // PeerConnectionTimeoutMax is a timeout of an initial connection attempt to a remote peer.
@@ -151,6 +152,10 @@ type EngineConfig struct {
 
 	LogPath string
 	TempDir string
+
+	// StateDir is the directory holding the state file. The sync response
+	// (network map) is serialized here on platforms that persist it to disk.
+	StateDir string
 }
 
 // EngineServices holds the external service dependencies required by the Engine.
@@ -231,10 +236,15 @@ type Engine struct {
 
 	afpacketCapture *capture.AFPacketCapture
 
-	// Sync response persistence (protected by syncRespMux)
-	syncRespMux         sync.RWMutex
-	persistSyncResponse bool
-	latestSyncResponse  *mgmProto.SyncResponse
+	// Sync response persistence (protected by syncRespMux).
+	// syncStore is nil unless persistence has been enabled; its presence is
+	// what marks persistence as active. The backend (disk or memory) is
+	// selected per-platform; see the syncstore package. syncStoreDir is where
+	// a disk-backed store serializes to.
+	syncRespMux  sync.RWMutex
+	syncStore    syncstore.Store
+	syncStoreDir string
+
 	flowManager         nftypes.FlowManager
 
 	// auto-update
@@ -298,6 +308,7 @@ func NewEngine(
 		jobExecutor:        jobexec.NewExecutor(),
 		clientMetrics:      services.ClientMetrics,
 		updateManager:      services.UpdateManager,
+		syncStoreDir:       config.StateDir,
 	}
 
 	log.Infof("I am: %s", config.WgPrivateKey.PublicKey().String())
@@ -923,19 +934,18 @@ func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
 	}
 
 	// Persist sync response under the dedicated lock (syncRespMux), not under syncMsgMux.
-	// Read the storage-enabled flag under the syncRespMux too.
+	// A non-nil syncStore is what marks persistence as enabled. Hold the lock for
+	// the whole Set so the store cannot be cleared (disabled / engine close)
+	// mid-call and have this write resurrect a file that was just removed.
 	e.syncRespMux.RLock()
-	enabled := e.persistSyncResponse
-	e.syncRespMux.RUnlock()
-
-	// Store sync response if persistence is enabled
-	if enabled {
-		e.syncRespMux.Lock()
-		e.latestSyncResponse = update
-		e.syncRespMux.Unlock()
-
-		log.Debugf("sync response persisted with serial %d", nm.GetSerial())
+	if e.syncStore != nil {
+		if err := e.syncStore.Set(update); err != nil {
+			log.Errorf("failed to persist sync response: %v", err)
+		} else {
+			log.Debugf("sync response persisted with serial %d", nm.GetSerial())
+		}
 	}
+	e.syncRespMux.RUnlock()
 
 	// only apply new changes and ignore old ones
 	if err := e.updateNetworkMap(nm); err != nil {
@@ -1078,6 +1088,7 @@ func (e *Engine) updateConfig(conf *mgmProto.PeerConfig) error {
 	state.PubKey = e.config.WgPrivateKey.PublicKey().String()
 	state.KernelInterface = !e.wgInterface.IsUserspaceBind()
 	state.FQDN = conf.GetFqdn()
+	state.WgPort = e.config.WgPort
 
 	e.statusRecorder.UpdateLocalPeerState(state)
 
@@ -1156,6 +1167,7 @@ func (e *Engine) handleBundle(params *mgmProto.BundleParameters) (*mgmProto.JobR
 		LogPath:        e.config.LogPath,
 		TempDir:        e.config.TempDir,
 		ClientMetrics:  e.clientMetrics,
+		DaemonVersion:  version.NetbirdVersion(),
 		RefreshStatus: func() {
 			e.RunHealthProbes(true)
 		},
@@ -1834,6 +1846,18 @@ func (e *Engine) close() {
 	if err := e.portForwardManager.GracefullyStop(ctx); err != nil {
 		log.Warnf("failed to gracefully stop port forwarding manager: %s", err)
 	}
+
+	// Drop any persisted sync response so its network map does not linger on
+	// disk after the engine stops (and cannot leak into a later run).
+	e.syncRespMux.Lock()
+	store := e.syncStore
+	e.syncStore = nil
+	e.syncRespMux.Unlock()
+	if store != nil {
+		if err := store.Clear(); err != nil {
+			log.Warnf("failed to clear persisted sync response on close: %v", err)
+		}
+	}
 }
 
 func (e *Engine) readInitialSettings() ([]*route.Route, *nbdns.Config, bool, error) {
@@ -1989,6 +2013,29 @@ func (e *Engine) GetClientMetrics() *metrics.ClientMetrics {
 	return e.clientMetrics
 }
 
+// Performance bundles runtime-adjustable tunnel pool knobs.
+// See Engine.SetPerformance. Nil fields are ignored.
+type Performance struct {
+	PreallocatedBuffersPerPool *uint32
+}
+
+// SetPerformance applies the given tuning to this engine's live Device.
+func (e *Engine) SetPerformance(t Performance) error {
+	e.syncMsgMux.Lock()
+	defer e.syncMsgMux.Unlock()
+	if e.wgInterface == nil {
+		return fmt.Errorf("wg interface not initialized")
+	}
+	dev := e.wgInterface.GetWGDevice()
+	if dev == nil {
+		return fmt.Errorf("wg device not initialized")
+	}
+	if t.PreallocatedBuffersPerPool != nil {
+		dev.SetPreallocatedBuffersPerPool(*t.PreallocatedBuffersPerPool)
+	}
+	return nil
+}
+
 func findIPFromInterfaceName(ifaceName string) (net.IP, error) {
 	iface, err := net.InterfaceByName(ifaceName)
 	if err != nil {
@@ -2141,45 +2188,42 @@ func (e *Engine) stopDNSServer() {
 	e.statusRecorder.UpdateDNSStates(nsGroupStates)
 }
 
-// SetSyncResponsePersistence enables or disables sync response persistence
+// SetSyncResponsePersistence enables or disables sync response persistence.
+// The store is only instantiated while persistence is enabled; construction
+// itself drops any stale data left over from an earlier run (see syncstore).
 func (e *Engine) SetSyncResponsePersistence(enabled bool) {
 	e.syncRespMux.Lock()
 	defer e.syncRespMux.Unlock()
 
-	if enabled == e.persistSyncResponse {
+	if enabled == (e.syncStore != nil) {
 		return
 	}
-	e.persistSyncResponse = enabled
 	log.Debugf("Sync response persistence is set to %t", enabled)
 
 	if !enabled {
-		e.latestSyncResponse = nil
+		if err := e.syncStore.Clear(); err != nil {
+			log.Warnf("failed to clear persisted sync response: %v", err)
+		}
+		e.syncStore = nil
+		return
 	}
+
+	e.syncStore = syncstore.New(e.syncStoreDir)
 }
 
 // GetLatestSyncResponse returns the stored sync response if persistence is enabled
 func (e *Engine) GetLatestSyncResponse() (*mgmProto.SyncResponse, error) {
+	// Hold the lock for the whole Get so the store cannot be cleared
+	// (disabled / engine close) mid-call.
 	e.syncRespMux.RLock()
-	enabled := e.persistSyncResponse
-	latest := e.latestSyncResponse
-	e.syncRespMux.RUnlock()
+	defer e.syncRespMux.RUnlock()
 
-	if !enabled {
+	if e.syncStore == nil {
 		return nil, errors.New("sync response persistence is disabled")
 	}
 
-	if latest == nil {
-		//nolint:nilnil
-		return nil, nil
-	}
-
-	log.Debugf("Retrieving latest sync response with size %d bytes", proto.Size(latest))
-	sr, ok := proto.Clone(latest).(*mgmProto.SyncResponse)
-	if !ok {
-		return nil, fmt.Errorf("failed to clone sync response")
-	}
-
-	return sr, nil
+	//nolint:nilnil
+	return e.syncStore.Get()
 }
 
 // GetWgAddr returns the wireguard address
@@ -2215,7 +2259,7 @@ func (e *Engine) updateDNSForwarder(
 	enabled bool,
 	fwdEntries []*dnsfwd.ForwarderEntry,
 ) {
-	if e.config.DisableServerRoutes {
+	if e.config.DisableServerRoutes || e.config.BlockInbound {
 		return
 	}