routeselector: make exit-node reconciliation atomic

enforceSingleExitNode took the RouteSelector lock three separate times
(IsDeselectAll, then DeselectRoutes, then SelectRoutes), so a concurrent
DeselectAllRoutes could interleave and be silently undone: SelectRoutes on
its deselectAll branch clears the flag and re-selects the preferred exit
node, overriding the user's "all off".

Move the whole reconciliation into a single locked RouteSelector method
(SetExclusiveExitNode) that checks deselectAll inside the critical section,
so a deselect-all either fully precedes the reconcile (left untouched) or
fully follows it (honoured). No interleaving is possible.

.github/workflows/golang-test-linux.yml

@@ -166,7 +166,7 @@ jobs:
         # resolve; the grep then drops the broken package by path. Without -e,
         # go list aborts with empty stdout and `go test` falls back to the repo
         # root, which has no Go files.
-        run: CGO_ENABLED=1 GOARCH=${{ matrix.arch }} CI=true go test -coverprofile=coverage.txt -tags 'devcert privileged' -exec 'sudo --preserve-env=CI,CGO_ENABLED' -timeout 10m -p 1 $(go list -e ./... | grep -v -e /management -e /signal -e /relay -e /proxy -e /combined -e /client/ui -e /client/testutil/privileged)
+        run: CGO_ENABLED=1 GOARCH=${{ matrix.arch }} CI=true go test -coverprofile=coverage.txt -tags devcert -timeout 10m -p 1 $(go list -e ./... | grep -v -e /management -e /signal -e /relay -e /proxy -e /combined -e /client/ui)
 
       - name: Upload coverage reports to Codecov
         if: matrix.arch == 'amd64'

.github/workflows/release.yml

@@ -293,11 +293,8 @@ jobs:
           ${{ steps.goreleaser.outputs.artifacts }}
           JSON
 
-          # dockers_v2 artifacts have no top-level goarch field, so match the
-          # per-platform -amd64 tag suffix instead; it works for both the old
-          # dockers and the new dockers_v2 image naming.
           mapfile -t src_images < <(
-            jq -r '.[] | select(.type == "Docker Image") | .name | select(startswith("ghcr.io/") and endswith("-amd64"))' /tmp/goreleaser-artifacts.json
+            jq -r '.[] | select(.type == "Docker Image") | select(.goarch == "amd64") | .name | select(startswith("ghcr.io/"))' /tmp/goreleaser-artifacts.json
           )
 
           for src in "${src_images[@]}"; do

.gitignore

@@ -1,4 +1,3 @@
-.claude
 .idea
 .run
 *.iml

agent-network/README.md

@@ -1,47 +1,16 @@
 # NetBird Agent Network
 
-Agent Network is NetBird's access control layer for AI agents and the people who run them. 
-It gives every agent a real identity, tied to an identity provider (IdP), and governs what it can reach: LLM APIs and 
-AI gateways it can call, and the internal resources it can access. Traffic flows only over the encrypted NetBird tunnel, 
-scoped by policy, with no API keys or other credentials to leak. It also gives you control over cost and token usage.
+Agent Network is NetBird's access control layer for AI agents and the people who run
+them. It gives every agent a real identity, tied to your identity provider (IdP), and
+governs what it can reach — the LLM APIs and AI gateways it can call, and the internal
+resources it can access. Traffic flows only over the encrypted NetBird tunnel, scoped by
+policy, with no API keys to leak.
 
-Because every LLM request passes through an
-identity-aware proxy, you can:
-
-- **Set spending and rate limits** per agent, per user, or per team — with hard caps
-  that stop requests once a budget is reached.
-- **Restrict models and providers** so agents can only call approved (and cost-appropriate)
-  endpoints, keeping expensive models off-limits unless explicitly allowed.
-- **Attribute usage** by tracking token consumption and cost per identity, group, or cost center so every
-  request is tied back to the agent and person responsible.
-- **Reuse your existing AI gateway** — point the proxy at a gateway you already run,
-  keeping its routing and config in place while it adds identity on top, so you skip
-  API key distribution.
-
-https://github.com/user-attachments/assets/44d18286-d8ab-49f8-a457-98ccd66f3268
-
-> **Beta.** Agent Network is in beta, but it's stable and already running in
-> production environments. It's fully open source and can be self-hosted on your own
-> infrastructure, with no vendor lock-in and no data leaving your environment.
+> **Beta.** Agent Network is open source and can be self-hosted on your own
+> infrastructure.
 
 ## How it works
 
-Say you have a simple use case: your Engineering or IT team needs access to Claude Code or Codex, and you want visibility into usage plus the ability to enforce budgets.
-How can you do that without creating a dedicated API key for every team?
-
-With Agent Network you get a private endpoint inside your network, for example: https://mirror.netbird.ai
-Teams configure their agents to point to that endpoint instead of using individual API keys directly. 
-
-This endpoint is only reachable when users are connected to your NetBird network and authenticated through your IdP. Otherwise, it is not accessible from the public internet.
-You can then use this private endpoint to configure your AI agents, whether that is Claude Code, Codex, or another tool.
-
-## Quickstart
-
-Full step-by-step setup:
-**https://docs.netbird.io/agent-network/quickstart**
-
-## Architecture 
-
 Agent Network is built on two existing NetBird capabilities:
 
 - **Overlay network** — the encrypted WireGuard mesh between peers.
@@ -53,9 +22,6 @@ LLM traffic is routed through the proxy's identity-aware pipeline, while interna
 resources (databases, internal APIs, self-hosted models) are reached directly over
 peer-to-peer WireGuard tunnels, governed by the same identities and access policies.
 
-<img width="4720" height="2218" alt="image" src="https://github.com/user-attachments/assets/1afa5da1-4b82-4f8a-a7a8-f417efadf1eb" />
-
-
 ## Where the code lives
 
 There is no separate "agent-network" service — it reuses the reverse-proxy and management

client/android/env_list.go

@@ -10,7 +10,7 @@ var (
 	EnvKeyNBForceRelay = peer.EnvKeyNBForceRelay
 
 	// EnvKeyNBLazyConn Exported for Android java client to configure lazy connection
-	EnvKeyNBLazyConn = lazyconn.EnvLazyConn
+	EnvKeyNBLazyConn = lazyconn.EnvEnableLazyConn
 
 	// EnvKeyNBInactivityThreshold Exported for Android java client to configure connection inactivity threshold
 	EnvKeyNBInactivityThreshold = lazyconn.EnvInactivityThreshold

client/cmd/root.go

@@ -71,14 +71,12 @@ var (
 	extraIFaceBlackList     []string
 	anonymizeFlag           bool
 	dnsRouteInterval        time.Duration
-	// lazyConnEnabled is the parse target for the deprecated --enable-lazy-connection
-	// flag. The flag is inert; the value is no longer read (use NB_LAZY_CONN instead).
-	lazyConnEnabled        bool
-	mtu                    uint16
-	profilesDisabled       bool
-	updateSettingsDisabled bool
-	captureEnabled         bool
-	networksDisabled       bool
+	lazyConnEnabled         bool
+	mtu                     uint16
+	profilesDisabled        bool
+	updateSettingsDisabled  bool
+	captureEnabled          bool
+	networksDisabled        bool
 
 	rootCmd = &cobra.Command{
 		Use:          "netbird",
@@ -212,8 +210,7 @@ func init() {
 	upCmd.PersistentFlags().BoolVar(&rosenpassEnabled, enableRosenpassFlag, false, "[Experimental] Enable Rosenpass feature. If enabled, the connection will be post-quantum secured via Rosenpass.")
 	upCmd.PersistentFlags().BoolVar(&rosenpassPermissive, rosenpassPermissiveFlag, false, "[Experimental] Enable Rosenpass in permissive mode to allow this peer to accept WireGuard connections without requiring Rosenpass functionality from peers that do not have Rosenpass enabled.")
 	upCmd.PersistentFlags().BoolVar(&autoConnectDisabled, disableAutoConnectFlag, false, "Disables auto-connect feature. If enabled, then the client won't connect automatically when the service starts.")
-	upCmd.PersistentFlags().BoolVar(&lazyConnEnabled, enableLazyConnectionFlag, false, "Deprecated: no longer used. Lazy connections are controlled by the server and the NB_LAZY_CONN environment variable.")
-	_ = upCmd.PersistentFlags().MarkDeprecated(enableLazyConnectionFlag, "no longer used; lazy connections are controlled by the server and the NB_LAZY_CONN environment variable")
+	upCmd.PersistentFlags().BoolVar(&lazyConnEnabled, enableLazyConnectionFlag, false, "[Experimental] Enable the lazy connection feature. If enabled, the client will establish connections on-demand. Note: this setting may be overridden by management configuration.")
 
 }
 

client/cmd/service.go

@@ -5,7 +5,6 @@ package cmd
 import (
 	"context"
 	"fmt"
-	"net/http"
 	"runtime"
 	"strings"
 	"sync"
@@ -23,21 +22,15 @@ var serviceCmd = &cobra.Command{
 	Short: "Manage the NetBird daemon service",
 }
 
-const defaultJSONSocket = "unix:///var/run/netbird-http.sock"
-
 var (
-	serviceName      string
-	serviceEnvVars   []string
-	jsonSocket       string
-	enableJSONSocket bool
+	serviceName    string
+	serviceEnvVars []string
 )
 
 type program struct {
 	ctx              context.Context
 	cancel           context.CancelFunc
 	serv             *grpc.Server
-	jsonServ         *http.Server
-	jsonServMu       sync.Mutex
 	serverInstance   *server.Server
 	serverInstanceMu sync.Mutex
 }
@@ -53,8 +46,6 @@ func init() {
 	serviceCmd.PersistentFlags().BoolVar(&updateSettingsDisabled, "disable-update-settings", false, "Disables update settings feature. If enabled, the client will not be able to change or edit any settings. To persist this setting, use: netbird service install --disable-update-settings")
 	serviceCmd.PersistentFlags().BoolVar(&captureEnabled, "enable-capture", false, "Enables packet capture via 'netbird debug capture'. To persist, use: netbird service install --enable-capture")
 	serviceCmd.PersistentFlags().BoolVar(&networksDisabled, "disable-networks", false, "Disables network selection. If enabled, the client will not allow listing, selecting, or deselecting networks. To persist, use: netbird service install --disable-networks")
-	serviceCmd.PersistentFlags().BoolVar(&enableJSONSocket, "enable-json-socket", false, "Enables the HTTP/JSON API socket served by grpc-gateway. To persist, use: netbird service install --enable-json-socket")
-	serviceCmd.PersistentFlags().StringVar(&jsonSocket, "json-socket", defaultJSONSocket, "HTTP/JSON API socket address [unix|tcp]://[path|host:port]. Requires --enable-json-socket to serve. To persist, use: netbird service install --enable-json-socket --json-socket")
 
 	rootCmd.PersistentFlags().StringVarP(&serviceName, "service", "s", defaultServiceName, "Netbird system service name")
 	serviceEnvDesc := `Sets extra environment variables for the service. ` +

client/cmd/service_controller.go

@@ -5,6 +5,9 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"net"
+	"os"
+	"strings"
 	"time"
 
 	"github.com/kardianos/service"
@@ -19,56 +22,41 @@ import (
 	"github.com/netbirdio/netbird/util"
 )
 
-func validateJSONSocketFlags() error {
-	if serviceCmd.PersistentFlags().Changed("json-socket") && !enableJSONSocket {
-		return fmt.Errorf("--json-socket requires --enable-json-socket to configure the daemon JSON gateway")
-	}
-	return nil
-}
-
 func (p *program) Start(svc service.Service) error {
 	// Start should not block. Do the actual work async.
 	log.Info("starting NetBird service") //nolint
 
-	if err := validateJSONSocketFlags(); err != nil {
-		return err
-	}
-
 	// Collect static system and platform information
 	system.UpdateStaticInfoAsync()
 
 	// in any case, even if configuration does not exists we run daemon to serve CLI gRPC API.
 	p.serv = grpc.NewServer()
 
-	daemonListener, err := listenOnAddress(daemonAddr)
+	split := strings.Split(daemonAddr, "://")
+	switch split[0] {
+	case "unix":
+		// cleanup failed close
+		stat, err := os.Stat(split[1])
+		if err == nil && !stat.IsDir() {
+			if err := os.Remove(split[1]); err != nil {
+				log.Debugf("remove socket file: %v", err)
+			}
+		}
+	case "tcp":
+	default:
+		return fmt.Errorf("unsupported daemon address protocol: %v", split[0])
+	}
+
+	listen, err := net.Listen(split[0], split[1])
 	if err != nil {
 		return fmt.Errorf("listen daemon interface: %w", err)
 	}
-
-	var jsonListener *socketListener
-	if enableJSONSocket {
-		jsonListener, err = listenOnAddress(jsonSocket)
-		if err != nil {
-			_ = daemonListener.Close()
-			return fmt.Errorf("listen daemon JSON interface: %w", err)
-		}
-	} else {
-		removeStaleUnixSocketForAddress(jsonSocket)
-	}
-
 	go func() {
-		defer daemonListener.Close()
-		if jsonListener != nil {
-			defer jsonListener.Close()
-		}
+		defer listen.Close()
 
-		if err := daemonListener.chmodUnixSocket("daemon"); err != nil {
-			log.Error(err)
-			return
-		}
-		if jsonListener != nil {
-			if err := jsonListener.chmodUnixSocket("daemon JSON"); err != nil {
-				log.Error(err)
+		if split[0] == "unix" {
+			if err := os.Chmod(split[1], 0666); err != nil {
+				log.Errorf("failed setting daemon permissions: %v", split[1])
 				return
 			}
 		}
@@ -83,16 +71,8 @@ func (p *program) Start(svc service.Service) error {
 		p.serverInstance = serverInstance
 		p.serverInstanceMu.Unlock()
 
-		if jsonListener != nil {
-			if err := p.startJSONGateway(jsonListener, daemonAddr); err != nil {
-				log.Fatalf("failed to start daemon JSON server: %v", err)
-			}
-		} else {
-			log.Debug("daemon JSON socket disabled")
-		}
-
-		log.Printf("started daemon server: %v", daemonListener.address)
-		if err := p.serv.Serve(daemonListener.Listener); err != nil {
+		log.Printf("started daemon server: %v", split[1])
+		if err := p.serv.Serve(listen); err != nil {
 			log.Errorf("failed to serve daemon requests: %v", err)
 		}
 	}()
@@ -112,20 +92,6 @@ func (p *program) Stop(srv service.Service) error {
 
 	p.cancel()
 
-	p.jsonServMu.Lock()
-	jsonServ := p.jsonServ
-	p.jsonServMu.Unlock()
-	if jsonServ != nil {
-		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 2*time.Second)
-		if err := jsonServ.Shutdown(shutdownCtx); err != nil {
-			log.Errorf("failed to stop daemon JSON server gracefully: %v", err)
-			if err := jsonServ.Close(); err != nil {
-				log.Errorf("failed to close daemon JSON server: %v", err)
-			}
-		}
-		shutdownCancel()
-	}
-
 	if p.serv != nil {
 		p.serv.Stop()
 	}
@@ -182,9 +148,6 @@ var runCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
-		if err := validateJSONSocketFlags(); err != nil {
-			return err
-		}
 
 		return s.Run()
 	},
@@ -199,9 +162,6 @@ var startCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
-		if err := validateJSONSocketFlags(); err != nil {
-			return err
-		}
 
 		if err := s.Start(); err != nil {
 			return fmt.Errorf("start service: %w", err)
@@ -238,9 +198,6 @@ var restartCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
-		if err := validateJSONSocketFlags(); err != nil {
-			return err
-		}
 
 		if err := s.Restart(); err != nil {
 			return fmt.Errorf("restart service: %w", err)

client/cmd/service_installer.go

@@ -67,10 +67,6 @@ func buildServiceArguments() []string {
 		args = append(args, "--disable-networks")
 	}
 
-	if enableJSONSocket {
-		args = append(args, "--enable-json-socket", "--json-socket", jsonSocket)
-	}
-
 	return args
 }
 
@@ -110,10 +106,6 @@ func configurePlatformSpecificSettings(svcConfig *service.Config) error {
 
 // Create fully configured service config for install/reconfigure
 func createServiceConfigForInstall() (*service.Config, error) {
-	if err := validateJSONSocketFlags(); err != nil {
-		return nil, err
-	}
-
 	svcConfig, err := newSVCConfig()
 	if err != nil {
 		return nil, fmt.Errorf("create service config: %w", err)

client/cmd/service_json_gateway.go

@@ -1,52 +0,0 @@
-//go:build !ios && !android
-
-package cmd
-
-import (
-	"context"
-	"errors"
-	"net"
-	"net/http"
-	"strings"
-	"time"
-
-	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
-	log "github.com/sirupsen/logrus"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-
-	"github.com/netbirdio/netbird/client/proto"
-)
-
-func grpcGatewayEndpoint(addr string) string {
-	return strings.TrimPrefix(addr, "tcp://")
-}
-
-func (p *program) startJSONGateway(jsonListener *socketListener, daemonEndpoint string) error {
-	mux := runtime.NewServeMux()
-	opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
-	if err := proto.RegisterDaemonServiceHandlerFromEndpoint(p.ctx, mux, grpcGatewayEndpoint(daemonEndpoint), opts); err != nil {
-		return err
-	}
-
-	jsonServer := &http.Server{
-		Handler:           mux,
-		ReadHeaderTimeout: 5 * time.Second,
-		BaseContext: func(net.Listener) context.Context {
-			return p.ctx
-		},
-	}
-
-	p.jsonServMu.Lock()
-	p.jsonServ = jsonServer
-	p.jsonServMu.Unlock()
-
-	go func() {
-		log.Printf("started daemon JSON server: %v", jsonListener.address)
-		if err := jsonServer.Serve(jsonListener.Listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
-			log.Errorf("failed to serve daemon JSON requests: %v", err)
-		}
-	}()
-
-	return nil
-}

client/cmd/service_json_socket_test.go

@@ -1,176 +0,0 @@
-//go:build !ios && !android
-
-package cmd
-
-import (
-	"net"
-	"os"
-	"path/filepath"
-	"runtime"
-	"testing"
-
-	"github.com/spf13/cobra"
-	"github.com/spf13/pflag"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func preserveJSONSocketTestState(t *testing.T) {
-	t.Helper()
-
-	origJSONSocket := jsonSocket
-	origEnableJSONSocket := enableJSONSocket
-	origChanged := map[string]bool{}
-	serviceCmd.PersistentFlags().VisitAll(func(flag *pflag.Flag) {
-		origChanged[flag.Name] = flag.Changed
-	})
-
-	t.Cleanup(func() {
-		jsonSocket = origJSONSocket
-		enableJSONSocket = origEnableJSONSocket
-		serviceCmd.PersistentFlags().VisitAll(func(flag *pflag.Flag) {
-			flag.Changed = origChanged[flag.Name]
-		})
-	})
-}
-
-func TestJSONSocketFlagsArePositiveEnableOnly(t *testing.T) {
-	assert.NotNil(t, serviceCmd.PersistentFlags().Lookup("enable-json-socket"))
-	assert.NotNil(t, serviceCmd.PersistentFlags().Lookup("json-socket"))
-	assert.Nil(t, serviceCmd.PersistentFlags().Lookup("disable-json-socket"))
-	assert.Equal(t, "false", serviceCmd.PersistentFlags().Lookup("enable-json-socket").DefValue)
-}
-
-func TestBuildServiceArgumentsDefaultDisablesJSONSocket(t *testing.T) {
-	preserveJSONSocketTestState(t)
-
-	enableJSONSocket = false
-	jsonSocket = "tcp://127.0.0.1:8080"
-
-	args := buildServiceArguments()
-
-	assert.NotContains(t, args, "--enable-json-socket")
-	assert.NotContains(t, args, "--json-socket")
-}
-
-func TestBuildServiceArgumentsIncludesJSONSocketWhenEnabled(t *testing.T) {
-	preserveJSONSocketTestState(t)
-
-	enableJSONSocket = true
-	jsonSocket = "tcp://127.0.0.1:8080"
-
-	args := buildServiceArguments()
-
-	enableIndex := indexOfArg(args, "--enable-json-socket")
-	jsonIndex := indexOfArg(args, "--json-socket")
-	require.NotEqual(t, -1, enableIndex)
-	require.NotEqual(t, -1, jsonIndex)
-	require.Less(t, enableIndex, jsonIndex)
-	require.Less(t, jsonIndex+1, len(args))
-	assert.Equal(t, "tcp://127.0.0.1:8080", args[jsonIndex+1])
-}
-
-func TestJSONSocketWithoutEnableValidation(t *testing.T) {
-	preserveJSONSocketTestState(t)
-
-	enableJSONSocket = false
-	require.NoError(t, serviceCmd.PersistentFlags().Set("json-socket", "tcp://127.0.0.1:8080"))
-
-	err := validateJSONSocketFlags()
-
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "--enable-json-socket")
-}
-
-func TestJSONSocketWithEnableValidation(t *testing.T) {
-	preserveJSONSocketTestState(t)
-
-	require.NoError(t, serviceCmd.PersistentFlags().Set("enable-json-socket", "true"))
-	require.NoError(t, serviceCmd.PersistentFlags().Set("json-socket", "tcp://127.0.0.1:8080"))
-
-	assert.NoError(t, validateJSONSocketFlags())
-}
-
-func TestJSONSocketServiceParamsPersistEnableAndAddress(t *testing.T) {
-	preserveJSONSocketTestState(t)
-	serviceCmd.PersistentFlags().VisitAll(func(flag *pflag.Flag) {
-		flag.Changed = false
-	})
-
-	enableJSONSocket = true
-	jsonSocket = "tcp://127.0.0.1:8080"
-
-	params := currentServiceParams()
-	require.True(t, params.EnableJSONSocket)
-	require.Equal(t, "tcp://127.0.0.1:8080", params.JSONSocket)
-
-	enableJSONSocket = false
-	jsonSocket = defaultJSONSocket
-	applyServiceParams(testServiceEnvCommand(), params)
-
-	assert.True(t, enableJSONSocket)
-	assert.Equal(t, "tcp://127.0.0.1:8080", jsonSocket)
-}
-
-func TestRemoveStaleUnixSocketDoesNotRemoveRegularFile(t *testing.T) {
-	path := filepath.Join(t.TempDir(), "netbird-http.sock")
-	require.NoError(t, os.WriteFile(path, []byte("not a socket"), 0600))
-
-	removeStaleUnixSocket(path)
-
-	data, err := os.ReadFile(path)
-	require.NoError(t, err)
-	assert.Equal(t, []byte("not a socket"), data)
-}
-
-func TestRemoveStaleUnixSocketRemovesSocket(t *testing.T) {
-	if runtime.GOOS == "windows" {
-		t.Skip("unix sockets are not available on Windows")
-	}
-
-	path := filepath.Join(t.TempDir(), "netbird-http.sock")
-	addr := &net.UnixAddr{Name: path, Net: "unix"}
-	listener, err := net.ListenUnix("unix", addr)
-	require.NoError(t, err)
-	listener.SetUnlinkOnClose(false)
-	require.NoError(t, listener.Close())
-
-	_, err = os.Lstat(path)
-	require.NoError(t, err, "test setup must leave a stale Unix socket path")
-
-	removeStaleUnixSocket(path)
-
-	_, err = os.Lstat(path)
-	assert.True(t, os.IsNotExist(err), "expected stale Unix socket to be removed, got %v", err)
-}
-
-func TestRemoveStaleUnixSocketDoesNotRemoveLiveSocket(t *testing.T) {
-	if runtime.GOOS == "windows" {
-		t.Skip("unix sockets are not available on Windows")
-	}
-
-	path := filepath.Join(t.TempDir(), "netbird-http.sock")
-	listener, err := net.Listen("unix", path)
-	require.NoError(t, err)
-	defer listener.Close()
-
-	removeStaleUnixSocket(path)
-
-	_, err = os.Lstat(path)
-	assert.NoError(t, err, "expected live Unix socket to be preserved")
-}
-
-func testServiceEnvCommand() *cobra.Command {
-	cmd := &cobra.Command{}
-	cmd.Flags().StringSlice("service-env", nil, "")
-	return cmd
-}
-
-func indexOfArg(args []string, arg string) int {
-	for i, candidate := range args {
-		if candidate == arg {
-			return i
-		}
-	}
-	return -1
-}

client/cmd/service_params.go

@@ -23,7 +23,6 @@ const serviceParamsFile = "service.json"
 type serviceParams struct {
 	LogLevel              string            `json:"log_level"`
 	DaemonAddr            string            `json:"daemon_addr"`
-	JSONSocket            string            `json:"json_socket"`
 	ManagementURL         string            `json:"management_url,omitempty"`
 	ConfigPath            string            `json:"config_path,omitempty"`
 	LogFiles              []string          `json:"log_files,omitempty"`
@@ -31,7 +30,6 @@ type serviceParams struct {
 	DisableUpdateSettings bool              `json:"disable_update_settings,omitempty"`
 	EnableCapture         bool              `json:"enable_capture,omitempty"`
 	DisableNetworks       bool              `json:"disable_networks,omitempty"`
-	EnableJSONSocket      bool              `json:"enable_json_socket,omitempty"`
 	ServiceEnvVars        map[string]string `json:"service_env_vars,omitempty"`
 }
 
@@ -77,7 +75,6 @@ func currentServiceParams() *serviceParams {
 	params := &serviceParams{
 		LogLevel:              logLevel,
 		DaemonAddr:            daemonAddr,
-		JSONSocket:            jsonSocket,
 		ManagementURL:         managementURL,
 		ConfigPath:            configPath,
 		LogFiles:              logFiles,
@@ -85,7 +82,6 @@ func currentServiceParams() *serviceParams {
 		DisableUpdateSettings: updateSettingsDisabled,
 		EnableCapture:         captureEnabled,
 		DisableNetworks:       networksDisabled,
-		EnableJSONSocket:      enableJSONSocket,
 	}
 
 	if len(serviceEnvVars) > 0 {
@@ -117,8 +113,9 @@ func applyServiceParams(cmd *cobra.Command, params *serviceParams) {
 		return
 	}
 
-	// For fields with non-empty defaults, keep the != "" guard so that an older
-	// service.json missing the field doesn't clobber the default with an empty string.
+	// For fields with non-empty defaults (log-level, daemon-addr), keep the
+	// != "" guard so that an older service.json missing the field doesn't
+	// clobber the default with an empty string.
 	if !rootCmd.PersistentFlags().Changed("log-level") && params.LogLevel != "" {
 		logLevel = params.LogLevel
 	}
@@ -127,14 +124,6 @@ func applyServiceParams(cmd *cobra.Command, params *serviceParams) {
 		daemonAddr = params.DaemonAddr
 	}
 
-	if !serviceCmd.PersistentFlags().Changed("json-socket") && params.JSONSocket != "" {
-		jsonSocket = params.JSONSocket
-	}
-
-	if !serviceCmd.PersistentFlags().Changed("enable-json-socket") {
-		enableJSONSocket = params.EnableJSONSocket
-	}
-
 	// For optional fields where empty means "use default", always apply so
 	// that an explicit clear (--management-url "") persists across reinstalls.
 	if !rootCmd.PersistentFlags().Changed("management-url") {

client/cmd/service_params_test.go

@@ -41,8 +41,6 @@ func TestSaveAndLoadServiceParams(t *testing.T) {
 	params := &serviceParams{
 		LogLevel:              "debug",
 		DaemonAddr:            "unix:///var/run/netbird.sock",
-		JSONSocket:            "tcp://127.0.0.1:8080",
-		EnableJSONSocket:      true,
 		ManagementURL:         "https://my.server.com",
 		ConfigPath:            "/etc/netbird/config.json",
 		LogFiles:              []string{"/var/log/netbird/client.log", "console"},
@@ -65,8 +63,6 @@ func TestSaveAndLoadServiceParams(t *testing.T) {
 
 	assert.Equal(t, params.LogLevel, loaded.LogLevel)
 	assert.Equal(t, params.DaemonAddr, loaded.DaemonAddr)
-	assert.Equal(t, params.JSONSocket, loaded.JSONSocket)
-	assert.Equal(t, params.EnableJSONSocket, loaded.EnableJSONSocket)
 	assert.Equal(t, params.ManagementURL, loaded.ManagementURL)
 	assert.Equal(t, params.ConfigPath, loaded.ConfigPath)
 	assert.Equal(t, params.LogFiles, loaded.LogFiles)
@@ -105,8 +101,6 @@ func TestLoadServiceParams_InvalidJSON(t *testing.T) {
 func TestCurrentServiceParams(t *testing.T) {
 	origLogLevel := logLevel
 	origDaemonAddr := daemonAddr
-	origJSONSocket := jsonSocket
-	origEnableJSONSocket := enableJSONSocket
 	origManagementURL := managementURL
 	origConfigPath := configPath
 	origLogFiles := logFiles
@@ -116,8 +110,6 @@ func TestCurrentServiceParams(t *testing.T) {
 	t.Cleanup(func() {
 		logLevel = origLogLevel
 		daemonAddr = origDaemonAddr
-		jsonSocket = origJSONSocket
-		enableJSONSocket = origEnableJSONSocket
 		managementURL = origManagementURL
 		configPath = origConfigPath
 		logFiles = origLogFiles
@@ -128,8 +120,6 @@ func TestCurrentServiceParams(t *testing.T) {
 
 	logLevel = "trace"
 	daemonAddr = "tcp://127.0.0.1:9999"
-	jsonSocket = "tcp://127.0.0.1:8080"
-	enableJSONSocket = true
 	managementURL = "https://mgmt.example.com"
 	configPath = "/tmp/test-config.json"
 	logFiles = []string{"/tmp/test.log"}
@@ -141,8 +131,6 @@ func TestCurrentServiceParams(t *testing.T) {
 
 	assert.Equal(t, "trace", params.LogLevel)
 	assert.Equal(t, "tcp://127.0.0.1:9999", params.DaemonAddr)
-	assert.Equal(t, "tcp://127.0.0.1:8080", params.JSONSocket)
-	assert.True(t, params.EnableJSONSocket)
 	assert.Equal(t, "https://mgmt.example.com", params.ManagementURL)
 	assert.Equal(t, "/tmp/test-config.json", params.ConfigPath)
 	assert.Equal(t, []string{"/tmp/test.log"}, params.LogFiles)
@@ -154,8 +142,6 @@ func TestCurrentServiceParams(t *testing.T) {
 func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 	origLogLevel := logLevel
 	origDaemonAddr := daemonAddr
-	origJSONSocket := jsonSocket
-	origEnableJSONSocket := enableJSONSocket
 	origManagementURL := managementURL
 	origConfigPath := configPath
 	origLogFiles := logFiles
@@ -165,8 +151,6 @@ func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 	t.Cleanup(func() {
 		logLevel = origLogLevel
 		daemonAddr = origDaemonAddr
-		jsonSocket = origJSONSocket
-		enableJSONSocket = origEnableJSONSocket
 		managementURL = origManagementURL
 		configPath = origConfigPath
 		logFiles = origLogFiles
@@ -178,8 +162,6 @@ func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 	// Reset all flags to defaults.
 	logLevel = "info"
 	daemonAddr = "unix:///var/run/netbird.sock"
-	jsonSocket = defaultJSONSocket
-	enableJSONSocket = false
 	managementURL = ""
 	configPath = "/etc/netbird/config.json"
 	logFiles = []string{"/var/log/netbird/client.log"}
@@ -202,8 +184,6 @@ func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 	saved := &serviceParams{
 		LogLevel:              "debug",
 		DaemonAddr:            "tcp://127.0.0.1:5555",
-		JSONSocket:            "tcp://127.0.0.1:8080",
-		EnableJSONSocket:      true,
 		ManagementURL:         "https://saved.example.com",
 		ConfigPath:            "/saved/config.json",
 		LogFiles:              []string{"/saved/client.log"},
@@ -221,8 +201,6 @@ func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 
 	// All other fields were not Changed, so they should use saved values.
 	assert.Equal(t, "tcp://127.0.0.1:5555", daemonAddr)
-	assert.Equal(t, "tcp://127.0.0.1:8080", jsonSocket)
-	assert.True(t, enableJSONSocket)
 	assert.Equal(t, "https://saved.example.com", managementURL)
 	assert.Equal(t, "/saved/config.json", configPath)
 	assert.Equal(t, []string{"/saved/client.log"}, logFiles)
@@ -234,17 +212,14 @@ func TestApplyServiceParams_OnlyUnchangedFlags(t *testing.T) {
 func TestApplyServiceParams_BooleanRevertToFalse(t *testing.T) {
 	origProfilesDisabled := profilesDisabled
 	origUpdateSettingsDisabled := updateSettingsDisabled
-	origEnableJSONSocket := enableJSONSocket
 	t.Cleanup(func() {
 		profilesDisabled = origProfilesDisabled
 		updateSettingsDisabled = origUpdateSettingsDisabled
-		enableJSONSocket = origEnableJSONSocket
 	})
 
 	// Simulate current state where booleans are true (e.g. set by previous install).
 	profilesDisabled = true
 	updateSettingsDisabled = true
-	enableJSONSocket = true
 
 	// Reset Changed state so flags appear unset.
 	serviceCmd.PersistentFlags().VisitAll(func(f *pflag.Flag) {
@@ -263,7 +238,6 @@ func TestApplyServiceParams_BooleanRevertToFalse(t *testing.T) {
 
 	assert.False(t, profilesDisabled, "saved false should override current true")
 	assert.False(t, updateSettingsDisabled, "saved false should override current true")
-	assert.False(t, enableJSONSocket, "saved false should override current true")
 }
 
 func TestApplyServiceParams_ClearManagementURL(t *testing.T) {
@@ -556,7 +530,6 @@ func fieldToGlobalVar(field string) string {
 	m := map[string]string{
 		"LogLevel":              "logLevel",
 		"DaemonAddr":            "daemonAddr",
-		"JSONSocket":            "jsonSocket",
 		"ManagementURL":         "managementURL",
 		"ConfigPath":            "configPath",
 		"LogFiles":              "logFiles",
@@ -564,7 +537,6 @@ func fieldToGlobalVar(field string) string {
 		"DisableUpdateSettings": "updateSettingsDisabled",
 		"EnableCapture":         "captureEnabled",
 		"DisableNetworks":       "networksDisabled",
-		"EnableJSONSocket":      "enableJSONSocket",
 		"ServiceEnvVars":        "serviceEnvVars",
 	}
 	if v, ok := m[field]; ok {

client/cmd/service_socket.go

@@ -1,111 +0,0 @@
-//go:build !ios && !android
-
-package cmd
-
-import (
-	"errors"
-	"fmt"
-	"net"
-	"os"
-	"strings"
-	"syscall"
-	"time"
-
-	log "github.com/sirupsen/logrus"
-)
-
-type socketListener struct {
-	net.Listener
-	network string
-	address string
-}
-
-func listenOnAddress(addr string) (*socketListener, error) {
-	network, address, err := parseListenAddress(addr)
-	if err != nil {
-		return nil, err
-	}
-
-	if network == "unix" {
-		removeStaleUnixSocket(address)
-	}
-
-	listener, err := net.Listen(network, address)
-	if err != nil {
-		return nil, err
-	}
-
-	return &socketListener{Listener: listener, network: network, address: address}, nil
-}
-
-func parseListenAddress(addr string) (string, string, error) {
-	network, address, ok := strings.Cut(addr, "://")
-	if !ok || network == "" || address == "" {
-		return "", "", fmt.Errorf("address must be in [unix|tcp]://[path|host:port] format: %q", addr)
-	}
-
-	switch network {
-	case "unix", "tcp":
-		return network, address, nil
-	default:
-		return "", "", fmt.Errorf("unsupported daemon address protocol: %v", network)
-	}
-}
-
-func removeStaleUnixSocket(path string) {
-	stat, err := os.Lstat(path)
-	if err != nil {
-		if !os.IsNotExist(err) {
-			log.Debugf("stat socket file: %v", err)
-		}
-		return
-	}
-
-	if stat.Mode()&os.ModeSocket == 0 {
-		return
-	}
-
-	if !isStaleUnixSocket(path) {
-		return
-	}
-
-	if err := os.Remove(path); err != nil {
-		log.Debugf("remove socket file: %v", err)
-	}
-}
-
-func isStaleUnixSocket(path string) bool {
-	conn, err := net.DialTimeout("unix", path, 100*time.Millisecond)
-	if err == nil {
-		if closeErr := conn.Close(); closeErr != nil {
-			log.Debugf("close unix socket probe: %v", closeErr)
-		}
-		return false
-	}
-
-	if os.IsNotExist(err) || os.IsPermission(err) || os.IsTimeout(err) {
-		log.Debugf("not removing unix socket %s after probe error: %v", path, err)
-		return false
-	}
-
-	return errors.Is(err, syscall.ECONNREFUSED)
-}
-
-func removeStaleUnixSocketForAddress(addr string) {
-	network, address, err := parseListenAddress(addr)
-	if err != nil || network != "unix" {
-		return
-	}
-	removeStaleUnixSocket(address)
-}
-
-func (l *socketListener) chmodUnixSocket(description string) error {
-	if l == nil || l.network != "unix" {
-		return nil
-	}
-
-	if err := os.Chmod(l.address, 0666); err != nil {
-		return fmt.Errorf("failed setting %s permissions for %s: %w", description, l.address, err)
-	}
-	return nil
-}

client/cmd/up.go

@@ -479,6 +479,10 @@ func setupSetConfigReq(customDNSAddressConverted []byte, cmd *cobra.Command, pro
 		req.DisableIpv6 = &disableIPv6
 	}
 
+	if cmd.Flag(enableLazyConnectionFlag).Changed {
+		req.LazyConnectionEnabled = &lazyConnEnabled
+	}
+
 	return &req
 }
 
@@ -596,6 +600,9 @@ func setupConfig(customDNSAddressConverted []byte, cmd *cobra.Command, configFil
 		ic.DisableIPv6 = &disableIPv6
 	}
 
+	if cmd.Flag(enableLazyConnectionFlag).Changed {
+		ic.LazyConnectionEnabled = &lazyConnEnabled
+	}
 	return &ic, nil
 }
 
@@ -711,6 +718,9 @@ func setupLoginRequest(providedSetupKey string, customDNSAddressConverted []byte
 		loginRequest.DisableIpv6 = &disableIPv6
 	}
 
+	if cmd.Flag(enableLazyConnectionFlag).Changed {
+		loginRequest.LazyConnectionEnabled = &lazyConnEnabled
+	}
 	return &loginRequest, nil
 }
 

client/iface/configurer/kernel_unix.go

@@ -17,15 +17,12 @@ import (
 
 type KernelConfigurer struct {
 	deviceName string
-	statsCache *statsCache
 }
 
 func NewKernelConfigurer(deviceName string) *KernelConfigurer {
-	c := &KernelConfigurer{
+	return &KernelConfigurer{
 		deviceName: deviceName,
 	}
-	c.statsCache = newStatsCache(statsCacheTTL, c.fetchStats)
-	return c
 }
 
 func (c *KernelConfigurer) ConfigureInterface(privateKey string, port int) error {
@@ -249,6 +246,12 @@ func (c *KernelConfigurer) configure(config wgtypes.Config) error {
 		}
 	}()
 
+	// validate if device with name exists
+	_, err = wg.Device(c.deviceName)
+	if err != nil {
+		return err
+	}
+
 	return wg.ConfigureDevice(c.deviceName, config)
 }
 
@@ -297,14 +300,6 @@ func (c *KernelConfigurer) FullStats() (*Stats, error) {
 }
 
 func (c *KernelConfigurer) GetStats() (map[string]WGStats, error) {
-	return c.statsCache.get()
-}
-
-func (c *KernelConfigurer) LastActivities() map[string]monotime.Time {
-	return nil
-}
-
-func (c *KernelConfigurer) fetchStats() (map[string]WGStats, error) {
 	stats := make(map[string]WGStats)
 	wg, err := wgctrl.New()
 	if err != nil {
@@ -331,3 +326,7 @@ func (c *KernelConfigurer) fetchStats() (map[string]WGStats, error) {
 	}
 	return stats, nil
 }
+
+func (c *KernelConfigurer) LastActivities() map[string]monotime.Time {
+	return nil
+}

client/iface/configurer/stats_cache.go

@@ -1,52 +0,0 @@
-package configurer
-
-import (
-	"sync"
-	"time"
-
-	"golang.org/x/sync/singleflight"
-)
-
-const statsCacheTTL = 1 * time.Second
-
-type statsCache struct {
-	ttl   time.Duration
-	fetch func() (map[string]WGStats, error)
-
-	mu       sync.RWMutex
-	value    map[string]WGStats
-	expireAt time.Time
-
-	sf singleflight.Group
-}
-
-func newStatsCache(ttl time.Duration, fetch func() (map[string]WGStats, error)) *statsCache {
-	return &statsCache{ttl: ttl, fetch: fetch}
-}
-
-func (c *statsCache) get() (map[string]WGStats, error) {
-	c.mu.RLock()
-	if c.value != nil && time.Now().Before(c.expireAt) {
-		value := c.value
-		c.mu.RUnlock()
-		return value, nil
-	}
-	c.mu.RUnlock()
-
-	value, err, _ := c.sf.Do("stats", func() (interface{}, error) {
-		res, err := c.fetch()
-		if err != nil {
-			return nil, err
-		}
-
-		c.mu.Lock()
-		c.value = res
-		c.expireAt = time.Now().Add(c.ttl)
-		c.mu.Unlock()
-		return res, nil
-	})
-	if err != nil {
-		return nil, err
-	}
-	return value.(map[string]WGStats), nil
-}

client/iface/configurer/stats_cache_test.go

@@ -1,70 +0,0 @@
-package configurer
-
-import (
-	"errors"
-	"sync"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/require"
-)
-
-func TestStatsCache_CachesWithinTTL(t *testing.T) {
-	var calls atomic.Int64
-	c := newStatsCache(50*time.Millisecond, func() (map[string]WGStats, error) {
-		calls.Add(1)
-		return map[string]WGStats{"p": {}}, nil
-	})
-
-	for i := 0; i < 10; i++ {
-		_, err := c.get()
-		require.NoError(t, err)
-	}
-	require.Equal(t, int64(1), calls.Load(), "within TTL only one underlying fetch")
-
-	time.Sleep(60 * time.Millisecond)
-	_, err := c.get()
-	require.NoError(t, err)
-	require.Equal(t, int64(2), calls.Load(), "after TTL expiry a fresh fetch happens")
-}
-
-func TestStatsCache_SingleFlight(t *testing.T) {
-	var calls atomic.Int64
-	release := make(chan struct{})
-	c := newStatsCache(time.Minute, func() (map[string]WGStats, error) {
-		calls.Add(1)
-		<-release
-		return map[string]WGStats{}, nil
-	})
-
-	const n = 50
-	var wg sync.WaitGroup
-	wg.Add(n)
-	for i := 0; i < n; i++ {
-		go func() {
-			defer wg.Done()
-			_, _ = c.get()
-		}()
-	}
-	time.Sleep(20 * time.Millisecond)
-	close(release)
-	wg.Wait()
-
-	require.Equal(t, int64(1), calls.Load(), "concurrent misses collapse into one fetch")
-}
-
-func TestStatsCache_ErrorNotCached(t *testing.T) {
-	var calls atomic.Int64
-	wantErr := errors.New("dump failed")
-	c := newStatsCache(time.Minute, func() (map[string]WGStats, error) {
-		calls.Add(1)
-		return nil, wantErr
-	})
-
-	_, err := c.get()
-	require.ErrorIs(t, err, wantErr)
-	_, err = c.get()
-	require.ErrorIs(t, err, wantErr)
-	require.Equal(t, int64(2), calls.Load(), "errors are not cached; each call retries")
-}

client/iface/configurer/usp.go

@@ -40,7 +40,6 @@ type WGUSPConfigurer struct {
 	device           *device.Device
 	deviceName       string
 	activityRecorder *bind.ActivityRecorder
-	statsCache       *statsCache
 
 	uapiListener net.Listener
 }
@@ -51,19 +50,16 @@ func NewUSPConfigurer(device *device.Device, deviceName string, activityRecorder
 		deviceName:       deviceName,
 		activityRecorder: activityRecorder,
 	}
-	wgCfg.statsCache = newStatsCache(statsCacheTTL, wgCfg.fetchStats)
 	wgCfg.startUAPI()
 	return wgCfg
 }
 
 func NewUSPConfigurerNoUAPI(device *device.Device, deviceName string, activityRecorder *bind.ActivityRecorder) *WGUSPConfigurer {
-	wgCfg := &WGUSPConfigurer{
+	return &WGUSPConfigurer{
 		device:           device,
 		deviceName:       deviceName,
 		activityRecorder: activityRecorder,
 	}
-	wgCfg.statsCache = newStatsCache(statsCacheTTL, wgCfg.fetchStats)
-	return wgCfg
 }
 
 func (c *WGUSPConfigurer) ConfigureInterface(privateKey string, port int) error {
@@ -352,10 +348,6 @@ func (t *WGUSPConfigurer) Close() {
 }
 
 func (t *WGUSPConfigurer) GetStats() (map[string]WGStats, error) {
-	return t.statsCache.get()
-}
-
-func (t *WGUSPConfigurer) fetchStats() (map[string]WGStats, error) {
 	ipc, err := t.device.IpcGet()
 	if err != nil {
 		return nil, fmt.Errorf("ipc get: %w", err)

client/internal/auth/auth.go

@@ -351,6 +351,7 @@ func (a *Auth) setSystemInfoFlags(info *system.Info) {
 		a.config.BlockLANAccess,
 		a.config.BlockInbound,
 		a.config.DisableIPv6,
+		a.config.LazyConnectionEnabled,
 		a.config.EnableSSHRoot,
 		a.config.EnableSSHSFTP,
 		a.config.EnableSSHLocalPortForwarding,

client/internal/conn_mgr.go

@@ -16,16 +16,6 @@ import (
 	"github.com/netbirdio/netbird/route"
 )
 
-// lazyForce is the resolved local decision for lazy connections, layered above the
-// management feature flag. lazyForceNone defers to management.
-type lazyForce int
-
-const (
-	lazyForceNone lazyForce = iota
-	lazyForceOn
-	lazyForceOff
-)
-
 // ConnMgr coordinates both lazy connections (established on-demand) and permanent peer connections.
 //
 // The connection manager is responsible for:
@@ -38,7 +28,7 @@ type ConnMgr struct {
 	peerStore        *peerstore.Store
 	statusRecorder   *peer.Status
 	iface            lazyconn.WGIface
-	force            lazyForce
+	enabledLocally   bool
 	rosenpassEnabled bool
 
 	lazyConnMgr *manager.Manager
@@ -53,34 +43,28 @@ func NewConnMgr(engineConfig *EngineConfig, statusRecorder *peer.Status, peerSto
 		peerStore:        peerStore,
 		statusRecorder:   statusRecorder,
 		iface:            iface,
-		force:            resolveLazyForce(engineConfig.LazyConnection),
 		rosenpassEnabled: engineConfig.RosenpassEnabled,
 	}
+	if engineConfig.LazyConnectionEnabled || lazyconn.IsLazyConnEnabledByEnv() {
+		e.enabledLocally = true
+	}
 	return e
 }
 
-// Start initializes the connection manager. It starts the lazy connection manager when a
-// local override forces it on; with no local override it waits for the management feature flag.
+// Start initializes the connection manager and starts the lazy connection manager if enabled by env var or cmd line option.
 func (e *ConnMgr) Start(ctx context.Context) {
 	if e.lazyConnMgr != nil {
 		log.Errorf("lazy connection manager is already started")
 		return
 	}
 
-	switch e.force {
-	case lazyForceOff:
-		log.Infof("lazy connection manager is disabled by local override (%s or MDM policy)", lazyconn.EnvLazyConn)
-		e.statusRecorder.UpdateLazyConnection(false)
-		return
-	case lazyForceNone:
-		log.Infof("lazy connection manager is managed by the management feature flag")
-		e.statusRecorder.UpdateLazyConnection(false)
+	if !e.enabledLocally {
+		log.Infof("lazy connection manager is disabled")
 		return
 	}
 
 	if e.rosenpassEnabled {
 		log.Warnf("rosenpass connection manager is enabled, lazy connection manager will not be started")
-		e.statusRecorder.UpdateLazyConnection(false)
 		return
 	}
 
@@ -92,8 +76,8 @@ func (e *ConnMgr) Start(ctx context.Context) {
 // If enabled, it initializes the lazy connection manager and start it. Do not need to call Start() again.
 // If disabled, then it closes the lazy connection manager and open the connections to all peers.
 func (e *ConnMgr) UpdatedRemoteFeatureFlag(ctx context.Context, enabled bool) error {
-	// a local override (NB_LAZY_CONN or local config) takes precedence over management
-	if e.force != lazyForceNone {
+	// do not disable lazy connection manager if it was enabled by env var
+	if e.enabledLocally {
 		return nil
 	}
 
@@ -105,7 +89,6 @@ func (e *ConnMgr) UpdatedRemoteFeatureFlag(ctx context.Context, enabled bool) er
 
 		if e.rosenpassEnabled {
 			log.Infof("rosenpass connection manager is enabled, lazy connection manager will not be started")
-			e.statusRecorder.UpdateLazyConnection(false)
 			return nil
 		}
 
@@ -115,7 +98,6 @@ func (e *ConnMgr) UpdatedRemoteFeatureFlag(ctx context.Context, enabled bool) er
 		return e.addPeersToLazyConnManager()
 	} else {
 		if e.lazyConnMgr == nil {
-			e.statusRecorder.UpdateLazyConnection(false)
 			return nil
 		}
 		log.Infof("lazy connection manager is disabled by management feature flag")
@@ -125,6 +107,37 @@ func (e *ConnMgr) UpdatedRemoteFeatureFlag(ctx context.Context, enabled bool) er
 	}
 }
 
+// SetLocalLazyConn applies a local lazy connection override (UI / CLI / env).
+// While enabledLocally is true, UpdatedRemoteFeatureFlag (management sync) is a
+// no-op, so the local setting wins until it is turned off again.
+func (e *ConnMgr) SetLocalLazyConn(ctx context.Context, enabled bool) error {
+	e.enabledLocally = enabled
+
+	if enabled {
+		if e.lazyConnMgr != nil {
+			return nil
+		}
+
+		if e.rosenpassEnabled {
+			log.Warnf("rosenpass connection manager is enabled, lazy connection manager will not be started")
+			return nil
+		}
+
+		log.Infof("lazy connection manager is enabled locally")
+		e.initLazyManager(ctx)
+		e.statusRecorder.UpdateLazyConnection(true)
+		return e.addPeersToLazyConnManager()
+	}
+
+	if e.lazyConnMgr == nil {
+		return nil
+	}
+	log.Infof("lazy connection manager is disabled locally")
+	e.closeManager(ctx)
+	e.statusRecorder.UpdateLazyConnection(false)
+	return nil
+}
+
 // UpdateRouteHAMap updates the route HA mappings in the lazy connection manager
 func (e *ConnMgr) UpdateRouteHAMap(haMap route.HAMap) {
 	if !e.isStartedWithLazyMgr() {
@@ -327,25 +340,6 @@ func (e *ConnMgr) isStartedWithLazyMgr() bool {
 	return e.lazyConnMgr != nil && e.lazyCtxCancel != nil
 }
 
-// resolveLazyForce determines the local override. NB_LAZY_CONN takes precedence; when it
-// is unset the MDM policy override (mdmState) applies. Either wins in both directions over
-// the management feature flag; StateUnset for both defers to management.
-func resolveLazyForce(mdmState lazyconn.State) lazyForce {
-	state := lazyconn.EnvState()
-	if state == lazyconn.StateUnset {
-		state = mdmState
-	}
-
-	switch state {
-	case lazyconn.StateOn:
-		return lazyForceOn
-	case lazyconn.StateOff:
-		return lazyForceOff
-	default:
-		return lazyForceNone
-	}
-}
-
 func inactivityThresholdEnv() *time.Duration {
 	envValue := os.Getenv(lazyconn.EnvInactivityThreshold)
 	if envValue == "" {

client/internal/conn_mgr_test.go

@@ -1,40 +0,0 @@
-package internal
-
-import (
-	"os"
-	"testing"
-
-	"github.com/netbirdio/netbird/client/internal/lazyconn"
-)
-
-func TestResolveLazyForce(t *testing.T) {
-	tests := []struct {
-		name   string
-		env    string
-		envSet bool
-		mdm    lazyconn.State
-		want   lazyForce
-	}{
-		{name: "env unset, mdm unset -> defer to management", mdm: lazyconn.StateUnset, want: lazyForceNone},
-		{name: "env on -> force on", env: "on", envSet: true, mdm: lazyconn.StateUnset, want: lazyForceOn},
-		{name: "env off -> force off", env: "off", envSet: true, mdm: lazyconn.StateUnset, want: lazyForceOff},
-		{name: "env unset, mdm on -> force on", mdm: lazyconn.StateOn, want: lazyForceOn},
-		{name: "env unset, mdm off -> force off", mdm: lazyconn.StateOff, want: lazyForceOff},
-		{name: "env on beats mdm off", env: "on", envSet: true, mdm: lazyconn.StateOff, want: lazyForceOn},
-		{name: "env off beats mdm on", env: "off", envSet: true, mdm: lazyconn.StateOn, want: lazyForceOff},
-		{name: "unrecognized env, mdm on -> mdm wins", env: "auto", envSet: true, mdm: lazyconn.StateOn, want: lazyForceOn},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			t.Setenv(lazyconn.EnvLazyConn, tt.env)
-			if !tt.envSet {
-				os.Unsetenv(lazyconn.EnvLazyConn)
-			}
-
-			if got := resolveLazyForce(tt.mdm); got != tt.want {
-				t.Fatalf("resolveLazyForce(%v) = %v, want %v", tt.mdm, got, tt.want)
-			}
-		})
-	}
-}

client/internal/connect.go

@@ -27,7 +27,6 @@ import (
 	"github.com/netbirdio/netbird/client/iface/device"
 	"github.com/netbirdio/netbird/client/iface/netstack"
 	"github.com/netbirdio/netbird/client/internal/dns"
-	"github.com/netbirdio/netbird/client/internal/lazyconn"
 	"github.com/netbirdio/netbird/client/internal/listener"
 	"github.com/netbirdio/netbird/client/internal/metrics"
 	"github.com/netbirdio/netbird/client/internal/peer"
@@ -619,7 +618,7 @@ func createEngineConfig(key wgtypes.Key, config *profilemanager.Config, peerConf
 		BlockInbound:        config.BlockInbound,
 		DisableIPv6:         config.DisableIPv6,
 
-		LazyConnection: lazyconn.ParseState(config.LazyConnection),
+		LazyConnectionEnabled: config.LazyConnectionEnabled,
 
 		MTU:     selectMTU(config.MTU, peerConfig.Mtu),
 		LogPath: logPath,
@@ -693,6 +692,7 @@ func loginToManagement(ctx context.Context, client mgm.Client, pubSSHKey []byte,
 		config.BlockLANAccess,
 		config.BlockInbound,
 		config.DisableIPv6,
+		config.LazyConnectionEnabled,
 		config.EnableSSHRoot,
 		config.EnableSSHSFTP,
 		config.EnableSSHLocalPortForwarding,

client/internal/debug/debug.go

@@ -695,7 +695,7 @@ func (g *BundleGenerator) addCommonConfigFields(configContent *strings.Builder)
 		configContent.WriteString(fmt.Sprintf("ClientCertKeyPath: %s\n", g.internalConfig.ClientCertKeyPath))
 	}
 
-	configContent.WriteString(fmt.Sprintf("LazyConnection: %q\n", g.internalConfig.LazyConnection))
+	configContent.WriteString(fmt.Sprintf("LazyConnectionEnabled: %v\n", g.internalConfig.LazyConnectionEnabled))
 	configContent.WriteString(fmt.Sprintf("MTU: %d\n", g.internalConfig.MTU))
 }
 

client/internal/debug/debug_test.go

@@ -885,7 +885,7 @@ func TestAddConfig_AllFieldsCovered(t *testing.T) {
 		DNSRouteInterval:              5 * time.Second,
 		ClientCertPath:                "/tmp/cert",
 		ClientCertKeyPath:             "/tmp/key",
-		LazyConnection:                "on",
+		LazyConnectionEnabled:         true,
 		MTU:                           1280,
 	}
 

client/internal/engine.go

@@ -40,7 +40,6 @@ import (
 	"github.com/netbirdio/netbird/client/internal/dnsfwd"
 	"github.com/netbirdio/netbird/client/internal/expose"
 	"github.com/netbirdio/netbird/client/internal/ingressgw"
-	"github.com/netbirdio/netbird/client/internal/lazyconn"
 	"github.com/netbirdio/netbird/client/internal/metrics"
 	"github.com/netbirdio/netbird/client/internal/netflow"
 	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
@@ -148,9 +147,7 @@ type EngineConfig struct {
 	BlockInbound        bool
 	DisableIPv6         bool
 
-	// LazyConnection is the MDM-sourced lazy-connection override; StateUnset defers to
-	// the env var and management feature flag.
-	LazyConnection lazyconn.State
+	LazyConnectionEnabled bool
 
 	MTU uint16
 
@@ -1164,6 +1161,7 @@ func (e *Engine) applyInfoFlags(info *system.Info) {
 		e.config.BlockLANAccess,
 		e.config.BlockInbound,
 		e.config.DisableIPv6,
+		e.config.LazyConnectionEnabled,
 		e.config.EnableSSHRoot,
 		e.config.EnableSSHSFTP,
 		e.config.EnableSSHLocalPortForwarding,
@@ -2032,6 +2030,7 @@ func (e *Engine) readInitialSettings() ([]*route.Route, *nbdns.Config, bool, err
 		e.config.BlockLANAccess,
 		e.config.BlockInbound,
 		e.config.DisableIPv6,
+		e.config.LazyConnectionEnabled,
 		e.config.EnableSSHRoot,
 		e.config.EnableSSHSFTP,
 		e.config.EnableSSHLocalPortForwarding,

client/internal/engine_lazyconn.go

@@ -0,0 +1,19 @@
+package internal
+
+import (
+	"errors"
+)
+
+// SetLazyConnEnabled applies a local lazy connection override to the running
+// engine. It pins the setting like an env/CLI flag, so a later management sync
+// cannot override it. syncMsgMux guards ConnMgr, which is not thread-safe.
+func (e *Engine) SetLazyConnEnabled(enabled bool) error {
+	e.syncMsgMux.Lock()
+	defer e.syncMsgMux.Unlock()
+
+	if e.connMgr == nil {
+		return errors.New("connection manager is not initialised")
+	}
+
+	return e.connMgr.SetLocalLazyConn(e.ctx, enabled)
+}

client/internal/lazyconn/env.go

@@ -3,57 +3,24 @@ package lazyconn
 import (
 	"os"
 	"strconv"
-	"strings"
 
 	log "github.com/sirupsen/logrus"
 )
 
 const (
-	EnvLazyConn            = "NB_LAZY_CONN"
+	EnvEnableLazyConn      = "NB_ENABLE_EXPERIMENTAL_LAZY_CONN"
 	EnvInactivityThreshold = "NB_LAZY_CONN_INACTIVITY_THRESHOLD"
 )
 
-// State is the tri-state local override for lazy connections read from the environment.
-type State int
-
-const (
-	// StateUnset means no local override; defer to the management feature flag.
-	StateUnset State = iota
-	// StateOn forces lazy connections on, overriding management.
-	StateOn
-	// StateOff forces lazy connections off, overriding management.
-	StateOff
-)
-
-// EnvState reads NB_LAZY_CONN and returns the local override state.
-func EnvState() State {
-	return ParseState(os.Getenv(EnvLazyConn))
-}
-
-// ParseState interprets a lazy-connection override value (from the environment or an MDM
-// policy). It accepts the on/off aliases plus any value strconv.ParseBool understands
-// (true/false/1/0). An empty or unrecognized value returns StateUnset so that the
-// management feature flag remains in control.
-func ParseState(raw string) State {
-	if raw == "" {
-		return StateUnset
+func IsLazyConnEnabledByEnv() bool {
+	val := os.Getenv(EnvEnableLazyConn)
+	if val == "" {
+		return false
 	}
-
-	normalized := strings.ToLower(strings.TrimSpace(raw))
-	switch normalized {
-	case "on":
-		return StateOn
-	case "off":
-		return StateOff
-	}
-
-	enabled, err := strconv.ParseBool(normalized)
+	enabled, err := strconv.ParseBool(val)
 	if err != nil {
-		log.Warnf("failed to parse lazy connection value %q (from %s env or MDM policy): %v", raw, EnvLazyConn, err)
-		return StateUnset
+		log.Warnf("failed to parse %s: %v", EnvEnableLazyConn, err)
+		return false
 	}
-	if enabled {
-		return StateOn
-	}
-	return StateOff
+	return enabled
 }

client/internal/lazyconn/env_test.go

@@ -1,45 +0,0 @@
-package lazyconn
-
-import (
-	"os"
-	"testing"
-)
-
-func TestEnvState(t *testing.T) {
-	tests := []struct {
-		value string
-		set   bool
-		want  State
-	}{
-		{set: false, want: StateUnset},
-		{value: "", set: true, want: StateUnset},
-		{value: "on", set: true, want: StateOn},
-		{value: "ON", set: true, want: StateOn},
-		{value: "true", set: true, want: StateOn},
-		{value: "1", set: true, want: StateOn},
-		{value: " on ", set: true, want: StateOn},
-		{value: "off", set: true, want: StateOff},
-		{value: "OFF", set: true, want: StateOff},
-		{value: "false", set: true, want: StateOff},
-		{value: "0", set: true, want: StateOff},
-		{value: "auto", set: true, want: StateUnset},
-		{value: "garbage", set: true, want: StateUnset},
-	}
-
-	for _, tt := range tests {
-		name := tt.value
-		if !tt.set {
-			name = "unset"
-		}
-		t.Run(name, func(t *testing.T) {
-			t.Setenv(EnvLazyConn, tt.value)
-			if !tt.set {
-				os.Unsetenv(EnvLazyConn)
-			}
-
-			if got := EnvState(); got != tt.want {
-				t.Fatalf("EnvState() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}

client/internal/netflow/logger/logger.go

@@ -27,7 +27,7 @@ type Logger struct {
 	wgIfaceNetV6       netip.Prefix
 	dnsCollection      atomic.Bool
 	exitNodeCollection atomic.Bool
-	Store              types.AggregatingStore
+	Store              types.Store
 }
 
 func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix) *Logger {
@@ -35,7 +35,7 @@ func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix)
 		statusRecorder: statusRecorder,
 		wgIfaceNet:     wgIfaceIPNet,
 		wgIfaceNetV6:   wgIfaceIPNetV6,
-		Store:          store.NewAggregatingMemoryStore(),
+		Store:          store.NewMemoryStore(),
 	}
 }
 
@@ -125,10 +125,6 @@ func (l *Logger) stop() {
 	l.mux.Unlock()
 }
 
-func (l *Logger) ResetAggregationWindow() types.FlowEventAggregator {
-	return l.Store.ResetAggregationWindow()
-}
-
 func (l *Logger) GetEvents() []*types.Event {
 	return l.Store.GetEvents()
 }

client/internal/netflow/manager.go

@@ -9,14 +9,12 @@ import (
 	"sync"
 	"time"
 
-	"github.com/cenkalti/backoff/v4"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/netbirdio/netbird/client/internal/netflow/conntrack"
 	"github.com/netbirdio/netbird/client/internal/netflow/logger"
-	"github.com/netbirdio/netbird/client/internal/netflow/store"
 	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/flow/client"
@@ -25,16 +23,14 @@ import (
 
 // Manager handles netflow tracking and logging
 type Manager struct {
-	mux               sync.Mutex
-	shutdownWg        sync.WaitGroup
-	logger            nftypes.FlowLogger
-	flowConfig        *nftypes.FlowConfig
-	conntrack         nftypes.ConnTracker
-	receiverClient    *client.GRPCClient
-	eventsWithoutAcks nftypes.Store
-	publicKey         []byte
-	cancel            context.CancelFunc
-	retryInterval     time.Duration
+	mux            sync.Mutex
+	shutdownWg     sync.WaitGroup
+	logger         nftypes.FlowLogger
+	flowConfig     *nftypes.FlowConfig
+	conntrack      nftypes.ConnTracker
+	receiverClient *client.GRPCClient
+	publicKey      []byte
+	cancel         context.CancelFunc
 }
 
 // NewManager creates a new netflow manager
@@ -52,11 +48,9 @@ func NewManager(iface nftypes.IFaceMapper, publicKey []byte, statusRecorder *pee
 	}
 
 	return &Manager{
-		logger:            flowLogger,
-		conntrack:         ct,
-		publicKey:         publicKey,
-		retryInterval:     time.Second,
-		eventsWithoutAcks: store.NewMemoryStore(),
+		logger:    flowLogger,
+		conntrack: ct,
+		publicKey: publicKey,
 	}
 }
 
@@ -72,7 +66,6 @@ func (m *Manager) needsNewClient(previous *nftypes.FlowConfig) bool {
 }
 
 // enableFlow starts components for flow tracking
-// must be called under m.mux lock
 func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
 	// first make sender ready so events don't pile up
 	if m.needsNewClient(previous) {
@@ -92,7 +85,6 @@ func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
 	return nil
 }
 
-// must be called under m.mux lock
 func (m *Manager) resetClient() error {
 	if m.receiverClient != nil {
 		if err := m.receiverClient.Close(); err != nil {
@@ -115,19 +107,14 @@ func (m *Manager) resetClient() error {
 	ctx, cancel := context.WithCancel(context.Background())
 	m.cancel = cancel
 
-	m.shutdownWg.Add(3)
-	flowConfigInterval := m.flowConfig.Interval
+	m.shutdownWg.Add(2)
 	go func() {
 		defer m.shutdownWg.Done()
-		m.receiveACKs(ctx, flowClient, flowConfigInterval)
+		m.receiveACKs(ctx, flowClient)
 	}()
 	go func() {
 		defer m.shutdownWg.Done()
-		m.startSender(ctx, flowConfigInterval)
-	}()
-	go func() {
-		defer m.shutdownWg.Done()
-		m.startRetries(ctx, flowConfigInterval)
+		m.startSender(ctx)
 	}()
 
 	return nil
@@ -211,8 +198,8 @@ func (m *Manager) GetLogger() nftypes.FlowLogger {
 	return m.logger
 }
 
-func (m *Manager) startSender(ctx context.Context, flowConfigInterval time.Duration) {
-	ticker := time.NewTicker(flowConfigInterval)
+func (m *Manager) startSender(ctx context.Context) {
+	ticker := time.NewTicker(m.flowConfig.Interval)
 	defer ticker.Stop()
 
 	for {
@@ -220,29 +207,27 @@ func (m *Manager) startSender(ctx context.Context, flowConfigInterval time.Durat
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			collectedEvents := m.logger.ResetAggregationWindow()
-			events := collectedEvents.GetAggregatedEvents()
+			events := m.logger.GetEvents()
 			for _, event := range events {
-				m.eventsWithoutAcks.StoreEvent(event)
 				if err := m.send(event); err != nil {
 					log.Errorf("failed to send flow event to server: %v", err)
-				} else {
-					log.Tracef("sent flow event: %s", event.ID)
+					continue
 				}
+				log.Tracef("sent flow event: %s", event.ID)
 			}
 		}
 	}
 }
 
-func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient, flowConfigInterval time.Duration) {
-	err := client.Receive(ctx, flowConfigInterval, func(ack *proto.FlowEventAck) error {
+func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
+	err := client.Receive(ctx, m.flowConfig.Interval, func(ack *proto.FlowEventAck) error {
 		id, err := uuid.FromBytes(ack.EventId)
 		if err != nil {
 			log.Warnf("failed to convert ack event id to uuid: %v", err)
 			return nil
 		}
 		log.Tracef("received flow event ack: %s", id)
-		m.eventsWithoutAcks.DeleteEvents([]uuid.UUID{id})
+		m.logger.DeleteEvents([]uuid.UUID{id})
 		return nil
 	})
 
@@ -251,51 +236,6 @@ func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient, fl
 	}
 }
 
-// We effectively never drop events (see MaxInterval), which makes eventsWithoutAcks unbounded.
-// We may want to limit the max size of the store, and start dropping oldest events when the threshold is reached.
-func (m *Manager) startRetries(ctx context.Context, flowConfigInterval time.Duration) {
-	timer := time.NewTimer(m.retryInterval)
-	retryBackoff := backoff.WithContext(&backoff.ExponentialBackOff{
-		InitialInterval:     1 * time.Second,
-		RandomizationFactor: 0.5,
-		Multiplier:          1.7,
-		MaxInterval:         flowConfigInterval / 2,
-		MaxElapsedTime:      3 * 30 * 24 * time.Hour, // 3 months
-		Stop:                backoff.Stop,
-		Clock:               backoff.SystemClock,
-	}, ctx)
-	defer timer.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-timer.C:
-			resetBackoff := true
-			for _, e := range m.eventsWithoutAcks.GetEvents() {
-				if e.Timestamp.Add(time.Second).After(time.Now()) {
-					// grace period on retries to avoid early retries
-					// do not retry if the event is less than 1 sec old
-					continue
-				}
-				if err := m.send(e); err != nil {
-					if nextBackoff := retryBackoff.NextBackOff(); nextBackoff != backoff.Stop {
-						timer = time.NewTimer(nextBackoff)
-						resetBackoff = false
-					} else {
-						resetBackoff = true // we exhausted retries, reset retry loop
-					}
-					break
-				}
-			}
-			if resetBackoff { // use regular retry interval in absence of network errors
-				retryBackoff.Reset()
-				timer = time.NewTimer(m.retryInterval)
-			}
-		}
-	}
-}
-
 func (m *Manager) send(event *nftypes.Event) error {
 	m.mux.Lock()
 	client := m.receiverClient
@@ -310,11 +250,9 @@ func (m *Manager) send(event *nftypes.Event) error {
 
 func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 	protoEvent := &proto.FlowEvent{
-		EventId:     event.ID[:],
-		Timestamp:   timestamppb.New(event.Timestamp),
-		PublicKey:   publicKey,
-		WindowStart: timestamppb.New(event.WindowStart),
-		WindowEnd:   timestamppb.New(event.WindowEnd),
+		EventId:   event.ID[:],
+		Timestamp: timestamppb.New(event.Timestamp),
+		PublicKey: publicKey,
 		FlowFields: &proto.FlowFields{
 			FlowId:           event.FlowID[:],
 			RuleId:           event.RuleID,
@@ -329,9 +267,6 @@ func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 			TxBytes:          event.TxBytes,
 			SourceResourceId: event.SourceResourceID,
 			DestResourceId:   event.DestResourceID,
-			NumOfStarts:      event.NumOfStarts,
-			NumOfEnds:        event.NumOfEnds,
-			NumOfDrops:       event.NumOfDrops,
 		},
 	}
 

client/internal/netflow/manager_integration_test.go

@@ -1,291 +0,0 @@
-package netflow
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"net"
-	"net/netip"
-	"slices"
-	"testing"
-	"time"
-
-	"github.com/google/uuid"
-	"github.com/netbirdio/netbird/client/iface/wgaddr"
-	"github.com/netbirdio/netbird/client/internal/netflow/types"
-	"github.com/netbirdio/netbird/flow/proto"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"google.golang.org/grpc"
-)
-
-type testServer struct {
-	proto.UnimplementedFlowServiceServer
-	events         chan *proto.FlowEvent
-	acks           chan *proto.FlowEventAck
-	grpcSrv        *grpc.Server
-	addr           string
-	handlerDone    chan struct{} // signaled each time Events() exits
-	handlerStarted chan struct{} // signaled each time Events() begins
-}
-
-func newTestServer(t *testing.T) *testServer {
-	listener, err := net.Listen("tcp", "127.0.0.1:0")
-	require.NoError(t, err)
-
-	s := &testServer{
-		events:         make(chan *proto.FlowEvent, 100),
-		acks:           make(chan *proto.FlowEventAck, 100),
-		grpcSrv:        grpc.NewServer(),
-		addr:           listener.Addr().String(),
-		handlerDone:    make(chan struct{}, 10),
-		handlerStarted: make(chan struct{}, 10),
-	}
-
-	proto.RegisterFlowServiceServer(s.grpcSrv, s)
-
-	go func() {
-		if err := s.grpcSrv.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
-			t.Logf("server error: %v", err)
-		}
-	}()
-
-	t.Cleanup(func() {
-		s.grpcSrv.Stop()
-	})
-
-	return s
-}
-
-func (s *testServer) Events(stream proto.FlowService_EventsServer) error {
-	defer func() {
-		select {
-		case s.handlerDone <- struct{}{}:
-		default:
-		}
-	}()
-
-	err := stream.Send(&proto.FlowEventAck{IsInitiator: true})
-	if err != nil {
-		return err
-	}
-
-	select {
-	case s.handlerStarted <- struct{}{}:
-	default:
-	}
-
-	ctx, cancel := context.WithCancel(stream.Context())
-	defer cancel()
-
-	go func() {
-		defer cancel()
-		for {
-			event, err := stream.Recv()
-			if err != nil {
-				return
-			}
-
-			if !event.IsInitiator {
-				select {
-				case s.events <- event:
-				case <-ctx.Done():
-					return
-				}
-			}
-		}
-	}()
-
-	for {
-		select {
-		case ack := <-s.acks:
-			if err := stream.Send(ack); err != nil {
-				return err
-			}
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
-}
-
-func TestSendEventReceiveAck(t *testing.T) {
-	_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	t.Cleanup(cancel)
-
-	server := newTestServer(t)
-	manager := createManager(t, server.addr, 60*time.Second) // set high to prevent retries in this test
-	defer manager.Close()
-
-	assert.Eventually(t, func() bool {
-		select {
-		case <-server.handlerStarted:
-			return true
-		default:
-			return false
-		}
-	}, 3*time.Second, 100*time.Millisecond)
-
-	event1 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.2"),
-		DestPort:  2345,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event1)
-	event2 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.1"),
-		DestPort:  1234,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event2)
-
-	// verify the server received logged events
-	serverSideEvents := make([]*proto.FlowEvent, 0)
-	assert.Eventually(t, func() bool {
-		select {
-		case event := <-server.events:
-			serverSideEvents = append(serverSideEvents, event)
-			if len(serverSideEvents) == 2 {
-				return true
-			}
-		default:
-			if len(serverSideEvents) == 2 {
-				return true
-			}
-		}
-		return false
-	}, 5*time.Second, 100*time.Millisecond)
-
-	serverSideFlowIds := make([]uuid.UUID, 0, 2)
-	slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
-		id, err := uuid.FromBytes(e.FlowFields.FlowId)
-		assert.NoError(t, err)
-		serverSideFlowIds = append(serverSideFlowIds, id)
-		return true
-	})
-	assert.ElementsMatch(t, []uuid.UUID{event1.FlowID, event2.FlowID}, serverSideFlowIds)
-
-	// verify the manager tracks un-acked events
-	unackedEvents := manager.eventsWithoutAcks.GetEvents()
-	assert.Len(t, unackedEvents, 2)
-	flowIds := make([]uuid.UUID, 0)
-	slices.Values(unackedEvents)(func(e *types.Event) bool {
-		flowIds = append(flowIds, e.FlowID)
-		return true
-	})
-	assert.ElementsMatch(t, flowIds, []uuid.UUID{event1.FlowID, event2.FlowID})
-}
-
-// verify handling of retries:
-//   - unacked events are retried
-//   - when acks arrive, events are removed from the un-acked event tracker
-func TestRetryEvents(t *testing.T) {
-	_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	t.Cleanup(cancel)
-
-	server := newTestServer(t)
-	manager := createManager(t, server.addr, time.Second) // set low to start retries sooner
-	defer manager.Close()
-
-	assert.Eventually(t, func() bool {
-		select {
-		case <-server.handlerStarted:
-			return true
-		default:
-			return false
-		}
-	}, 3*time.Second, 100*time.Millisecond)
-
-	event1 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.2"),
-		DestPort:  2345,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event1)
-	event2 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.1"),
-		DestPort:  1234,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event2)
-
-	// verify the server received retries of logged events
-	serverSideEvents := make([]*proto.FlowEvent, 0)
-	func() {
-		c := time.After(2500 * time.Millisecond)
-		for {
-			select {
-			case event := <-server.events:
-				serverSideEvents = append(serverSideEvents, event)
-			case <-c:
-				return
-			}
-		}
-	}()
-	assert.True(t, len(serverSideEvents) > 2) // must see retries
-
-	uniqueServerSideEvents := make(map[uuid.UUID]*proto.FlowEvent)
-	slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
-		id, err := uuid.FromBytes(e.FlowFields.FlowId)
-		assert.NoError(t, err)
-		uniqueServerSideEvents[id] = e
-		return true
-	})
-	assert.Contains(t, uniqueServerSideEvents, event1.FlowID)
-	assert.Contains(t, uniqueServerSideEvents, event2.FlowID)
-
-	// ack events
-	server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event1.FlowID].EventId}
-	server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event2.FlowID].EventId}
-
-	assert.EventuallyWithT(t, func(c *assert.CollectT) {
-		unackedEvents := manager.eventsWithoutAcks.GetEvents()
-		assert.Empty(c, unackedEvents)
-
-	}, 3*time.Second, 100*time.Millisecond)
-}
-
-func createManager(t *testing.T, serverAddr string, retryInterval time.Duration) *Manager {
-	t.Helper()
-
-	mockIFace := &mockIFaceMapper{
-		address: wgaddr.Address{
-			Network: netip.MustParsePrefix("192.168.1.1/32"),
-		},
-		isUserspaceBind: true,
-	}
-
-	publicKey := []byte("test-public-key")
-	manager := NewManager(mockIFace, publicKey, nil)
-	manager.retryInterval = retryInterval
-
-	initialConfig := &types.FlowConfig{
-		Enabled:        true,
-		URL:            fmt.Sprintf("http://%s", serverAddr),
-		TokenPayload:   "initial-payload",
-		TokenSignature: "initial-signature",
-		Interval:       500 * time.Millisecond,
-	}
-
-	err := manager.Update(initialConfig)
-	require.NoError(t, err)
-
-	return manager
-}
-
-func ipAddr(a string) netip.Addr {
-	addr, _ := netip.ParseAddr(a)
-	return addr
-}

client/internal/netflow/store/event_aggregation_test.go

@@ -1,362 +0,0 @@
-package store
-
-import (
-	"math/rand"
-	"net/netip"
-	"testing"
-	"time"
-
-	"github.com/google/uuid"
-	"github.com/netbirdio/netbird/client/internal/netflow/types"
-	"github.com/stretchr/testify/assert"
-)
-
-var random = rand.New(rand.NewSource(time.Now().UnixNano()))
-
-func TestFlowAggregation(t *testing.T) {
-	var protocols = []types.Protocol{types.ICMP, types.ICMPv6, types.TCP, types.UDP}
-	var tests = []struct {
-		description string
-		addresses   [][]netip.Addr
-		dstPort     uint16
-		eventTypes  []types.Type
-	}{
-		{
-			description: "start and stop",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
-		},
-		{
-			description: "start and drop",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
-		},
-		{
-			description: "start only",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeStart},
-		},
-		{
-			description: "drop only",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeDrop},
-		}}
-
-	for _, protocol := range protocols {
-		for _, tt := range tests {
-			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
-				store := NewAggregatingMemoryStore()
-				store.WindowEnd = time.Now().Add(5 * time.Second)
-
-				allExpected := make([]*types.Event, 0)
-
-				for _, srcAndDst := range tt.addresses {
-					inEvents, expected := generateEvents(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, protocol, types.Ingress, 0, store.WindowStart, store.WindowEnd)
-					for _, e := range inEvents {
-						store.StoreEvent(e)
-					}
-					allExpected = append(allExpected, expected)
-				}
-
-				events := store.GetAggregatedEvents()
-				assert.ElementsMatch(t, events, allExpected)
-			})
-		}
-	}
-}
-
-func TestIcmpEventAggregation(t *testing.T) {
-	var protocols = []types.Protocol{types.ICMP, types.ICMPv6}
-	var icmpTypes = []uint8{1, 2, 3}
-
-	var tests = []struct {
-		description string
-		addresses   [][]netip.Addr
-		eventTypes  []types.Type
-	}{
-		{
-			description: "start and stop",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
-			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
-		},
-		{
-			description: "start and drop",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
-			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
-		},
-		{
-			description: "start only",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
-			eventTypes:  []types.Type{types.TypeStart},
-		},
-		{
-			description: "drop only",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
-			eventTypes:  []types.Type{types.TypeDrop},
-		}}
-
-	for _, protocol := range protocols {
-		for _, tt := range tests {
-			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
-				store := NewAggregatingMemoryStore()
-				store.WindowEnd = time.Now().Add(5 * time.Second)
-
-				allExpected := make([]*types.Event, 0)
-				for _, icmpType := range icmpTypes {
-					events, expected := generateEvents(tt.addresses[0][0], tt.addresses[0][1], 0, tt.eventTypes, protocol, types.Ingress, icmpType, store.WindowStart, store.WindowEnd)
-					for _, e := range events {
-						store.StoreEvent(e)
-					}
-					allExpected = append(allExpected, expected)
-				}
-				aggregatedEvents := store.GetAggregatedEvents()
-				assert.Len(t, aggregatedEvents, len(allExpected))
-				assert.ElementsMatch(t, aggregatedEvents, allExpected)
-			})
-		}
-	}
-}
-
-func TestFlowAggregationOfUnknownProtocols(t *testing.T) {
-	var tests = []struct {
-		description string
-		addresses   [][]netip.Addr
-		dstPort     uint16
-		eventTypes  []types.Type
-	}{
-		{
-			description: "start and stop",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
-		},
-		{
-			description: "start and drop",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
-		},
-		{
-			description: "start only",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeStart},
-		},
-		{
-			description: "drop only",
-			addresses:   [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeDrop},
-		}}
-
-	for _, tt := range tests {
-		t.Run(tt.description+" "+types.ProtocolUnknown.String(), func(t *testing.T) {
-			store := NewAggregatingMemoryStore()
-			store.WindowEnd = time.Now().Add(5 * time.Second)
-
-			allExpected := make([]*types.Event, 0)
-
-			for _, srcAndDst := range tt.addresses {
-				inEvents, expected := generateEventsForUnknownProtocol(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, types.ProtocolUnknown, types.Ingress, store.WindowStart, store.WindowEnd)
-				for _, e := range inEvents {
-					store.StoreEvent(e)
-				}
-				allExpected = append(allExpected, expected...)
-			}
-
-			events := store.GetAggregatedEvents()
-			assert.ElementsMatch(t, events, allExpected)
-		})
-	}
-}
-
-func TestResetAggregationWindow(t *testing.T) {
-	store := NewAggregatingMemoryStore()
-	store.StoreEvent(&types.Event{
-		ID:        uuid.New(),
-		Timestamp: time.Now(),
-		EventFields: types.EventFields{
-			FlowID:           uuid.New(),
-			Type:             types.TypeStart,
-			Protocol:         types.TCP,
-			RuleID:           []byte("rule-id-1"),
-			Direction:        types.Ingress,
-			SourceIP:         netip.MustParseAddr("1.1.1.1"),
-			SourcePort:       1234,
-			DestIP:           netip.MustParseAddr("2.2.2.2"),
-			DestPort:         5678,
-			SourceResourceID: []byte("source-resource-id"),
-			DestResourceID:   []byte("dest-resource-id"),
-			RxPackets:        random.Uint64(),
-			TxPackets:        random.Uint64(),
-			RxBytes:          random.Uint64(),
-			TxBytes:          random.Uint64(),
-		},
-	})
-
-	reset := store.ResetAggregationWindow()
-	previousEvents, ok := reset.(*AggregatingMemory)
-	assert.True(t, ok)
-	assert.NotEqual(t, previousEvents.WindowStart, store.WindowStart)
-	assert.Equal(t, previousEvents.WindowEnd, store.WindowStart)
-	assert.NotEmpty(t, previousEvents.events)
-	assert.Empty(t, store.events)
-}
-
-func generateEvents(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol,
-	direction types.Direction, icmpType uint8, windowStart, windowEnd time.Time) ([]*types.Event, *types.Event) {
-	var rxPackets, txPackets, rxBytes, txBytes uint64
-	inEvents := make([]*types.Event, 0)
-	ts := time.Now()
-	flowId := uuid.New()
-	srcPort := uint16(random.Uint32() >> 16)
-
-	for idx, eventType := range eventTypes {
-		e := &types.Event{
-			ID:        uuid.New(),
-			Timestamp: ts.Add(time.Duration(idx) * time.Second),
-			EventFields: types.EventFields{
-				FlowID:           flowId,
-				Type:             eventType,
-				Protocol:         protocol,
-				RuleID:           []byte("rule-id-1"),
-				Direction:        direction,
-				SourceIP:         srcIp,
-				SourcePort:       srcPort,
-				DestIP:           dstIp,
-				DestPort:         dstPort,
-				SourceResourceID: []byte("source-resource-id"),
-				DestResourceID:   []byte("dest-resource-id"),
-				RxPackets:        random.Uint64(),
-				TxPackets:        random.Uint64(),
-				RxBytes:          random.Uint64(),
-				TxBytes:          random.Uint64(),
-			}}
-		rxBytes += e.RxBytes
-		txBytes += e.TxBytes
-		rxPackets += e.RxPackets
-		txPackets += e.TxPackets
-		inEvents = append(inEvents, e)
-		if protocol == types.ICMP || protocol == types.ICMPv6 {
-			e.ICMPType = icmpType
-		}
-	}
-
-	var start, end, drop uint64
-	for _, eventType := range eventTypes {
-		switch eventType {
-		case types.TypeStart:
-			start += 1
-		case types.TypeDrop:
-			drop += 1
-		case types.TypeEnd:
-			end += 1
-		}
-	}
-	aggregatedEvent := &types.Event{
-		ID:          inEvents[0].ID,
-		Timestamp:   inEvents[0].Timestamp,
-		WindowStart: windowStart,
-		WindowEnd:   windowEnd,
-		EventFields: types.EventFields{
-			FlowID:           flowId,
-			Type:             types.TypeUnknown,
-			Protocol:         inEvents[0].Protocol,
-			RuleID:           []byte("rule-id-1"),
-			Direction:        inEvents[0].Direction,
-			SourceIP:         srcIp,
-			SourcePort:       srcPort,
-			DestIP:           dstIp,
-			DestPort:         dstPort,
-			SourceResourceID: []byte("source-resource-id"),
-			DestResourceID:   []byte("dest-resource-id"),
-			RxPackets:        rxPackets,
-			TxPackets:        txPackets,
-			RxBytes:          rxBytes,
-			TxBytes:          txBytes,
-			NumOfStarts:      start,
-			NumOfEnds:        end,
-			NumOfDrops:       drop,
-		}}
-	if protocol == types.ICMP || protocol == types.ICMPv6 {
-		aggregatedEvent.ICMPType = icmpType
-	}
-
-	return inEvents, aggregatedEvent
-}
-
-func generateEventsForUnknownProtocol(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol,
-	direction types.Direction, windowStart, windowEnd time.Time) ([]*types.Event, []*types.Event) {
-	inEvents := make([]*types.Event, 0)
-	expectedEvents := make([]*types.Event, 0)
-
-	ts := time.Now()
-	flowId := uuid.New()
-	srcPort := uint16(random.Uint32() >> 16)
-
-	for idx, eventType := range eventTypes {
-		e := &types.Event{
-			ID:        uuid.New(),
-			Timestamp: ts.Add(time.Duration(idx) * time.Second),
-			EventFields: types.EventFields{
-				FlowID:           flowId,
-				Type:             eventType,
-				Protocol:         protocol,
-				RuleID:           []byte("rule-id-1"),
-				Direction:        direction,
-				SourceIP:         srcIp,
-				SourcePort:       srcPort,
-				DestIP:           dstIp,
-				DestPort:         dstPort,
-				SourceResourceID: []byte("source-resource-id"),
-				DestResourceID:   []byte("dest-resource-id"),
-				RxPackets:        random.Uint64(),
-				TxPackets:        random.Uint64(),
-				RxBytes:          random.Uint64(),
-				TxBytes:          random.Uint64(),
-			}}
-		inEvents = append(inEvents, e)
-
-		var start, end, drop uint64
-		switch eventType {
-		case types.TypeStart:
-			start = 1
-		case types.TypeDrop:
-			drop = 1
-		case types.TypeEnd:
-			end = 1
-		}
-
-		expectedEvents = append(expectedEvents, &types.Event{
-			ID:          e.ID,
-			Timestamp:   e.Timestamp,
-			WindowStart: windowStart,
-			WindowEnd:   windowEnd,
-			EventFields: types.EventFields{
-				FlowID:           flowId,
-				Type:             types.TypeUnknown,
-				Protocol:         e.Protocol,
-				RuleID:           []byte("rule-id-1"),
-				Direction:        e.Direction,
-				SourceIP:         srcIp,
-				SourcePort:       srcPort,
-				DestIP:           dstIp,
-				DestPort:         dstPort,
-				SourceResourceID: []byte("source-resource-id"),
-				DestResourceID:   []byte("dest-resource-id"),
-				RxPackets:        e.RxPackets,
-				TxPackets:        e.TxPackets,
-				RxBytes:          e.RxBytes,
-				TxBytes:          e.TxBytes,
-				NumOfStarts:      start,
-				NumOfEnds:        end,
-				NumOfDrops:       drop,
-			}})
-	}
-
-	return inEvents, expectedEvents
-}

client/internal/netflow/store/memory.go

@@ -1,15 +1,10 @@
 package store
 
 import (
-	"maps"
-	"math/rand"
-	v2 "math/rand/v2"
-	"net/netip"
-	"slices"
 	"sync"
-	"time"
 
 	"github.com/google/uuid"
+
 	"github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
@@ -24,13 +19,6 @@ type Memory struct {
 	events map[uuid.UUID]*types.Event
 }
 
-type AggregatingMemory struct {
-	Memory
-	WindowStart time.Time
-	WindowEnd   time.Time
-	rnd         *v2.PCG
-}
-
 func (m *Memory) StoreEvent(event *types.Event) {
 	m.mux.Lock()
 	defer m.mux.Unlock()
@@ -60,95 +48,3 @@ func (m *Memory) DeleteEvents(ids []uuid.UUID) {
 		delete(m.events, id)
 	}
 }
-
-func NewAggregatingMemoryStore() *AggregatingMemory {
-	return &AggregatingMemory{WindowStart: time.Now(), Memory: Memory{events: make(map[uuid.UUID]*types.Event)}, rnd: v2.NewPCG(rand.Uint64(), rand.Uint64())}
-}
-
-func (am *AggregatingMemory) ResetAggregationWindow() types.FlowEventAggregator {
-	am.mux.Lock()
-	defer am.mux.Unlock()
-
-	now := time.Now()
-	toret := AggregatingMemory{WindowStart: am.WindowStart, WindowEnd: now, Memory: Memory{events: am.events}, rnd: v2.NewPCG(rand.Uint64(), rand.Uint64())}
-
-	am.events = make(map[uuid.UUID]*types.Event)
-	am.WindowStart = now
-
-	return &toret
-}
-
-type aggregationKey struct {
-	srcAddr   netip.Addr
-	destAddr  netip.Addr
-	destPort  uint16
-	direction int
-	protocol  uint8
-	icmpType  uint8
-	unique    uint64 // used to prevent aggregation on non icmp/udp/tcp events
-}
-
-func (am *AggregatingMemory) GetAggregatedEvents() []*types.Event {
-	am.mux.Lock()
-	defer am.mux.Unlock()
-
-	aggregated := make(map[aggregationKey]*types.Event)
-	for _, v := range am.events {
-		lookupKey := aggregationKey{srcAddr: v.SourceIP, destAddr: v.DestIP, destPort: v.DestPort, direction: int(v.Direction), protocol: uint8(v.Protocol), icmpType: v.ICMPType}
-		if _, ok := aggregated[lookupKey]; !ok {
-			event := v.Clone()
-
-			switch event.Type {
-			case types.TypeStart:
-				event.NumOfStarts += 1
-			case types.TypeDrop:
-				event.NumOfDrops += 1
-			case types.TypeEnd:
-				event.NumOfEnds += 1
-			}
-			event.Type = types.TypeUnknown
-
-			// Please note that ICMPCode field isn't propagated by the manager (see flow/proto/flow.pb.go, FlowFields struct)
-			// so the field value in an icmp event in the "aggregated" doesn't matter
-
-			event.WindowStart = am.WindowStart
-			event.WindowEnd = am.WindowEnd
-
-			if event.Protocol != types.ICMP && event.Protocol != types.ICMPv6 && event.Protocol != types.UDP && event.Protocol != types.TCP {
-				lookupKey.unique = am.rnd.Uint64() // to make the lookup key unique so we don't aggregate on it
-			}
-
-			aggregated[lookupKey] = event
-			continue
-		}
-
-		aggregatedEvent := aggregated[lookupKey]
-		if aggregatedEvent.Protocol != types.ICMP && aggregatedEvent.Protocol != types.ICMPv6 && aggregatedEvent.Protocol != types.UDP && aggregatedEvent.Protocol != types.TCP {
-			continue // we don't aggregate this type of events; shouldn't ever get here
-		}
-
-		// track the number of connections, duration?, open and close events?
-		aggregatedEvent.RxBytes += v.RxBytes
-		aggregatedEvent.RxPackets += v.RxPackets
-		aggregatedEvent.TxBytes += v.TxBytes
-		aggregatedEvent.TxPackets += v.TxPackets
-		switch v.Type {
-		case types.TypeStart:
-			aggregatedEvent.NumOfStarts += 1
-		case types.TypeDrop:
-			aggregatedEvent.NumOfDrops += 1
-		case types.TypeEnd:
-			aggregatedEvent.NumOfEnds += 1
-		}
-		if aggregatedEvent.Timestamp.Compare(v.Timestamp) > 0 {
-			aggregatedEvent.Timestamp = v.Timestamp
-			aggregatedEvent.ID = v.ID
-			aggregatedEvent.SourcePort = v.SourcePort
-		}
-		if len(aggregatedEvent.RuleID) == 0 && len(v.RuleID) != 0 {
-			aggregatedEvent.RuleID = slices.Clone(v.RuleID)
-		}
-	}
-
-	return slices.Collect(maps.Values(aggregated)) // could return an iterator instead here
-}

client/internal/netflow/types/types.go

@@ -2,7 +2,6 @@ package types
 
 import (
 	"net/netip"
-	"slices"
 	"strconv"
 	"time"
 
@@ -70,10 +69,8 @@ const (
 )
 
 type Event struct {
-	ID          uuid.UUID
-	Timestamp   time.Time
-	WindowStart time.Time
-	WindowEnd   time.Time
+	ID        uuid.UUID
+	Timestamp time.Time
 	EventFields
 }
 
@@ -95,17 +92,6 @@ type EventFields struct {
 	TxPackets        uint64
 	RxBytes          uint64
 	TxBytes          uint64
-	NumOfStarts      uint64
-	NumOfEnds        uint64
-	NumOfDrops       uint64
-}
-
-func (e *Event) Clone() *Event {
-	toret := *e
-	toret.RuleID = slices.Clone(e.RuleID)
-	toret.SourceResourceID = slices.Clone(e.SourceResourceID)
-	toret.DestResourceID = slices.Clone(e.DestResourceID)
-	return &toret
 }
 
 type FlowConfig struct {
@@ -128,15 +114,13 @@ type FlowManager interface {
 	GetLogger() FlowLogger
 }
 
-type FlowEventAggregator interface {
-	ResetAggregationWindow() FlowEventAggregator
-	GetAggregatedEvents() []*Event
-}
-
 type FlowLogger interface {
-	ResetAggregationWindow() FlowEventAggregator
 	// StoreEvent stores a flow event
 	StoreEvent(flowEvent EventFields)
+	// GetEvents returns all stored events
+	GetEvents() []*Event
+	// DeleteEvents deletes events from the store
+	DeleteEvents([]uuid.UUID)
 	// Close closes the logger
 	Close()
 	// Enable enables the flow logger receiver
@@ -156,11 +140,6 @@ type Store interface {
 	Close()
 }
 
-type AggregatingStore interface {
-	FlowEventAggregator
-	Store
-}
-
 // ConnTracker defines the interface for connection tracking functionality
 type ConnTracker interface {
 	// Start begins tracking connections by listening for conntrack events.

client/internal/peer/guard/guard.go

@@ -85,11 +85,7 @@ func (g *Guard) reconnectLoopWithRetry(ctx context.Context, callback func()) {
 	defer g.srWatcher.RemoveListener(srReconnectedChan)
 
 	ticker := g.initialTicker(ctx)
-	defer func() {
-		// If backoff.Ticker.send is blocked, context.Done will not close the Ticker goroutine.
-		// We have to explicitly call Stop, even if we use backoff.WithContext.
-		ticker.Stop()
-	}()
+	defer ticker.Stop()
 
 	tickerChannel := ticker.C
 

client/internal/peer/guard/guard_leak_test.go

@@ -1,92 +0,0 @@
-package guard
-
-import (
-	"context"
-	"runtime"
-	"strings"
-	"sync"
-	"testing"
-	"time"
-
-	log "github.com/sirupsen/logrus"
-
-	"github.com/netbirdio/netbird/client/internal/peer/ice"
-)
-
-func newTestGuard(status connStatusFunc) *Guard {
-	srw := NewSRWatcher(nil, nil, nil, ice.Config{})
-	return NewGuard(log.WithField("test", "guard"), status, 50*time.Millisecond, srw)
-}
-
-// countBackoffTickerGoroutines returns how many goroutines are currently sitting
-// in backoff/v4.(*Ticker).run (a ticker goroutine that has not exited).
-func countBackoffTickerGoroutines() int {
-	buf := make([]byte, 1<<25) // 32MB
-	n := runtime.Stack(buf, true)
-	return strings.Count(string(buf[:n]), "backoff/v4.(*Ticker).run")
-}
-
-// TestGuard_ReconnectTicker_NoGoroutineLeakOnShutdown reproduces a observed
-// leak: after a shutdown burst, ticker run/send goroutines stay parked
-// forever even though every reconnect loop has exited.
-func TestGuard_ReconnectTicker_NoGoroutineLeakOnShutdown(t *testing.T) {
-	before := countBackoffTickerGoroutines()
-
-	const peers = 6000
-	cancels := make([]context.CancelFunc, 0, peers)
-	var wg sync.WaitGroup
-
-	// A status check slower than the tick cadence. This models the real
-	// isConnectedOnAllWay/callback doing work: while the loop is busy in the
-	// handler, the ticker fires the next tick and parks in send(), because
-	// send() never selects on ctx.
-	slowStatus := func() ConnStatus {
-		time.Sleep(70 * time.Millisecond)
-		return ConnStatusConnected
-	}
-
-	for range peers {
-		g := newTestGuard(slowStatus)
-		ctx, cancel := context.WithCancel(context.Background())
-		cancels = append(cancels, cancel)
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			g.Start(ctx, func() {})
-		}()
-		// Force the live ticker to be a newReconnectTicker.
-		g.SetRelayedConnDisconnected()
-	}
-
-	// Let the replacement tickers get past their 800ms initial interval, so
-	// many are parked in send() waiting on the (slow) consumer when we tear
-	// everything down.
-	time.Sleep(1500 * time.Millisecond)
-
-	// Shutdown burst: cancel every peer at once, like engine teardown.
-	for _, c := range cancels {
-		c()
-	}
-
-	// Every reconnect loop must return
-	waitCh := make(chan struct{})
-	go func() { wg.Wait(); close(waitCh) }()
-	select {
-	case <-waitCh:
-	case <-time.After(30 * time.Second):
-		t.Fatal("not all reconnect loops returned after ctx cancel")
-	}
-
-	// Give any correctly-stopped ticker goroutines time to unwind.
-	for range 50 {
-		runtime.Gosched()
-		time.Sleep(10 * time.Millisecond)
-	}
-
-	leaked := countBackoffTickerGoroutines() - before
-	t.Logf("backoff Ticker.run goroutines still parked after teardown of %d peers: %d", peers, leaked)
-	if leaked > 0 {
-		t.Errorf("LEAK: %d backoff ticker goroutines parked after all reconnect loops exited "+
-			"(defer ticker.Stop() stops the initial ticker, not the live replacement)", leaked)
-	}
-}

client/internal/profilemanager/config.go

@@ -101,6 +101,8 @@ type ConfigInput struct {
 
 	DNSLabels domain.List
 
+	LazyConnectionEnabled *bool
+
 	MTU *uint16
 }
 
@@ -178,9 +180,7 @@ type Config struct {
 
 	ClientCertKeyPair *tls.Certificate `json:"-"`
 
-	// LazyConnection is the MDM-managed lazy-connection override ("on"/"off"/"").
-	// Runtime-only: re-derived from MDM policy on each load, never persisted.
-	LazyConnection string `json:"-"`
+	LazyConnectionEnabled bool
 
 	MTU uint16
 
@@ -632,6 +632,12 @@ func (config *Config) apply(input ConfigInput) (updated bool, err error) {
 		updated = true
 	}
 
+	if input.LazyConnectionEnabled != nil && *input.LazyConnectionEnabled != config.LazyConnectionEnabled {
+		log.Infof("switching lazy connection to %t", *input.LazyConnectionEnabled)
+		config.LazyConnectionEnabled = *input.LazyConnectionEnabled
+		updated = true
+	}
+
 	if input.MTU != nil && *input.MTU != config.MTU {
 		log.Infof("updating MTU to %d (old value %d)", *input.MTU, config.MTU)
 		config.MTU = *input.MTU
@@ -722,15 +728,6 @@ func (config *Config) applyMDMPolicy(policy *mdm.Policy) {
 			log.Warnf("MDM wireguard port %d out of range [1,65535]; keeping previous value", v)
 		}
 	}
-
-	if v, ok := policy.GetBool(mdm.KeyLazyConnection); ok {
-		state := "off"
-		if v {
-			state = "on"
-		}
-		config.LazyConnection = state
-		logApplied(mdm.KeyLazyConnection, state)
-	}
 }
 
 // parseURL parses and validates the URL for the named service. The URL

client/internal/profilemanager/config_mdm_test.go

@@ -130,37 +130,6 @@ func TestApply_MDMBoolKeysOverrideOnDiskValue(t *testing.T) {
 	assert.True(t, cfg.Policy().HasKey(mdm.KeyRosenpassEnabled))
 }
 
-func TestApply_MDMLazyConnection(t *testing.T) {
-	cases := []struct {
-		name string
-		raw  any
-		want string
-	}{
-		{"native true", true, "on"},
-		{"native false", false, "off"},
-		{"string on", "on", "on"},
-		{"string off", "off", "off"},
-		{"string yes", "yes", "on"},
-		{"string no", "no", "off"},
-	}
-	for _, c := range cases {
-		t.Run(c.name, func(t *testing.T) {
-			withMDMPolicy(t, mdm.NewPolicy(map[string]any{
-				mdm.KeyLazyConnection: c.raw,
-			}))
-
-			cfg, err := UpdateOrCreateConfig(ConfigInput{
-				ConfigPath: filepath.Join(t.TempDir(), "config.json"),
-			})
-			require.NoError(t, err)
-			require.NotNil(t, cfg)
-
-			assert.Equal(t, c.want, cfg.LazyConnection)
-			assert.True(t, cfg.Policy().HasKey(mdm.KeyLazyConnection))
-		})
-	}
-}
-
 func TestApply_MDMPreSharedKeyRedactionSentinelRejected(t *testing.T) {
 	const maskSentinel = "**********"
 

client/ios/NetBirdSDK/env_list.go

@@ -38,7 +38,7 @@ func GetEnvKeyNBForceRelay() string {
 
 // GetEnvKeyNBLazyConn Exports the environment variable for the iOS client
 func GetEnvKeyNBLazyConn() string {
-	return lazyconn.EnvLazyConn
+	return lazyconn.EnvEnableLazyConn
 }
 
 // GetEnvKeyNBInactivityThreshold Exports the environment variable for the iOS client

client/mdm/canonical_loaders.go

@@ -28,7 +28,6 @@ var allKeys = []string{
 	KeyWireguardPort,
 	KeySplitTunnelMode,
 	KeySplitTunnelApps,
-	KeyLazyConnection,
 }
 
 // canonicalKey maps the lowercase form of a managed-config value name to

client/mdm/policy.go

@@ -11,7 +11,6 @@ package mdm
 import (
 	"sort"
 	"strconv"
-	"strings"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -49,11 +48,6 @@ const (
 	// construction — only one mode can be set at a time.
 	KeySplitTunnelMode = "splitTunnelMode"
 	KeySplitTunnelApps = "splitTunnelApps"
-
-	// KeyLazyConnection forces the lazy-connection feature on or off, overriding
-	// the management feature flag. Read as a bool (native bool, or on/off,
-	// true/false, 1/0, yes/no); absent = defer to management.
-	KeyLazyConnection = "lazyConnection"
 )
 
 // Split-tunnel mode literals (KeySplitTunnelMode values).
@@ -75,13 +69,12 @@ var boolStringLiterals = map[string]bool{
 	"true":  true,
 	"1":     true,
 	"yes":   true,
-	"on":    true,
 	"false": false,
 	"0":     false,
 	"no":    false,
-	"off":   false,
 }
 
+
 // Policy holds MDM-managed settings read from the platform source. A nil or
 // empty Policy means no enforcement is active.
 type Policy struct {
@@ -164,8 +157,7 @@ func (p *Policy) GetString(key string) (string, bool) {
 }
 
 // GetBool returns the managed value for key coerced to bool, and whether the
-// key was set. Accepts native bool and string literals (true/false, 1/0,
-// yes/no, on/off), case-insensitively and trimmed of surrounding whitespace.
+// key was set. Accepts native bool and string literals "true"/"false"/"1"/"0".
 func (p *Policy) GetBool(key string) (bool, bool) {
 	if p == nil {
 		return false, false
@@ -178,7 +170,7 @@ func (p *Policy) GetBool(key string) (bool, bool) {
 	case bool:
 		return t, true
 	case string:
-		b, known := boolStringLiterals[strings.ToLower(strings.TrimSpace(t))]
+		b, known := boolStringLiterals[t]
 		return b, known
 	case int:
 		return t != 0, true

client/mdm/policy_test.go

@@ -31,8 +31,8 @@ func TestPolicy_Empty(t *testing.T) {
 
 func TestPolicy_HasKey(t *testing.T) {
 	p := NewPolicy(map[string]any{
-		KeyManagementURL:   "https://corp.example.com",
-		KeyDisableProfiles: true,
+		KeyManagementURL:    "https://corp.example.com",
+		KeyDisableProfiles:  true,
 	})
 	assert.False(t, p.IsEmpty())
 	assert.True(t, p.HasKey(KeyManagementURL))
@@ -53,8 +53,8 @@ func TestPolicy_ManagedKeysSorted(t *testing.T) {
 func TestPolicy_GetString(t *testing.T) {
 	p := NewPolicy(map[string]any{
 		KeyManagementURL:   "https://corp.example.com",
-		KeyDisableProfiles: true, // wrong type for GetString
-		KeyPreSharedKey:    "",   // empty rejected
+		KeyDisableProfiles: true,            // wrong type for GetString
+		KeyPreSharedKey:        "",              // empty rejected
 	})
 	v, ok := p.GetString(KeyManagementURL)
 	assert.True(t, ok)
@@ -85,11 +85,6 @@ func TestPolicy_GetBool(t *testing.T) {
 		{"string 0", "0", false, true},
 		{"string yes", "yes", true, true},
 		{"string no", "no", false, true},
-		{"string on", "on", true, true},
-		{"string off", "off", false, true},
-		{"mixed case On", "On", true, true},
-		{"upper TRUE", "TRUE", true, true},
-		{"padded yes", "  yes  ", true, true},
 		{"int nonzero", 1, true, true},
 		{"int zero", 0, false, true},
 		{"int64 nonzero", int64(2), true, true},

client/proto/daemon_gateway_test.go

@@ -1,80 +0,0 @@
-package proto
-
-import (
-	"context"
-	"net"
-	"net/http"
-	"net/http/httptest"
-	"strings"
-	"testing"
-
-	gatewayruntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/test/bufconn"
-)
-
-func TestGatewayServerRoutesCoverDaemonRPCs(t *testing.T) {
-	mux := gatewayruntime.NewServeMux()
-	if err := RegisterDaemonServiceHandlerServer(context.Background(), mux, UnimplementedDaemonServiceServer{}); err != nil {
-		t.Fatalf("register daemon gateway server handlers: %v", err)
-	}
-
-	assertAllDaemonGatewayRoutesRegistered(t, mux)
-}
-
-func TestGatewayClientRoutesCoverDaemonRPCs(t *testing.T) {
-	listener := bufconn.Listen(1024 * 1024)
-	server := grpc.NewServer()
-	RegisterDaemonServiceServer(server, UnimplementedDaemonServiceServer{})
-	go func() {
-		if err := server.Serve(listener); err != nil && err != grpc.ErrServerStopped {
-			t.Errorf("serve bufconn gRPC server: %v", err)
-		}
-	}()
-	t.Cleanup(func() {
-		server.Stop()
-		_ = listener.Close()
-	})
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	mux := gatewayruntime.NewServeMux()
-	opts := []grpc.DialOption{
-		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
-			return listener.Dial()
-		}),
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-	}
-	if err := RegisterDaemonServiceHandlerFromEndpoint(ctx, mux, "passthrough:///bufnet", opts); err != nil {
-		t.Fatalf("register daemon gateway client handlers: %v", err)
-	}
-
-	assertAllDaemonGatewayRoutesRegistered(t, mux)
-}
-
-func assertAllDaemonGatewayRoutesRegistered(t *testing.T, mux http.Handler) {
-	t.Helper()
-	for _, method := range DaemonService_ServiceDesc.Methods {
-		assertGatewayRouteRegistered(t, mux, method.MethodName)
-	}
-	for _, stream := range DaemonService_ServiceDesc.Streams {
-		assertGatewayRouteRegistered(t, mux, stream.StreamName)
-	}
-}
-
-func assertGatewayRouteRegistered(t *testing.T, mux http.Handler, methodName string) {
-	t.Helper()
-
-	path := "/daemon.DaemonService/" + methodName
-	req := httptest.NewRequest(http.MethodPost, path, strings.NewReader("{}"))
-	req.Header.Set("Content-Type", "application/json")
-	res := httptest.NewRecorder()
-
-	mux.ServeHTTP(res, req)
-
-	if res.Code == http.StatusNotFound {
-		t.Fatalf("gateway route for %s is not registered", methodName)
-	}
-}

client/proto/generate.sh

@@ -12,11 +12,5 @@ script_path=$(dirname "$(realpath "$0")")
 cd "$script_path"
 go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.36.6
 go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.6.1
-go install github.com/grpc-ecosystem/grpc-gateway/v2/protoc-gen-grpc-gateway@v2.26.3
-protoc -I ./ ./daemon.proto \
-  --go_out=../ \
-  --go-grpc_out=../ \
-  --grpc-gateway_out=../ \
-  --grpc-gateway_opt=generate_unbound_methods=true \
-  --experimental_allow_proto3_optional
+protoc -I ./ ./daemon.proto --go_out=../ --go-grpc_out=../ --experimental_allow_proto3_optional
 cd "$old_pwd"

client/server/mdm.go

@@ -3,7 +3,6 @@ package server
 import (
 	"context"
 	"fmt"
-	"net/url"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -156,6 +155,7 @@ func (s *Server) restartEngineForMDMLocked() error {
 	s.config = config
 	s.statusRecorder.UpdateManagementAddress(config.ManagementURL.String())
 	s.statusRecorder.UpdateRosenpass(config.RosenpassEnabled, config.RosenpassPermissive)
+	s.statusRecorder.UpdateLazyConnection(config.LazyConnectionEnabled)
 
 	ctx, cancel := context.WithCancel(s.rootCtx)
 	s.actCancel = cancel
@@ -185,37 +185,6 @@ func conflictBool(key string, p *bool) conflictCheck {
 	}
 }
 
-func canonicalURL(s string) string {
-	u, err := url.ParseRequestURI(s)
-	if err != nil {
-		return s
-	}
-	if u.Port() == "" {
-		switch u.Scheme {
-		case "https":
-			u.Host += ":443"
-		case "http":
-			u.Host += ":80"
-		}
-	}
-	return u.String()
-}
-
-// conflictURL is conflictString for URL-typed keys: both sides are
-// normalized via canonicalURL before comparison.
-func conflictURL(key, got string) conflictCheck {
-	return conflictCheck{
-		key: key,
-		check: func(pol *mdm.Policy) bool {
-			if got == "" {
-				return true
-			}
-			want, ok := pol.GetString(key)
-			return ok && canonicalURL(want) == canonicalURL(got)
-		},
-	}
-}
-
 // conflictString builds a conflictCheck for a string MDM key. An empty
 // `got` is treated as "field not set" (no override requested); otherwise
 // the check returns true only when the policy contains the key and its
@@ -291,7 +260,7 @@ func mdmManagedFieldConflicts(msg *proto.SetConfigRequest, policy *mdm.Policy) [
 	}
 
 	return resolveConflicts(policy, []conflictCheck{
-		conflictURL(mdm.KeyManagementURL, msg.ManagementUrl),
+		conflictString(mdm.KeyManagementURL, msg.ManagementUrl),
 		conflictString(mdm.KeyPreSharedKey, pskGot),
 		conflictBool(mdm.KeyRosenpassEnabled, msg.RosenpassEnabled),
 		conflictBool(mdm.KeyRosenpassPermissive, msg.RosenpassPermissive),
@@ -339,6 +308,7 @@ func setConfigRequestHasConfigOverrides(msg *proto.SetConfigRequest) bool {
 		msg.DisableFirewall != nil ||
 		msg.BlockLanAccess != nil ||
 		msg.DisableNotifications != nil ||
+		msg.LazyConnectionEnabled != nil ||
 		msg.BlockInbound != nil ||
 		msg.DisableIpv6 != nil ||
 		msg.EnableSSHRoot != nil ||
@@ -381,6 +351,7 @@ func loginRequestHasConfigOverrides(msg *proto.LoginRequest) bool {
 		msg.BlockLanAccess != nil ||
 		msg.DisableNotifications != nil ||
 		len(msg.DnsLabels) > 0 || msg.CleanDNSLabels ||
+		msg.LazyConnectionEnabled != nil ||
 		msg.BlockInbound != nil
 }
 
@@ -412,7 +383,7 @@ func loginRequestMDMConflicts(msg *proto.LoginRequest, policy *mdm.Policy) []str
 	}
 
 	return resolveConflicts(policy, []conflictCheck{
-		conflictURL(mdm.KeyManagementURL, msg.ManagementUrl),
+		conflictString(mdm.KeyManagementURL, msg.ManagementUrl),
 		conflictString(mdm.KeyPreSharedKey, pskGot),
 		conflictBool(mdm.KeyRosenpassEnabled, msg.RosenpassEnabled),
 		conflictBool(mdm.KeyRosenpassPermissive, msg.RosenpassPermissive),

client/server/server.go

@@ -238,6 +238,7 @@ func (s *Server) Start() error {
 
 	s.statusRecorder.UpdateManagementAddress(config.ManagementURL.String())
 	s.statusRecorder.UpdateRosenpass(config.RosenpassEnabled, config.RosenpassPermissive)
+	s.statusRecorder.UpdateLazyConnection(config.LazyConnectionEnabled)
 
 	if s.sessionWatcher == nil {
 		s.sessionWatcher = internal.NewSessionWatcher(s.rootCtx, s.statusRecorder)
@@ -421,6 +422,16 @@ func (s *Server) SetConfig(callerCtx context.Context, msg *proto.SetConfigReques
 		return nil, fmt.Errorf("failed to update profile config: %w", err)
 	}
 
+	// Apply the lazy connection toggle to the running engine so it takes
+	// effect without a down/up. s.mutex is already held.
+	if msg.LazyConnectionEnabled != nil && s.connectClient != nil {
+		if engine := s.connectClient.Engine(); engine != nil {
+			if err := engine.SetLazyConnEnabled(msg.GetLazyConnectionEnabled()); err != nil {
+				log.Errorf("failed to apply lazy connection change at runtime: %v", err)
+			}
+		}
+	}
+
 	return &proto.SetConfigResponse{}, nil
 }
 
@@ -499,6 +510,7 @@ func (s *Server) setConfigInputFromRequest(msg *proto.SetConfigRequest) (profile
 	config.DisableFirewall = msg.DisableFirewall
 	config.BlockLANAccess = msg.BlockLanAccess
 	config.DisableNotifications = msg.DisableNotifications
+	config.LazyConnectionEnabled = msg.LazyConnectionEnabled
 	config.BlockInbound = msg.BlockInbound
 	config.DisableIPv6 = msg.DisableIpv6
 	config.EnableSSHRoot = msg.EnableSSHRoot
@@ -1969,6 +1981,7 @@ func (s *Server) GetConfig(ctx context.Context, req *proto.GetConfigRequest) (*p
 		ServerSSHAllowed:              *cfg.ServerSSHAllowed,
 		RosenpassEnabled:              cfg.RosenpassEnabled,
 		RosenpassPermissive:           cfg.RosenpassPermissive,
+		LazyConnectionEnabled:         cfg.LazyConnectionEnabled,
 		BlockInbound:                  cfg.BlockInbound,
 		DisableNotifications:          disableNotifications,
 		NetworkMonitor:                networkMonitor,

client/server/setconfig_mdm_test.go

@@ -181,43 +181,6 @@ func TestSetConfig_MDMAllow_NonManagedFields(t *testing.T) {
 	require.NotNil(t, resp)
 }
 
-// TestSetConfig_MDMAllow_ManagementURLPortNormalized covers the
-// regression from discussion #6483: MDM URL without explicit port vs
-// UI echo with the parseURL-appended default port must be treated as
-// a no-op echo, not a conflict.
-func TestSetConfig_MDMAllow_ManagementURLPortNormalized(t *testing.T) {
-	tests := []struct {
-		name      string
-		mdmURL    string
-		submitURL string
-	}{
-		{"policy_no_port_submit_with_443", "https://netbird.corp.example", "https://netbird.corp.example:443"},
-		{"policy_with_443_submit_no_port", "https://netbird.corp.example:443", "https://netbird.corp.example"},
-		{"http_policy_no_port_submit_with_80", "http://netbird.corp.example", "http://netbird.corp.example:80"},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			withMDMPolicy(t, mdm.NewPolicy(map[string]any{
-				mdm.KeyManagementURL: tc.mdmURL,
-			}))
-
-			s, ctx, profName, username, _ := setupServerWithProfile(t)
-
-			rosenpassEnabled := true
-			resp, err := s.SetConfig(ctx, &proto.SetConfigRequest{
-				ProfileName:      profName,
-				Username:         username,
-				ManagementUrl:    tc.submitURL,
-				RosenpassEnabled: &rosenpassEnabled,
-			})
-
-			require.NoError(t, err, "port-normalized URL echo must not trip MDM conflict gate")
-			require.NotNil(t, resp)
-		})
-	}
-}
-
 func TestSetConfig_MDMEmpty_NoEnforcement(t *testing.T) {
 	// No MDM policy active: any field can be written.
 	withMDMPolicy(t, mdm.NewPolicy(nil))

client/server/setconfig_test.go

@@ -69,41 +69,43 @@ func TestSetConfig_AllFieldsSaved(t *testing.T) {
 	disableFirewall := true
 	blockLANAccess := true
 	disableNotifications := true
+	lazyConnectionEnabled := true
 	blockInbound := true
 	disableIPv6 := true
 	mtu := int64(1280)
 	sshJWTCacheTTL := int32(300)
 
 	req := &proto.SetConfigRequest{
-		ProfileName:          profName,
-		Username:             currUser.Username,
-		ManagementUrl:        "https://new-api.netbird.io:443",
-		AdminURL:             "https://new-admin.netbird.io",
-		RosenpassEnabled:     &rosenpassEnabled,
-		RosenpassPermissive:  &rosenpassPermissive,
-		ServerSSHAllowed:     &serverSSHAllowed,
-		InterfaceName:        &interfaceName,
-		WireguardPort:        &wireguardPort,
-		OptionalPreSharedKey: &preSharedKey,
-		DisableAutoConnect:   &disableAutoConnect,
-		NetworkMonitor:       &networkMonitor,
-		DisableClientRoutes:  &disableClientRoutes,
-		DisableServerRoutes:  &disableServerRoutes,
-		DisableDns:           &disableDNS,
-		DisableFirewall:      &disableFirewall,
-		BlockLanAccess:       &blockLANAccess,
-		DisableNotifications: &disableNotifications,
-		BlockInbound:         &blockInbound,
-		DisableIpv6:          &disableIPv6,
-		NatExternalIPs:       []string{"1.2.3.4", "5.6.7.8"},
-		CleanNATExternalIPs:  false,
-		CustomDNSAddress:     []byte("1.1.1.1:53"),
-		ExtraIFaceBlacklist:  []string{"eth1", "eth2"},
-		DnsLabels:            []string{"label1", "label2"},
-		CleanDNSLabels:       false,
-		DnsRouteInterval:     durationpb.New(2 * time.Minute),
-		Mtu:                  &mtu,
-		SshJWTCacheTTL:       &sshJWTCacheTTL,
+		ProfileName:           profName,
+		Username:              currUser.Username,
+		ManagementUrl:         "https://new-api.netbird.io:443",
+		AdminURL:              "https://new-admin.netbird.io",
+		RosenpassEnabled:      &rosenpassEnabled,
+		RosenpassPermissive:   &rosenpassPermissive,
+		ServerSSHAllowed:      &serverSSHAllowed,
+		InterfaceName:         &interfaceName,
+		WireguardPort:         &wireguardPort,
+		OptionalPreSharedKey:  &preSharedKey,
+		DisableAutoConnect:    &disableAutoConnect,
+		NetworkMonitor:        &networkMonitor,
+		DisableClientRoutes:   &disableClientRoutes,
+		DisableServerRoutes:   &disableServerRoutes,
+		DisableDns:            &disableDNS,
+		DisableFirewall:       &disableFirewall,
+		BlockLanAccess:        &blockLANAccess,
+		DisableNotifications:  &disableNotifications,
+		LazyConnectionEnabled: &lazyConnectionEnabled,
+		BlockInbound:          &blockInbound,
+		DisableIpv6:           &disableIPv6,
+		NatExternalIPs:        []string{"1.2.3.4", "5.6.7.8"},
+		CleanNATExternalIPs:   false,
+		CustomDNSAddress:      []byte("1.1.1.1:53"),
+		ExtraIFaceBlacklist:   []string{"eth1", "eth2"},
+		DnsLabels:             []string{"label1", "label2"},
+		CleanDNSLabels:        false,
+		DnsRouteInterval:      durationpb.New(2 * time.Minute),
+		Mtu:                   &mtu,
+		SshJWTCacheTTL:        &sshJWTCacheTTL,
 	}
 
 	_, err = s.SetConfig(ctx, req)
@@ -138,6 +140,7 @@ func TestSetConfig_AllFieldsSaved(t *testing.T) {
 	require.Equal(t, blockLANAccess, cfg.BlockLANAccess)
 	require.NotNil(t, cfg.DisableNotifications)
 	require.Equal(t, disableNotifications, *cfg.DisableNotifications)
+	require.Equal(t, lazyConnectionEnabled, cfg.LazyConnectionEnabled)
 	require.Equal(t, blockInbound, cfg.BlockInbound)
 	require.Equal(t, disableIPv6, cfg.DisableIPv6)
 	require.Equal(t, []string{"1.2.3.4", "5.6.7.8"}, cfg.NATExternalIPs)
@@ -161,14 +164,13 @@ func verifyAllFieldsCovered(t *testing.T, req *proto.SetConfigRequest) {
 	t.Helper()
 
 	metadataFields := map[string]bool{
-		"state":                 true, // protobuf internal
-		"sizeCache":             true, // protobuf internal
-		"unknownFields":         true, // protobuf internal
-		"Username":              true, // metadata
-		"ProfileName":           true, // metadata
-		"CleanNATExternalIPs":   true, // control flag for clearing
-		"CleanDNSLabels":        true, // control flag for clearing
-		"LazyConnectionEnabled": true, // deprecated: proto field retained for compat, no longer applied
+		"state":               true, // protobuf internal
+		"sizeCache":           true, // protobuf internal
+		"unknownFields":       true, // protobuf internal
+		"Username":            true, // metadata
+		"ProfileName":         true, // metadata
+		"CleanNATExternalIPs": true, // control flag for clearing
+		"CleanDNSLabels":      true, // control flag for clearing
 	}
 
 	expectedFields := map[string]bool{
@@ -188,6 +190,7 @@ func verifyAllFieldsCovered(t *testing.T, req *proto.SetConfigRequest) {
 		"DisableFirewall":               true,
 		"BlockLanAccess":                true,
 		"DisableNotifications":          true,
+		"LazyConnectionEnabled":         true,
 		"BlockInbound":                  true,
 		"DisableIpv6":                   true,
 		"NatExternalIPs":                true,
@@ -249,6 +252,7 @@ func TestCLIFlags_MappedToSetConfig(t *testing.T) {
 		"block-lan-access":                  "BlockLanAccess",
 		"block-inbound":                     "BlockInbound",
 		"disable-ipv6":                      "DisableIpv6",
+		"enable-lazy-connection":            "LazyConnectionEnabled",
 		"external-ip-map":                   "NatExternalIPs",
 		"dns-resolver-address":              "CustomDNSAddress",
 		"extra-iface-blacklist":             "ExtraIFaceBlacklist",
@@ -265,8 +269,7 @@ func TestCLIFlags_MappedToSetConfig(t *testing.T) {
 
 	// SetConfigRequest fields that don't have CLI flags (settable only via UI or other means).
 	fieldsWithoutCLIFlags := map[string]bool{
-		"DisableNotifications":  true, // Only settable via UI
-		"LazyConnectionEnabled": true, // deprecated: no longer settable (managed by server + NB_LAZY_CONN)
+		"DisableNotifications": true, // Only settable via UI
 	}
 
 	// Get all SetConfigRequest fields to verify our map is complete.

client/system/info.go

@@ -74,6 +74,8 @@ type Info struct {
 	BlockInbound        bool
 	DisableIPv6         bool
 
+	LazyConnectionEnabled bool
+
 	EnableSSHRoot                 bool
 	EnableSSHSFTP                 bool
 	EnableSSHLocalPortForwarding  bool
@@ -85,7 +87,7 @@ func (i *Info) SetFlags(
 	rosenpassEnabled, rosenpassPermissive bool,
 	serverSSHAllowed *bool,
 	disableClientRoutes, disableServerRoutes,
-	disableDNS, disableFirewall, blockLANAccess, blockInbound, disableIPv6 bool,
+	disableDNS, disableFirewall, blockLANAccess, blockInbound, disableIPv6, lazyConnectionEnabled bool,
 	enableSSHRoot, enableSSHSFTP, enableSSHLocalPortForwarding, enableSSHRemotePortForwarding *bool,
 	disableSSHAuth *bool,
 ) {
@@ -103,6 +105,8 @@ func (i *Info) SetFlags(
 	i.BlockInbound = blockInbound
 	i.DisableIPv6 = disableIPv6
 
+	i.LazyConnectionEnabled = lazyConnectionEnabled
+
 	if enableSSHRoot != nil {
 		i.EnableSSHRoot = *enableSSHRoot
 	}

client/test/json-socket-docker.sh

@@ -1,223 +0,0 @@
-#!/usr/bin/env bash
-set -eEuo pipefail
-
-usage() {
-  cat <<'EOF'
-Usage: client/test/json-socket-docker.sh [tcp|unix|both]
-
-Builds the NetBird client Docker image from the local source tree, starts
-`netbird service run` in a container with --enable-json-socket, and verifies
-that the HTTP/JSON daemon gateway responds to Status requests.
-
-Modes:
-  tcp   Validate tcp://0.0.0.0:8080 via a published localhost port (default)
-  unix  Validate unix:///sock/netbird-http.sock via a bind-mounted socket dir
-  both  Run both validations
-
-Environment:
-  CONTAINER_RUNTIME  docker or podman. Auto-detected if unset.
-  IMAGE              Image tag to build. Default: netbird-json-socket-test:local
-  TARGETARCH         Go/Docker target arch. Default: `go env GOARCH`
-  PLATFORM           Docker platform. Default: linux/$TARGETARCH
-  WAIT_TIMEOUT       Seconds to wait for the JSON socket. Default: 30
-EOF
-}
-
-if [[ "${1:-}" == "-h" || "${1:-}" == "--help" ]]; then
-  usage
-  exit 0
-fi
-
-MODE="${1:-tcp}"
-case "${MODE}" in
-  tcp|unix|both) ;;
-  *)
-    usage >&2
-    echo "invalid mode: ${MODE}" >&2
-    exit 2
-    ;;
-esac
-
-RUNTIME="${CONTAINER_RUNTIME:-}"
-if [[ -z "${RUNTIME}" ]]; then
-  if command -v docker >/dev/null 2>&1; then
-    RUNTIME=docker
-  elif command -v podman >/dev/null 2>&1; then
-    RUNTIME=podman
-  else
-    echo "docker or podman is required" >&2
-    exit 127
-  fi
-fi
-if ! command -v "${RUNTIME}" >/dev/null 2>&1; then
-  echo "container runtime not found: ${RUNTIME}" >&2
-  exit 127
-fi
-
-if ! command -v curl >/dev/null 2>&1; then
-  echo "curl is required" >&2
-  exit 127
-fi
-
-ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
-IMAGE="${IMAGE:-netbird-json-socket-test:local}"
-TARGETARCH="${TARGETARCH:-$(go env GOARCH)}"
-PLATFORM="${PLATFORM:-linux/${TARGETARCH}}"
-WAIT_TIMEOUT="${WAIT_TIMEOUT:-30}"
-TMP_DIR="$(mktemp -d)"
-CONTAINERS=()
-
-cleanup() {
-  local status=$?
-  for container in "${CONTAINERS[@]:-}"; do
-    "${RUNTIME}" rm -f "${container}" >/dev/null 2>&1 || true
-  done
-  rm -rf "${TMP_DIR}"
-  exit "${status}"
-}
-trap cleanup EXIT
-
-build_image() {
-  echo "==> Building Linux ${TARGETARCH} netbird binary"
-  mkdir -p "${TMP_DIR}/context/client"
-  cp "${ROOT_DIR}/client/Dockerfile" "${TMP_DIR}/context/Dockerfile"
-  cp "${ROOT_DIR}/client/netbird-entrypoint.sh" "${TMP_DIR}/context/client/netbird-entrypoint.sh"
-
-  (cd "${ROOT_DIR}" && CGO_ENABLED=0 GOOS=linux GOARCH="${TARGETARCH}" go build -o "${TMP_DIR}/context/netbird" ./client)
-
-  echo "==> Building ${IMAGE} for ${PLATFORM}"
-  "${RUNTIME}" build \
-    --platform "${PLATFORM}" \
-    --build-arg NETBIRD_BINARY=netbird \
-    -t "${IMAGE}" \
-    -f "${TMP_DIR}/context/Dockerfile" \
-    "${TMP_DIR}/context"
-}
-
-pick_port() {
-  python3 - <<'PY'
-import socket
-sock = socket.socket()
-sock.bind(("127.0.0.1", 0))
-print(sock.getsockname()[1])
-sock.close()
-PY
-}
-
-assert_status_json() {
-  local response_file="$1"
-  if command -v python3 >/dev/null 2>&1; then
-    python3 - "${response_file}" <<'PY'
-import json
-import sys
-with open(sys.argv[1], encoding="utf-8") as fh:
-    data = json.load(fh)
-if not data.get("status"):
-    raise SystemExit("missing non-empty status field")
-if "daemonVersion" not in data:
-    raise SystemExit("missing daemonVersion field")
-print(f"status={data['status']} daemonVersion={data['daemonVersion']}")
-PY
-  else
-    grep -q '"status"' "${response_file}"
-    grep -q '"daemonVersion"' "${response_file}"
-    cat "${response_file}"
-  fi
-}
-
-container_logs() {
-  local container="$1"
-  echo "---- ${container} logs ----" >&2
-  "${RUNTIME}" logs "${container}" >&2 || true
-  echo "--------------------------" >&2
-}
-
-wait_for_http_status() {
-  local container="$1"
-  local response="${TMP_DIR}/${container}.json"
-  local curl_err="${TMP_DIR}/${container}.curl.err"
-  shift
-  local deadline=$((SECONDS + WAIT_TIMEOUT))
-
-  while (( SECONDS < deadline )); do
-    if curl -fsS "$@" \
-      -X POST \
-      -H 'Content-Type: application/json' \
-      -d '{}' \
-      -o "${response}" \
-      2>"${curl_err}"; then
-      assert_status_json "${response}"
-      return 0
-    fi
-
-    if ! "${RUNTIME}" ps --format '{{.Names}}' | grep -Fxq "${container}"; then
-      echo "container exited before JSON socket became ready" >&2
-      container_logs "${container}"
-      return 1
-    fi
-    sleep 1
-  done
-
-  echo "timed out waiting for JSON socket after ${WAIT_TIMEOUT}s" >&2
-  cat "${curl_err}" >&2 || true
-  container_logs "${container}"
-  return 1
-}
-
-run_netbird_container() {
-  local container="$1"
-  local json_socket="$2"
-  shift 2
-
-  CONTAINERS+=("${container}")
-  "${RUNTIME}" run --rm -d \
-    --name "${container}" \
-    -e NB_STATE_DIR=/tmp/netbird-state \
-    --entrypoint /usr/local/bin/netbird \
-    "$@" \
-    "${IMAGE}" \
-    --log-file console \
-    --daemon-addr unix:///tmp/netbird.sock \
-    service run \
-    --enable-json-socket \
-    --json-socket "${json_socket}" >/dev/null
-}
-
-run_tcp_test() {
-  local port container
-  port="$(pick_port)"
-  container="nb-json-socket-tcp-$RANDOM-$RANDOM"
-
-  echo "==> Validating TCP JSON socket on 127.0.0.1:${port}"
-  run_netbird_container "${container}" "tcp://0.0.0.0:8080" -p "127.0.0.1:${port}:8080"
-  wait_for_http_status "${container}" "http://127.0.0.1:${port}/daemon.DaemonService/Status"
-}
-
-run_unix_test() {
-  local sock_dir sock_path container
-  sock_dir="${TMP_DIR}/sock"
-  sock_path="${sock_dir}/netbird-http.sock"
-  container="nb-json-socket-unix-$RANDOM-$RANDOM"
-  mkdir -p "${sock_dir}"
-
-  echo "==> Validating Unix JSON socket at ${sock_path}"
-  run_netbird_container "${container}" "unix:///sock/netbird-http.sock" -v "${sock_dir}:/sock"
-  wait_for_http_status "${container}" --unix-socket "${sock_path}" "http://unix/daemon.DaemonService/Status"
-}
-
-build_image
-
-case "${MODE}" in
-  tcp)
-    run_tcp_test
-    ;;
-  unix)
-    run_unix_test
-    ;;
-  both)
-    run_tcp_test
-    run_unix_test
-    ;;
-esac
-
-echo "==> Docker JSON socket validation passed (${MODE})"

client/ui/frontend/src/modules/settings/SettingsNetwork.tsx

@@ -12,6 +12,12 @@ export function SettingsNetwork() {
     return (
         <>
             <SectionGroup title={t("settings.network.section.connectivity")}>
+                <FancyToggleSwitch
+                    value={config.lazyConnectionEnabled}
+                    onChange={(v) => setField("lazyConnectionEnabled", v)}
+                    label={t("settings.network.lazy.label")}
+                    helpText={t("settings.network.lazy.help")}
+                />
                 <FancyToggleSwitch
                     value={config.networkMonitor}
                     onChange={(v) => setField("networkMonitor", v)}

client/ui/i18n/locales/de/common.json

@@ -599,6 +599,12 @@
     "settings.network.section.routingDns": {
         "message": "Routing & DNS"
     },
+    "settings.network.lazy.label": {
+        "message": "Lazy-Verbindungen"
+    },
+    "settings.network.lazy.help": {
+        "message": "Statt durchgehend aktive Verbindungen zu halten, aktiviert NetBird sie bei Bedarf anhand von Aktivität oder Signalisierung."
+    },
     "settings.network.monitor.label": {
         "message": "Bei Netzwerkwechsel neu verbinden"
     },

client/ui/i18n/locales/en/common.json

@@ -799,6 +799,14 @@
         "message": "Routing & DNS",
         "description": "Section heading for routing and DNS options. 'DNS' is an acronym — keep it."
     },
+    "settings.network.lazy.label": {
+        "message": "Lazy Connections",
+        "description": "Toggle label: Lazy Connections (on-demand connections)."
+    },
+    "settings.network.lazy.help": {
+        "message": "Instead of maintaining always-on connections, NetBird activates them on-demand based on activity or signaling.",
+        "description": "Helper text for lazy connections."
+    },
     "settings.network.monitor.label": {
         "message": "Reconnect on Network Change",
         "description": "Toggle label: reconnect automatically on network change."

client/ui/i18n/locales/es/common.json

@@ -599,6 +599,12 @@
     "settings.network.section.routingDns": {
         "message": "Enrutamiento y DNS"
     },
+    "settings.network.lazy.label": {
+        "message": "Conexiones bajo demanda"
+    },
+    "settings.network.lazy.help": {
+        "message": "En lugar de mantener conexiones permanentes, NetBird las activa bajo demanda según la actividad o la señalización."
+    },
     "settings.network.monitor.label": {
         "message": "Reconectar al cambiar de red"
     },

client/ui/i18n/locales/fr/common.json

@@ -599,6 +599,12 @@
     "settings.network.section.routingDns": {
         "message": "Routage et DNS"
     },
+    "settings.network.lazy.label": {
+        "message": "Connexions à la demande"
+    },
+    "settings.network.lazy.help": {
+        "message": "Au lieu de maintenir des connexions permanentes, NetBird les active à la demande en fonction de l’activité ou de la signalisation."
+    },
     "settings.network.monitor.label": {
         "message": "Reconnecter en cas de changement de réseau"
     },

client/ui/i18n/locales/hu/common.json

@@ -599,6 +599,12 @@
     "settings.network.section.routingDns": {
         "message": "Útválasztás és DNS"
     },
+    "settings.network.lazy.label": {
+        "message": "Igény szerinti kapcsolatok"
+    },
+    "settings.network.lazy.help": {
+        "message": "Állandó kapcsolatok fenntartása helyett a NetBird igény szerint, aktivitás vagy jelzés alapján aktiválja azokat."
+    },
     "settings.network.monitor.label": {
         "message": "Újracsatlakozás hálózatváltáskor"
     },

client/ui/i18n/locales/it/common.json

@@ -599,6 +599,12 @@
     "settings.network.section.routingDns": {
         "message": "Routing e DNS"
     },
+    "settings.network.lazy.label": {
+        "message": "Connessioni lazy"
+    },
+    "settings.network.lazy.help": {
+        "message": "Invece di mantenere connessioni sempre attive, NetBird le attiva su richiesta in base all'attività o al signaling."
+    },
     "settings.network.monitor.label": {
         "message": "Riconnetti al cambio di rete"
     },

client/ui/i18n/locales/pt/common.json

@@ -599,6 +599,12 @@
     "settings.network.section.routingDns": {
         "message": "Roteamento e DNS"
     },
+    "settings.network.lazy.label": {
+        "message": "Conexões sob demanda"
+    },
+    "settings.network.lazy.help": {
+        "message": "Em vez de manter conexões sempre ativas, o NetBird as ativa sob demanda com base na atividade ou na sinalização."
+    },
     "settings.network.monitor.label": {
         "message": "Reconectar ao mudar de rede"
     },

client/ui/i18n/locales/ru/common.json

@@ -599,6 +599,12 @@
     "settings.network.section.routingDns": {
         "message": "Маршрутизация и DNS"
     },
+    "settings.network.lazy.label": {
+        "message": "Подключения по требованию"
+    },
+    "settings.network.lazy.help": {
+        "message": "Вместо постоянно активных подключений NetBird активирует их по требованию — на основе активности или сигналинга."
+    },
     "settings.network.monitor.label": {
         "message": "Переподключаться при смене сети"
     },

client/ui/i18n/locales/zh-CN/common.json

@@ -599,6 +599,12 @@
     "settings.network.section.routingDns": {
         "message": "路由与 DNS"
     },
+    "settings.network.lazy.label": {
+        "message": "懒连接"
+    },
+    "settings.network.lazy.help": {
+        "message": "NetBird 不会维持始终在线的连接，而是根据活动或信令按需激活连接。"
+    },
     "settings.network.monitor.label": {
         "message": "网络变化时重新连接"
     },

client/ui/services/settings.go

@@ -57,6 +57,7 @@ type Config struct {
 	RosenpassEnabled              bool   `json:"rosenpassEnabled"`
 	RosenpassPermissive           bool   `json:"rosenpassPermissive"`
 	DisableNotifications          bool   `json:"disableNotifications"`
+	LazyConnectionEnabled         bool   `json:"lazyConnectionEnabled"`
 	BlockInbound                  bool   `json:"blockInbound"`
 	NetworkMonitor                bool   `json:"networkMonitor"`
 	DisableClientRoutes           bool   `json:"disableClientRoutes"`
@@ -88,6 +89,7 @@ type SetConfigParams struct {
 	RosenpassEnabled              *bool   `json:"rosenpassEnabled,omitempty"`
 	RosenpassPermissive           *bool   `json:"rosenpassPermissive,omitempty"`
 	DisableNotifications          *bool   `json:"disableNotifications,omitempty"`
+	LazyConnectionEnabled         *bool   `json:"lazyConnectionEnabled,omitempty"`
 	BlockInbound                  *bool   `json:"blockInbound,omitempty"`
 	NetworkMonitor                *bool   `json:"networkMonitor,omitempty"`
 	DisableClientRoutes           *bool   `json:"disableClientRoutes,omitempty"`
@@ -138,6 +140,7 @@ func (s *Settings) GetConfig(ctx context.Context, p ConfigParams) (Config, error
 		RosenpassEnabled:              resp.GetRosenpassEnabled(),
 		RosenpassPermissive:           resp.GetRosenpassPermissive(),
 		DisableNotifications:          resp.GetDisableNotifications(),
+		LazyConnectionEnabled:         resp.GetLazyConnectionEnabled(),
 		BlockInbound:                  resp.GetBlockInbound(),
 		NetworkMonitor:                resp.GetNetworkMonitor(),
 		DisableClientRoutes:           resp.GetDisableClientRoutes(),
@@ -173,6 +176,7 @@ func (s *Settings) SetConfig(ctx context.Context, p SetConfigParams) error {
 		RosenpassEnabled:              p.RosenpassEnabled,
 		RosenpassPermissive:           p.RosenpassPermissive,
 		DisableNotifications:          p.DisableNotifications,
+		LazyConnectionEnabled:         p.LazyConnectionEnabled,
 		BlockInbound:                  p.BlockInbound,
 		NetworkMonitor:                p.NetworkMonitor,
 		DisableClientRoutes:           p.DisableClientRoutes,

e2e/agentnetwork/vllm_test.go

@@ -1,171 +0,0 @@
-//go:build e2e
-
-package agentnetwork
-
-import (
-	"context"
-	"strings"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/netbirdio/netbird/e2e/harness"
-	"github.com/netbirdio/netbird/shared/management/http/api"
-)
-
-// TestVLLMProvider proves the proxy supports a self-hosted vLLM backend. vLLM is
-// OpenAI-compatible, so it uses the "vllm" catalog entry (KindCustom) and is
-// reached over plain HTTP — no TLS anywhere on the path:
-//
-//	client --tunnel--> netbird proxy --http--> vllm (:8000, OpenAI-compatible)
-//
-// The mock vLLM server answers /v1/chat/completions with an OpenAI-shaped
-// completion carrying a non-zero usage block. The test asserts the chat returns
-// 200 with the completion, that the request is recorded in the access log by its
-// session id, and that vLLM's usage block is metered into a consumption row —
-// which together prove request routing, response parsing, and token accounting
-// all work for a self-hosted OpenAI-compatible provider.
-//
-// It needs no external credentials (the mock ignores auth), so it always runs.
-func TestVLLMProvider(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
-	defer cancel()
-
-	vllm, err := harness.StartVLLM(ctx, srv)
-	require.NoError(t, err, "start mock vLLM server")
-	t.Cleanup(func() { _ = vllm.Terminate(context.Background()) })
-
-	grp, err := srv.API().Groups.Create(ctx, api.PostApiGroupsJSONRequestBody{Name: "e2e-vllm"})
-	require.NoError(t, err, "create group")
-	t.Cleanup(func() { _ = srv.API().Groups.Delete(context.Background(), grp.Id) })
-
-	ephemeral := false
-	sk, err := srv.API().SetupKeys.Create(ctx, api.PostApiSetupKeysJSONRequestBody{
-		Name:       "e2e-vllm-client",
-		Type:       "reusable",
-		ExpiresIn:  86400,
-		UsageLimit: 0,
-		AutoGroups: []string{grp.Id},
-		Ephemeral:  &ephemeral,
-	})
-	require.NoError(t, err, "mint setup key")
-	require.NotEmpty(t, sk.Key, "setup key plaintext")
-
-	// vLLM provider pointed at the mock over plain HTTP. The mock ignores auth,
-	// so a dummy key satisfies the "Bearer ${API_KEY}" template. The served model
-	// is enumerated so the router dispatches this model string to this provider.
-	dummyKey := "sk-vllm-e2e"
-	prov, err := srv.CreateProvider(ctx, api.AgentNetworkProviderRequest{
-		Name:             "vllm",
-		ProviderId:       "vllm",
-		UpstreamUrl:      vllm.URL,
-		ApiKey:           &dummyKey,
-		Enabled:          ptr(true),
-		BootstrapCluster: ptr(harness.AgentNetworkCluster),
-		Models: &[]api.AgentNetworkProviderModel{
-			{Id: harness.VLLMModel, InputPer1k: 0.001, OutputPer1k: 0.002},
-		},
-	})
-	require.NoError(t, err, "create vllm provider")
-	t.Cleanup(func() { _ = srv.DeleteProvider(context.Background(), prov.Id) })
-
-	// Token limit far above the handful of tokens this test drives, so it never
-	// blocks but switches on usage metering — the switch that makes consumption
-	// rows get recorded.
-	enabled := true
-	pol, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
-		Name:                   "e2e-vllm-allow",
-		Enabled:                &enabled,
-		SourceGroups:           []string{grp.Id},
-		DestinationProviderIds: []string{prov.Id},
-		Limits: &api.AgentNetworkPolicyLimits{
-			TokenLimit: api.AgentNetworkPolicyTokenLimit{
-				Enabled:       true,
-				GroupCap:      10_000_000,
-				UserCap:       10_000_000,
-				WindowSeconds: 60,
-			},
-		},
-	})
-	require.NoError(t, err, "create policy")
-	t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), pol.Id) })
-
-	settings, err := srv.GetSettings(ctx)
-	require.NoError(t, err, "read settings")
-	require.NotEmpty(t, settings.Endpoint, "endpoint must be assigned")
-
-	proxyToken, err := srv.CreateProxyTokenCLI(ctx, "e2e-vllm-proxy")
-	require.NoError(t, err, "mint proxy token")
-	px, err := harness.StartProxy(ctx, srv, proxyToken)
-	require.NoError(t, err, "start proxy")
-	t.Cleanup(func() { _ = px.Terminate(context.Background()) })
-
-	cl, err := harness.StartClient(ctx, srv, sk.Key)
-	require.NoError(t, err, "start client")
-	t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
-
-	require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
-	if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
-		t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
-	}
-	proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
-	require.NoError(t, err, "resolve endpoint to proxy IP")
-
-	before, _ := srv.ListAccessLogs(ctx)
-	sessionID := "e2e-session-vllm"
-
-	// Retry to absorb tunnel/DNS jitter on the first call.
-	var code int
-	var body string
-	deadline := time.Now().Add(90 * time.Second)
-	for time.Now().Before(deadline) {
-		c, b, cerr := cl.Chat(ctx, settings.Endpoint, proxyIP, harness.WireChat, harness.VLLMModel, "Reply with exactly: pong", sessionID)
-		if cerr == nil {
-			code, body = c, b
-			if code == 200 {
-				break
-			}
-		}
-		time.Sleep(5 * time.Second)
-	}
-	require.Equal(t, 200, code,
-		"chat through the vLLM provider must return 200; body: %s\n=== vllm logs ===\n%s\n=== proxy logs ===\n%s",
-		body, vllm.Logs(context.Background()), px.Logs(context.Background()))
-	require.True(t, strings.Contains(body, "chat.completion"),
-		"body should be an OpenAI-compatible chat completion; got: %s", body)
-
-	// The request must surface as an access-log row carrying our session id.
-	require.Eventually(t, func() bool {
-		logs, lerr := srv.ListAccessLogs(ctx)
-		return lerr == nil && logs.TotalRecords > before.TotalRecords
-	}, 30*time.Second, 2*time.Second, "an access-log row should be ingested for the vLLM provider")
-
-	require.Eventually(t, func() bool {
-		logs, lerr := srv.ListAccessLogs(ctx)
-		if lerr != nil {
-			return false
-		}
-		for _, r := range logs.Data {
-			if r.SessionId != nil && *r.SessionId == sessionID {
-				return true
-			}
-		}
-		return false
-	}, 30*time.Second, 2*time.Second, "session id %q must be recorded in an access-log row", sessionID)
-
-	// vLLM's usage block (prompt_tokens=11, completion_tokens=2) must be parsed
-	// and metered into a consumption row with positive token counts.
-	require.Eventually(t, func() bool {
-		rows, lerr := srv.ListConsumption(ctx)
-		if lerr != nil {
-			return false
-		}
-		for _, r := range rows {
-			if r.TokensInput > 0 && r.TokensOutput > 0 {
-				return true
-			}
-		}
-		return false
-	}, 60*time.Second, 3*time.Second, "vLLM usage must be metered into a consumption row")
-}

e2e/harness/vllm.go

@@ -1,113 +0,0 @@
-//go:build e2e
-
-package harness
-
-import (
-	"context"
-	"fmt"
-	"os"
-	"path/filepath"
-	"time"
-
-	"github.com/docker/docker/api/types/container"
-	"github.com/testcontainers/testcontainers-go"
-	"github.com/testcontainers/testcontainers-go/wait"
-)
-
-const (
-	vllmImage = "nginx:alpine"
-	vllmAlias = "vllm"
-	vllmPort  = "8000/tcp"
-	// VLLMModel is the served model id the mock advertises and echoes back. It
-	// matches a real small model commonly served by vLLM so the provider's
-	// enumerated model and the client's request line up.
-	VLLMModel = "Qwen/Qwen2.5-0.5B-Instruct"
-)
-
-// vllmNginxConf emulates a vLLM OpenAI-compatible server over plain HTTP (vLLM's
-// default: no TLS, port 8000). It answers /v1/models with a one-model list and
-// any chat/completions path with a canned OpenAI-shaped chat completion carrying
-// a non-zero usage block, so the proxy's OpenAI parser records real token
-// consumption. Running actual vLLM in CI is infeasible (GPU + multi-GB model
-// download), so this stands in for the wire contract the proxy depends on.
-const vllmNginxConf = `pid /tmp/nginx.pid;
-events {}
-http {
-  server {
-    listen 8000;
-    location = /v1/models {
-      default_type application/json;
-      return 200 '{"object":"list","data":[{"id":"Qwen/Qwen2.5-0.5B-Instruct","object":"model","owned_by":"vllm"}]}';
-    }
-    location / {
-      default_type application/json;
-      return 200 '{"id":"chatcmpl-e2e-vllm","object":"chat.completion","created":1700000000,"model":"Qwen/Qwen2.5-0.5B-Instruct","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}],"usage":{"prompt_tokens":11,"completion_tokens":2,"total_tokens":13}}';
-    }
-  }
-}
-`
-
-// VLLM is a mock vLLM OpenAI-compatible server on the combined server's network,
-// reachable at http://vllm:8000. A "vllm" provider points at it to exercise the
-// proxy's support for self-hosted OpenAI-compatible backends.
-type VLLM struct {
-	container testcontainers.Container
-	workDir   string
-	// URL is the upstream URL the vllm provider points at (http://<alias>:8000).
-	URL string
-}
-
-// StartVLLM runs the mock vLLM server on the shared network over plain HTTP.
-func StartVLLM(ctx context.Context, c *Combined) (*VLLM, error) {
-	workDir, err := os.MkdirTemp("/tmp", "nb-e2e-vllm-*")
-	if err != nil {
-		return nil, fmt.Errorf("create vllm work dir: %w", err)
-	}
-	// Widen so the (non-root worker) nginx container can traverse the bind mount.
-	if err := os.Chmod(workDir, 0o755); err != nil { //nolint:gosec // throwaway e2e config dir
-		return nil, fmt.Errorf("chmod vllm dir: %w", err)
-	}
-	if err := os.WriteFile(filepath.Join(workDir, "nginx.conf"), []byte(vllmNginxConf), 0o644); err != nil { //nolint:gosec // non-secret e2e config
-		return nil, fmt.Errorf("write nginx conf: %w", err)
-	}
-
-	req := testcontainers.ContainerRequest{
-		Image:          vllmImage,
-		ExposedPorts:   []string{vllmPort},
-		Networks:       []string{c.network.Name},
-		NetworkAliases: map[string][]string{c.network.Name: {vllmAlias}},
-		Cmd:            []string{"nginx", "-c", "/conf/nginx.conf", "-g", "daemon off;"},
-		HostConfigModifier: func(hc *container.HostConfig) {
-			hc.Binds = append(hc.Binds, workDir+":/conf:ro")
-		},
-		WaitingFor: wait.ForListeningPort(vllmPort).WithStartupTimeout(60 * time.Second),
-	}
-
-	ctr, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
-		ContainerRequest: req,
-		Started:          true,
-	})
-	if err != nil {
-		_ = os.RemoveAll(workDir)
-		return nil, fmt.Errorf("start vllm container: %w", err)
-	}
-
-	return &VLLM{container: ctr, workDir: workDir, URL: "http://" + vllmAlias + ":8000"}, nil
-}
-
-// Logs returns the vLLM container logs, for diagnostics on failure.
-func (v *VLLM) Logs(ctx context.Context) string {
-	return containerLogs(ctx, v.container)
-}
-
-// Terminate stops the vLLM container and cleans its work dir.
-func (v *VLLM) Terminate(ctx context.Context) error {
-	var err error
-	if v.container != nil {
-		err = v.container.Terminate(ctx)
-	}
-	if v.workDir != "" {
-		_ = os.RemoveAll(v.workDir)
-	}
-	return err
-}

flow/client/client.go

@@ -109,7 +109,7 @@ func (c *GRPCClient) Close() error {
 func (c *GRPCClient) Send(event *proto.FlowEvent) error {
 	c.mu.Lock()
 	stream := c.stream
-	defer c.mu.Unlock() // stream.Send() is not safe to call concurrently from multiple goroutines
+	c.mu.Unlock()
 
 	if stream == nil {
 		return errors.New("stream not initialized")

flow/proto/flow.pb.go

@@ -134,11 +134,9 @@ type FlowEvent struct {
 	// When the event occurred
 	Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
 	// Public key of the sending peer
-	PublicKey   []byte                 `protobuf:"bytes,3,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
-	FlowFields  *FlowFields            `protobuf:"bytes,4,opt,name=flow_fields,json=flowFields,proto3" json:"flow_fields,omitempty"`
-	IsInitiator bool                   `protobuf:"varint,5,opt,name=isInitiator,proto3" json:"isInitiator,omitempty"`
-	WindowStart *timestamppb.Timestamp `protobuf:"bytes,6,opt,name=window_start,json=windowStart,proto3" json:"window_start,omitempty"`
-	WindowEnd   *timestamppb.Timestamp `protobuf:"bytes,7,opt,name=window_end,json=windowEnd,proto3" json:"window_end,omitempty"`
+	PublicKey   []byte      `protobuf:"bytes,3,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
+	FlowFields  *FlowFields `protobuf:"bytes,4,opt,name=flow_fields,json=flowFields,proto3" json:"flow_fields,omitempty"`
+	IsInitiator bool        `protobuf:"varint,5,opt,name=isInitiator,proto3" json:"isInitiator,omitempty"`
 }
 
 func (x *FlowEvent) Reset() {
@@ -208,20 +206,6 @@ func (x *FlowEvent) GetIsInitiator() bool {
 	return false
 }
 
-func (x *FlowEvent) GetWindowStart() *timestamppb.Timestamp {
-	if x != nil {
-		return x.WindowStart
-	}
-	return nil
-}
-
-func (x *FlowEvent) GetWindowEnd() *timestamppb.Timestamp {
-	if x != nil {
-		return x.WindowEnd
-	}
-	return nil
-}
-
 type FlowEventAck struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -300,6 +284,7 @@ type FlowFields struct {
 	// Layer 4 -specific information
 	//
 	// Types that are assignable to ConnectionInfo:
+	//
 	//	*FlowFields_PortInfo
 	//	*FlowFields_IcmpInfo
 	ConnectionInfo isFlowFields_ConnectionInfo `protobuf_oneof:"connection_info"`
@@ -312,9 +297,6 @@ type FlowFields struct {
 	// Resource ID
 	SourceResourceId []byte `protobuf:"bytes,14,opt,name=source_resource_id,json=sourceResourceId,proto3" json:"source_resource_id,omitempty"`
 	DestResourceId   []byte `protobuf:"bytes,15,opt,name=dest_resource_id,json=destResourceId,proto3" json:"dest_resource_id,omitempty"`
-	NumOfStarts      uint64 `protobuf:"varint,16,opt,name=num_of_starts,json=numOfStarts,proto3" json:"num_of_starts,omitempty"`
-	NumOfEnds        uint64 `protobuf:"varint,17,opt,name=num_of_ends,json=numOfEnds,proto3" json:"num_of_ends,omitempty"`
-	NumOfDrops       uint64 `protobuf:"varint,18,opt,name=num_of_drops,json=numOfDrops,proto3" json:"num_of_drops,omitempty"`
 }
 
 func (x *FlowFields) Reset() {
@@ -461,27 +443,6 @@ func (x *FlowFields) GetDestResourceId() []byte {
 	return nil
 }
 
-func (x *FlowFields) GetNumOfStarts() uint64 {
-	if x != nil {
-		return x.NumOfStarts
-	}
-	return 0
-}
-
-func (x *FlowFields) GetNumOfEnds() uint64 {
-	if x != nil {
-		return x.NumOfEnds
-	}
-	return 0
-}
-
-func (x *FlowFields) GetNumOfDrops() uint64 {
-	if x != nil {
-		return x.NumOfDrops
-	}
-	return 0
-}
-
 type isFlowFields_ConnectionInfo interface {
 	isFlowFields_ConnectionInfo()
 }
@@ -618,7 +579,7 @@ var file_flow_proto_rawDesc = []byte{
 	0x0a, 0x0a, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x04, 0x66, 0x6c,
 	0x6f, 0x77, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
 	0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x22, 0xce, 0x02, 0x0a, 0x09, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
+	0x6f, 0x74, 0x6f, 0x22, 0xd4, 0x01, 0x0a, 0x09, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
 	0x74, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20,
 	0x01, 0x28, 0x0c, 0x52, 0x07, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x38, 0x0a, 0x09,
 	0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32,
@@ -631,59 +592,45 @@ var file_flow_proto_rawDesc = []byte{
 	0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x52, 0x0a, 0x66, 0x6c,
 	0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x12, 0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e,
 	0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x18, 0x05, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69,
-	0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x12, 0x3d, 0x0a, 0x0c, 0x77, 0x69,
-	0x6e, 0x64, 0x6f, 0x77, 0x5f, 0x73, 0x74, 0x61, 0x72, 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b,
-	0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62,
-	0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0b, 0x77, 0x69,
-	0x6e, 0x64, 0x6f, 0x77, 0x53, 0x74, 0x61, 0x72, 0x74, 0x12, 0x39, 0x0a, 0x0a, 0x77, 0x69, 0x6e,
-	0x64, 0x6f, 0x77, 0x5f, 0x65, 0x6e, 0x64, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e,
-	0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e,
-	0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x77, 0x69, 0x6e, 0x64, 0x6f,
-	0x77, 0x45, 0x6e, 0x64, 0x22, 0x4b, 0x0a, 0x0c, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
-	0x74, 0x41, 0x63, 0x6b, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12,
-	0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x18, 0x02,
-	0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69, 0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f,
-	0x72, 0x22, 0x82, 0x05, 0x0a, 0x0a, 0x46, 0x6c, 0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73,
-	0x12, 0x17, 0x0a, 0x07, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
-	0x0c, 0x52, 0x06, 0x66, 0x6c, 0x6f, 0x77, 0x49, 0x64, 0x12, 0x1e, 0x0a, 0x04, 0x74, 0x79, 0x70,
-	0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0a, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x54,
-	0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x17, 0x0a, 0x07, 0x72, 0x75, 0x6c,
-	0x65, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x72, 0x75, 0x6c, 0x65,
-	0x49, 0x64, 0x12, 0x2d, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18,
-	0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0f, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x44, 0x69, 0x72,
-	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f,
-	0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x18, 0x05, 0x20,
-	0x01, 0x28, 0x0d, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x1b, 0x0a,
-	0x09, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x70, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0c,
-	0x52, 0x08, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x70, 0x12, 0x17, 0x0a, 0x07, 0x64, 0x65,
-	0x73, 0x74, 0x5f, 0x69, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x64, 0x65, 0x73,
-	0x74, 0x49, 0x70, 0x12, 0x2d, 0x0a, 0x09, 0x70, 0x6f, 0x72, 0x74, 0x5f, 0x69, 0x6e, 0x66, 0x6f,
-	0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x50, 0x6f,
-	0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x70, 0x6f, 0x72, 0x74, 0x49, 0x6e,
-	0x66, 0x6f, 0x12, 0x2d, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18,
-	0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x49, 0x43, 0x4d,
-	0x50, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x49, 0x6e, 0x66,
-	0x6f, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x78, 0x5f, 0x70, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x18,
-	0x0a, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x72, 0x78, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73,
-	0x12, 0x1d, 0x0a, 0x0a, 0x74, 0x78, 0x5f, 0x70, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x18, 0x0b,
-	0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x74, 0x78, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x12,
-	0x19, 0x0a, 0x08, 0x72, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0c, 0x20, 0x01, 0x28,
-	0x04, 0x52, 0x07, 0x72, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x19, 0x0a, 0x08, 0x74, 0x78,
-	0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0d, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x74, 0x78,
-	0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x2c, 0x0a, 0x12, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0e, 0x20, 0x01, 0x28,
-	0x0c, 0x52, 0x10, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
-	0x65, 0x49, 0x64, 0x12, 0x28, 0x0a, 0x10, 0x64, 0x65, 0x73, 0x74, 0x5f, 0x72, 0x65, 0x73, 0x6f,
-	0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0f, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0e, 0x64,
-	0x65, 0x73, 0x74, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x64, 0x12, 0x22, 0x0a,
-	0x0d, 0x6e, 0x75, 0x6d, 0x5f, 0x6f, 0x66, 0x5f, 0x73, 0x74, 0x61, 0x72, 0x74, 0x73, 0x18, 0x10,
-	0x20, 0x01, 0x28, 0x04, 0x52, 0x0b, 0x6e, 0x75, 0x6d, 0x4f, 0x66, 0x53, 0x74, 0x61, 0x72, 0x74,
-	0x73, 0x12, 0x1e, 0x0a, 0x0b, 0x6e, 0x75, 0x6d, 0x5f, 0x6f, 0x66, 0x5f, 0x65, 0x6e, 0x64, 0x73,
-	0x18, 0x11, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x6e, 0x75, 0x6d, 0x4f, 0x66, 0x45, 0x6e, 0x64,
-	0x73, 0x12, 0x20, 0x0a, 0x0c, 0x6e, 0x75, 0x6d, 0x5f, 0x6f, 0x66, 0x5f, 0x64, 0x72, 0x6f, 0x70,
-	0x73, 0x18, 0x12, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0a, 0x6e, 0x75, 0x6d, 0x4f, 0x66, 0x44, 0x72,
-	0x6f, 0x70, 0x73, 0x42, 0x11, 0x0a, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x22, 0x4b, 0x0a, 0x0c, 0x46, 0x6c,
+	0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x41, 0x63, 0x6b, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76,
+	0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x65, 0x76,
+	0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e, 0x69, 0x74, 0x69,
+	0x61, 0x74, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69, 0x73, 0x49, 0x6e,
+	0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x22, 0x9c, 0x04, 0x0a, 0x0a, 0x46, 0x6c, 0x6f, 0x77,
+	0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x12, 0x17, 0x0a, 0x07, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x69,
+	0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x66, 0x6c, 0x6f, 0x77, 0x49, 0x64, 0x12,
+	0x1e, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0a, 0x2e,
+	0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12,
+	0x17, 0x0a, 0x07, 0x72, 0x75, 0x6c, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c,
+	0x52, 0x06, 0x72, 0x75, 0x6c, 0x65, 0x49, 0x64, 0x12, 0x2d, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65,
+	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0f, 0x2e, 0x66, 0x6c,
+	0x6f, 0x77, 0x2e, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69,
+	0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x63, 0x6f, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x63, 0x6f, 0x6c, 0x12, 0x1b, 0x0a, 0x09, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x70,
+	0x18, 0x06, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x08, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x70,
+	0x12, 0x17, 0x0a, 0x07, 0x64, 0x65, 0x73, 0x74, 0x5f, 0x69, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28,
+	0x0c, 0x52, 0x06, 0x64, 0x65, 0x73, 0x74, 0x49, 0x70, 0x12, 0x2d, 0x0a, 0x09, 0x70, 0x6f, 0x72,
+	0x74, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66,
+	0x6c, 0x6f, 0x77, 0x2e, 0x50, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08,
+	0x70, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x2d, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70,
+	0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c,
+	0x6f, 0x77, 0x2e, 0x49, 0x43, 0x4d, 0x50, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x69,
+	0x63, 0x6d, 0x70, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x78, 0x5f, 0x70, 0x61,
+	0x63, 0x6b, 0x65, 0x74, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x72, 0x78, 0x50,
+	0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x12, 0x1d, 0x0a, 0x0a, 0x74, 0x78, 0x5f, 0x70, 0x61, 0x63,
+	0x6b, 0x65, 0x74, 0x73, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x74, 0x78, 0x50, 0x61,
+	0x63, 0x6b, 0x65, 0x74, 0x73, 0x12, 0x19, 0x0a, 0x08, 0x72, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65,
+	0x73, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x72, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73,
+	0x12, 0x19, 0x0a, 0x08, 0x74, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0d, 0x20, 0x01,
+	0x28, 0x04, 0x52, 0x07, 0x74, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x2c, 0x0a, 0x12, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69,
+	0x64, 0x18, 0x0e, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x10, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52,
+	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x64, 0x12, 0x28, 0x0a, 0x10, 0x64, 0x65, 0x73,
+	0x74, 0x5f, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0f, 0x20,
+	0x01, 0x28, 0x0c, 0x52, 0x0e, 0x64, 0x65, 0x73, 0x74, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
+	0x65, 0x49, 0x64, 0x42, 0x11, 0x0a, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f,
 	0x6e, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x22, 0x48, 0x0a, 0x08, 0x50, 0x6f, 0x72, 0x74, 0x49, 0x6e,
 	0x66, 0x6f, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x70, 0x6f, 0x72,
 	0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x50,
@@ -736,19 +683,17 @@ var file_flow_proto_goTypes = []interface{}{
 var file_flow_proto_depIdxs = []int32{
 	7, // 0: flow.FlowEvent.timestamp:type_name -> google.protobuf.Timestamp
 	4, // 1: flow.FlowEvent.flow_fields:type_name -> flow.FlowFields
-	7, // 2: flow.FlowEvent.window_start:type_name -> google.protobuf.Timestamp
-	7, // 3: flow.FlowEvent.window_end:type_name -> google.protobuf.Timestamp
-	0, // 4: flow.FlowFields.type:type_name -> flow.Type
-	1, // 5: flow.FlowFields.direction:type_name -> flow.Direction
-	5, // 6: flow.FlowFields.port_info:type_name -> flow.PortInfo
-	6, // 7: flow.FlowFields.icmp_info:type_name -> flow.ICMPInfo
-	2, // 8: flow.FlowService.Events:input_type -> flow.FlowEvent
-	3, // 9: flow.FlowService.Events:output_type -> flow.FlowEventAck
-	9, // [9:10] is the sub-list for method output_type
-	8, // [8:9] is the sub-list for method input_type
-	8, // [8:8] is the sub-list for extension type_name
-	8, // [8:8] is the sub-list for extension extendee
-	0, // [0:8] is the sub-list for field type_name
+	0, // 2: flow.FlowFields.type:type_name -> flow.Type
+	1, // 3: flow.FlowFields.direction:type_name -> flow.Direction
+	5, // 4: flow.FlowFields.port_info:type_name -> flow.PortInfo
+	6, // 5: flow.FlowFields.icmp_info:type_name -> flow.ICMPInfo
+	2, // 6: flow.FlowService.Events:input_type -> flow.FlowEvent
+	3, // 7: flow.FlowService.Events:output_type -> flow.FlowEventAck
+	7, // [7:8] is the sub-list for method output_type
+	6, // [6:7] is the sub-list for method input_type
+	6, // [6:6] is the sub-list for extension type_name
+	6, // [6:6] is the sub-list for extension extendee
+	0, // [0:6] is the sub-list for field type_name
 }
 
 func init() { file_flow_proto_init() }

flow/proto/flow.proto

@@ -24,9 +24,6 @@ message FlowEvent {
   FlowFields flow_fields = 4;
 
   bool isInitiator = 5;
-
-  google.protobuf.Timestamp window_start = 6;
-  google.protobuf.Timestamp window_end = 7;
 }
 
 message FlowEventAck {
@@ -78,9 +75,6 @@ message FlowFields {
   bytes source_resource_id = 14;
   bytes dest_resource_id = 15;
 
-  uint64 num_of_starts = 16;
-  uint64 num_of_ends = 17;
-  uint64 num_of_drops = 18;
 }
 
 // Flow event types

flow/proto/flow_grpc.pb.go

@@ -1,8 +1,4 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
-// versions:
-// - protoc-gen-go-grpc v1.6.1
-// - protoc             v3.21.9
-// source: flow.proto
 
 package proto
 
@@ -15,19 +11,15 @@ import (
 
 // This is a compile-time assertion to ensure that this generated file
 // is compatible with the grpc package it is being compiled against.
-// Requires gRPC-Go v1.64.0 or later.
-const _ = grpc.SupportPackageIsVersion9
-
-const (
-	FlowService_Events_FullMethodName = "/flow.FlowService/Events"
-)
+// Requires gRPC-Go v1.32.0 or later.
+const _ = grpc.SupportPackageIsVersion7
 
 // FlowServiceClient is the client API for FlowService service.
 //
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
 type FlowServiceClient interface {
 	// Client to receiver streams of events and acknowledgements
-	Events(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[FlowEvent, FlowEventAck], error)
+	Events(ctx context.Context, opts ...grpc.CallOption) (FlowService_EventsClient, error)
 }
 
 type flowServiceClient struct {
@@ -38,40 +30,54 @@ func NewFlowServiceClient(cc grpc.ClientConnInterface) FlowServiceClient {
 	return &flowServiceClient{cc}
 }
 
-func (c *flowServiceClient) Events(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[FlowEvent, FlowEventAck], error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &FlowService_ServiceDesc.Streams[0], FlowService_Events_FullMethodName, cOpts...)
+func (c *flowServiceClient) Events(ctx context.Context, opts ...grpc.CallOption) (FlowService_EventsClient, error) {
+	stream, err := c.cc.NewStream(ctx, &FlowService_ServiceDesc.Streams[0], "/flow.FlowService/Events", opts...)
 	if err != nil {
 		return nil, err
 	}
-	x := &grpc.GenericClientStream[FlowEvent, FlowEventAck]{ClientStream: stream}
+	x := &flowServiceEventsClient{stream}
 	return x, nil
 }
 
-// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
-type FlowService_EventsClient = grpc.BidiStreamingClient[FlowEvent, FlowEventAck]
+type FlowService_EventsClient interface {
+	Send(*FlowEvent) error
+	Recv() (*FlowEventAck, error)
+	grpc.ClientStream
+}
+
+type flowServiceEventsClient struct {
+	grpc.ClientStream
+}
+
+func (x *flowServiceEventsClient) Send(m *FlowEvent) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *flowServiceEventsClient) Recv() (*FlowEventAck, error) {
+	m := new(FlowEventAck)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
 
 // FlowServiceServer is the server API for FlowService service.
 // All implementations must embed UnimplementedFlowServiceServer
-// for forward compatibility.
+// for forward compatibility
 type FlowServiceServer interface {
 	// Client to receiver streams of events and acknowledgements
-	Events(grpc.BidiStreamingServer[FlowEvent, FlowEventAck]) error
+	Events(FlowService_EventsServer) error
 	mustEmbedUnimplementedFlowServiceServer()
 }
 
-// UnimplementedFlowServiceServer must be embedded to have
-// forward compatible implementations.
-//
-// NOTE: this should be embedded by value instead of pointer to avoid a nil
-// pointer dereference when methods are called.
-type UnimplementedFlowServiceServer struct{}
+// UnimplementedFlowServiceServer must be embedded to have forward compatible implementations.
+type UnimplementedFlowServiceServer struct {
+}
 
-func (UnimplementedFlowServiceServer) Events(grpc.BidiStreamingServer[FlowEvent, FlowEventAck]) error {
-	return status.Error(codes.Unimplemented, "method Events not implemented")
+func (UnimplementedFlowServiceServer) Events(FlowService_EventsServer) error {
+	return status.Errorf(codes.Unimplemented, "method Events not implemented")
 }
 func (UnimplementedFlowServiceServer) mustEmbedUnimplementedFlowServiceServer() {}
-func (UnimplementedFlowServiceServer) testEmbeddedByValue()                     {}
 
 // UnsafeFlowServiceServer may be embedded to opt out of forward compatibility for this service.
 // Use of this interface is not recommended, as added methods to FlowServiceServer will
@@ -81,22 +87,34 @@ type UnsafeFlowServiceServer interface {
 }
 
 func RegisterFlowServiceServer(s grpc.ServiceRegistrar, srv FlowServiceServer) {
-	// If the following call panics, it indicates UnimplementedFlowServiceServer was
-	// embedded by pointer and is nil.  This will cause panics if an
-	// unimplemented method is ever invoked, so we test this at initialization
-	// time to prevent it from happening at runtime later due to I/O.
-	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
-		t.testEmbeddedByValue()
-	}
 	s.RegisterService(&FlowService_ServiceDesc, srv)
 }
 
 func _FlowService_Events_Handler(srv interface{}, stream grpc.ServerStream) error {
-	return srv.(FlowServiceServer).Events(&grpc.GenericServerStream[FlowEvent, FlowEventAck]{ServerStream: stream})
+	return srv.(FlowServiceServer).Events(&flowServiceEventsServer{stream})
 }
 
-// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
-type FlowService_EventsServer = grpc.BidiStreamingServer[FlowEvent, FlowEventAck]
+type FlowService_EventsServer interface {
+	Send(*FlowEventAck) error
+	Recv() (*FlowEvent, error)
+	grpc.ServerStream
+}
+
+type flowServiceEventsServer struct {
+	grpc.ServerStream
+}
+
+func (x *flowServiceEventsServer) Send(m *FlowEventAck) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *flowServiceEventsServer) Recv() (*FlowEvent, error) {
+	m := new(FlowEvent)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
 
 // FlowService_ServiceDesc is the grpc.ServiceDesc for FlowService service.
 // It's only intended for direct use with grpc.RegisterService,

go.mod

@@ -66,7 +66,6 @@ require (
 	github.com/google/nftables v0.3.0
 	github.com/gopacket/gopacket v1.4.0
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.2-0.20240212192251-757544f21357
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/go-secure-stdlib/base62 v0.1.2
 	github.com/hashicorp/go-version v1.7.0
@@ -113,7 +112,7 @@ require (
 	github.com/ti-mo/conntrack v0.5.1
 	github.com/ti-mo/netfilter v0.5.2
 	github.com/vmihailenco/msgpack/v5 v5.4.1
-	github.com/wailsapp/wails/v3 v3.0.0-alpha2.111
+	github.com/wailsapp/wails/v3 v3.0.0-alpha2.106
 	github.com/yusufpapurcu/wmi v1.2.4
 	github.com/zcalusic/sysinfo v1.1.3
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0
@@ -319,7 +318,6 @@ require (
 	golang.org/x/text v0.37.0 // indirect
 	golang.org/x/tools v0.44.0 // indirect
 	golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20260319201613-d00831a3d3e7 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 // indirect
 	gopkg.in/square/go-jose.v2 v2.6.0 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect

go.sum

@@ -675,8 +675,8 @@ github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IU
 github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
 github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
 github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
-github.com/wailsapp/wails/v3 v3.0.0-alpha2.111 h1:MKx1nOnhnDuEGrRBmtxLOJq1NERwailu2cI4BvzWhi4=
-github.com/wailsapp/wails/v3 v3.0.0-alpha2.111/go.mod h1:wrdvmyeCsB/K3YqJDoH8E3MwcN8NXAMnEFaDTW46w60=
+github.com/wailsapp/wails/v3 v3.0.0-alpha2.106 h1:vay8+y89EbUTGVBjKw1r4eCd2gi7lgvO5QTlSQJe41g=
+github.com/wailsapp/wails/v3 v3.0.0-alpha2.106/go.mod h1:wrdvmyeCsB/K3YqJDoH8E3MwcN8NXAMnEFaDTW46w60=
 github.com/wailsapp/wails/webview2 v1.0.27 h1:wjgAi/I8BBZ7kUGU8um3XF3ILEfzr96Q2Q1G4GPjMns=
 github.com/wailsapp/wails/webview2 v1.0.27/go.mod h1:zdM4jcO1IaC61RiJL5F1BzgoqBHFIdacz8gPr5exr0o=
 github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=

infrastructure_files/getting-started.sh

@@ -351,6 +351,11 @@ initialize_default_values() {
   NETBIRD_STUN_PORT=3478
 
   # Docker images
+  # Record whether the operator explicitly pinned the server/proxy images via
+  # env vars, so the agent-network preset can pick its own defaults without
+  # clobbering an explicit override.
+  NETBIRD_SERVER_IMAGE_EXPLICIT=${NETBIRD_SERVER_IMAGE:+true}
+  NETBIRD_PROXY_IMAGE_EXPLICIT=${NETBIRD_PROXY_IMAGE:+true}
   DASHBOARD_IMAGE=${DASHBOARD_IMAGE:-"netbirdio/dashboard:latest"}
   # Combined server replaces separate signal, relay, and management containers
   NETBIRD_SERVER_IMAGE=${NETBIRD_SERVER_IMAGE:-"netbirdio/netbird-server:latest"}
@@ -410,6 +415,15 @@ apply_agent_network_preset() {
   ENABLE_PROXY="true"
   ENABLE_CROWDSEC="false"
 
+  # Agent-network ships dedicated server/proxy images. Honor an explicit
+  # env override; otherwise pin the agent-network builds.
+  if [[ "${NETBIRD_SERVER_IMAGE_EXPLICIT}" != "true" ]]; then
+    NETBIRD_SERVER_IMAGE="netbirdio/netbird-server:0.74.0-rc.2"
+  fi
+  if [[ "${NETBIRD_PROXY_IMAGE_EXPLICIT}" != "true" ]]; then
+    NETBIRD_PROXY_IMAGE="netbirdio/reverse-proxy:0.74.0-rc.2"
+  fi
+
   if [[ -n "${NETBIRD_LETSENCRYPT_EMAIL}" ]]; then
     TRAEFIK_ACME_EMAIL="${NETBIRD_LETSENCRYPT_EMAIL}"
   else

management/internals/modules/agentnetwork/catalog/catalog.go

@@ -627,21 +627,6 @@ var providers = []Provider{
 		},
 		Models: []Model{},
 	},
-	{
-		// vLLM is an OpenAI-compatible self-hosted server. It behaves like
-		// the generic custom entry; it gets its own catalog id purely so it
-		// surfaces as a named "vLLM" choice in the provider picker.
-		ID:                 "vllm",
-		Kind:               KindCustom,
-		Name:               "vLLM",
-		Description:        "Self-hosted vLLM (OpenAI-compatible)",
-		DefaultHost:        "",
-		AuthHeaderName:     "Authorization",
-		AuthHeaderTemplate: "Bearer ${API_KEY}",
-		DefaultContentType: "application/json",
-		BrandColor:         "#30A2FF",
-		Models:             []Model{},
-	},
 	{
 		ID:                 "custom",
 		Kind:               KindCustom,

management/internals/shared/grpc/conversion.go

@@ -47,13 +47,16 @@ func init() {
 	precomputedDeprecatedRemotePeersConstraint = constraint
 }
 
-// toNetbirdConfig converts the server configuration to the wire representation. It returns
-// nil when no server config is set (the fan-out network-map path) because clients treat any
-// non-nil config as authoritative: a config without a relay section is interpreted as relay
-// disabled and wipes the clients' relay URLs.
 func toNetbirdConfig(config *nbconfig.Config, turnCredentials *Token, relayToken *Token, extraSettings *types.ExtraSettings, settings *types.Settings) *proto.NetbirdConfig {
 	if config == nil {
-		return nil
+		if settings == nil {
+			return nil
+		}
+		return &proto.NetbirdConfig{
+			Metrics: &proto.MetricsConfig{
+				Enabled: settings.MetricsPushEnabled,
+			},
+		}
 	}
 
 	var stuns []*proto.HostConfig

management/internals/shared/grpc/conversion_test.go

@@ -8,13 +8,11 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 
 	nbdns "github.com/netbirdio/netbird/dns"
 	"github.com/netbirdio/netbird/management/internals/controllers/network_map"
 	"github.com/netbirdio/netbird/management/internals/controllers/network_map/controller/cache"
 	nbconfig "github.com/netbirdio/netbird/management/internals/server/config"
-	"github.com/netbirdio/netbird/management/server/types"
 )
 
 func TestToProtocolDNSConfigWithCache(t *testing.T) {
@@ -265,39 +263,3 @@ func TestEncodeSessionExpiresAt(t *testing.T) {
 		assert.True(t, got.AsTime().Equal(deadline))
 	})
 }
-
-// TestToNetbirdConfig_RelayInvariant guards against the v0.74.0 relay-wipe regression.
-// Clients treat any non-nil NetbirdConfig as authoritative and interpret a missing relay
-// section as relay disabled, wiping their relay URLs. toNetbirdConfig must therefore
-// return nil when no server config is set (the fan-out network-map path) instead of a
-// partial config, and a result built from a relay-enabled config must carry the relay
-// section.
-func TestToNetbirdConfig_RelayInvariant(t *testing.T) {
-	settings := &types.Settings{MetricsPushEnabled: true}
-
-	t.Run("nil server config returns nil config", func(t *testing.T) {
-		nbCfg := toNetbirdConfig(nil, nil, nil, nil, settings)
-		assert.Nil(t, nbCfg, "fan-out updates must not carry a partial NetbirdConfig even when settings are present")
-	})
-
-	t.Run("relay-enabled config carries relay section", func(t *testing.T) {
-		cfg := &nbconfig.Config{
-			Stuns: []*nbconfig.Host{{Proto: nbconfig.UDP, URI: "stun:stun.example.com:3478"}},
-			TURNConfig: &nbconfig.TURNConfig{
-				Turns: []*nbconfig.Host{{Proto: nbconfig.UDP, URI: "turn:turn.example.com:3478", Username: "user", Password: "pass"}},
-			},
-			Relay:  &nbconfig.Relay{Addresses: []string{"rels://relay.example.com:443"}},
-			Signal: &nbconfig.Host{Proto: nbconfig.HTTP, URI: "signal.example.com:10000"},
-		}
-		relayToken := &Token{Payload: "token-payload", Signature: "token-signature"}
-
-		nbCfg := toNetbirdConfig(cfg, nil, relayToken, nil, settings)
-		require.NotNil(t, nbCfg)
-		require.NotNil(t, nbCfg.Relay, "non-nil NetbirdConfig must include the relay section")
-		assert.Equal(t, cfg.Relay.Addresses, nbCfg.Relay.Urls, "relay URLs should match the server config")
-		assert.Equal(t, relayToken.Payload, nbCfg.Relay.TokenPayload, "relay token payload should be set")
-		assert.Equal(t, relayToken.Signature, nbCfg.Relay.TokenSignature, "relay token signature should be set")
-		require.NotNil(t, nbCfg.Metrics)
-		assert.True(t, nbCfg.Metrics.Enabled, "metrics flag should carry the settings value")
-	})
-}

management/server/peer_test.go

@@ -1048,7 +1048,11 @@ func testUpdateAccountPeers(t *testing.T) {
 
 			for _, channel := range peerChannels {
 				update := <-channel
-				assert.Nil(t, update.Update.NetbirdConfig, "fan-out updates must not carry a NetbirdConfig; clients treat a config without relay as relay disabled and wipe their relay URLs")
+				assert.NotNil(t, update.Update.NetbirdConfig)
+				assert.Nil(t, update.Update.NetbirdConfig.Stuns)
+				assert.Nil(t, update.Update.NetbirdConfig.Turns)
+				assert.Nil(t, update.Update.NetbirdConfig.Signal)
+				assert.Nil(t, update.Update.NetbirdConfig.Relay)
 				assert.Equal(t, tc.peers, len(update.Update.NetworkMap.RemotePeers))
 				assert.Equal(t, tc.peers*2, len(update.Update.NetworkMap.FirewallRules))
 			}

shared/management/client/grpc.go

@@ -33,15 +33,10 @@ const ConnectTimeout = 10 * time.Second
 const healthCheckTimeout = 5 * time.Second
 
 const (
-	// EnvMaxRecvMsgSize overrides the default gRPC max receive message size
+	// EnvMaxRecvMsgSize overrides the default gRPC max receive message size (4 MB)
 	// for the management client connection. Value is in bytes.
 	EnvMaxRecvMsgSize = "NB_MANAGEMENT_GRPC_MAX_MSG_SIZE"
 
-	// defaultMaxRecvMsgSize is the max gRPC receive message size used for the
-	// management client connection when EnvMaxRecvMsgSize is unset or invalid.
-	// It overrides the gRPC library default of 4 MB.
-	defaultMaxRecvMsgSize = 1024 * 1024 * 16
-
 	errMsgMgmtPublicKey    = "failed getting Management Service public key: %s"
 	errMsgNoMgmtConnection = "no connection to management"
 )
@@ -89,22 +84,22 @@ type ExposeResponse struct {
 }
 
 // MaxRecvMsgSize returns the configured max gRPC receive message size from
-// the environment, or defaultMaxRecvMsgSize (16 MB) if unset or invalid.
+// the environment, or 0 if unset (which uses the gRPC default of 4 MB).
 func MaxRecvMsgSize() int {
 	val := os.Getenv(EnvMaxRecvMsgSize)
 	if val == "" {
-		return defaultMaxRecvMsgSize
+		return 0
 	}
 
 	size, err := strconv.Atoi(val)
 	if err != nil {
 		log.Warnf("invalid %s value %q, using default: %v", EnvMaxRecvMsgSize, val, err)
-		return defaultMaxRecvMsgSize
+		return 0
 	}
 
 	if size <= 0 {
 		log.Warnf("invalid %s value %d, must be positive, using default", EnvMaxRecvMsgSize, size)
-		return defaultMaxRecvMsgSize
+		return 0
 	}
 
 	return size
@@ -1023,6 +1018,8 @@ func infoToMetaData(info *system.Info) *proto.PeerSystemMeta {
 			BlockLANAccess:      info.BlockLANAccess,
 			BlockInbound:        info.BlockInbound,
 			DisableIPv6:         info.DisableIPv6,
+
+			LazyConnectionEnabled: info.LazyConnectionEnabled,
 		},
 
 		Capabilities: peerCapabilities(*info),

shared/management/client/grpc_test.go

@@ -21,11 +21,11 @@ func TestMaxRecvMsgSize(t *testing.T) {
 		envValue string
 		expected int
 	}{
-		{name: "unset returns default", envValue: "", expected: defaultMaxRecvMsgSize},
+		{name: "unset returns 0", envValue: "", expected: 0},
 		{name: "valid value", envValue: "10485760", expected: 10485760},
-		{name: "non-numeric returns default", envValue: "abc", expected: defaultMaxRecvMsgSize},
-		{name: "negative returns default", envValue: "-1", expected: defaultMaxRecvMsgSize},
-		{name: "zero returns default", envValue: "0", expected: defaultMaxRecvMsgSize},
+		{name: "non-numeric returns 0", envValue: "abc", expected: 0},
+		{name: "negative returns 0", envValue: "-1", expected: 0},
+		{name: "zero returns 0", envValue: "0", expected: 0},
 	}
 
 	for _, tt := range tests {

shared/management/http/api/openapi.yml

@@ -2769,28 +2769,6 @@ components:
           type: integer
           description: "Number of packets transmitted."
           example: 5
-        num_of_starts:
-          type: integer
-          description: "Number of start events."
-          example: 3
-        num_of_ends:
-          type: integer
-          description: "Number of end events."
-          example: 4
-        num_of_drops:
-          type: integer
-          description: "Number of drop events."
-          example: 5
-        window_start:
-          type: string
-          format: date-time
-          description: Timestamp of the start of the aggregation window.
-          example: 2025-03-20T16:23:58.125397Z
-        window_end:
-          type: string
-          format: date-time
-          description: Timestamp of the end of the aggregation window.
-          example: 2025-03-20T16:23:58.125397Z
         events:
           type: array
           description: "List of events that are correlated to this flow (e.g., start, end)."
@@ -2812,11 +2790,6 @@ components:
         - rx_packets
         - tx_bytes
         - tx_packets
-        - num_of_starts
-        - num_of_ends
-        - num_of_drops
-        - window_start
-        - window_end
         - events
     NetworkTrafficEventsResponse:
       type: object

shared/management/http/api/types.gen.go

@@ -3709,18 +3709,9 @@ type NetworkTrafficEvent struct {
 	Events []NetworkTrafficSubEvent `json:"events"`
 
 	// FlowId FlowID is the ID of the connection flow. Not unique because it can be the same for multiple events (e.g., start and end of the connection).
-	FlowId string             `json:"flow_id"`
-	Icmp   NetworkTrafficICMP `json:"icmp"`
-
-	// NumOfDrops Number of drop events.
-	NumOfDrops int `json:"num_of_drops"`
-
-	// NumOfEnds Number of end events.
-	NumOfEnds int `json:"num_of_ends"`
-
-	// NumOfStarts Number of start events.
-	NumOfStarts int                  `json:"num_of_starts"`
-	Policy      NetworkTrafficPolicy `json:"policy"`
+	FlowId string               `json:"flow_id"`
+	Icmp   NetworkTrafficICMP   `json:"icmp"`
+	Policy NetworkTrafficPolicy `json:"policy"`
 
 	// Protocol Protocol is the protocol of the traffic (e.g. 1 = ICMP, 6 = TCP, 17 = UDP, etc.).
 	Protocol int `json:"protocol"`
@@ -3741,12 +3732,6 @@ type NetworkTrafficEvent struct {
 	// TxPackets Number of packets transmitted.
 	TxPackets int                `json:"tx_packets"`
 	User      NetworkTrafficUser `json:"user"`
-
-	// WindowEnd Timestamp of the end of the aggregation window.
-	WindowEnd time.Time `json:"window_end"`
-
-	// WindowStart Timestamp of the start of the aggregation window.
-	WindowStart time.Time `json:"window_start"`
 }
 
 // NetworkTrafficEventsResponse defines model for NetworkTrafficEventsResponse.